Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

BufferedStreamConsumerTest: remove non-determinism in size of generated test records #9274

Merged
merged 2 commits into from
Jan 4, 2022
Merged

BufferedStreamConsumerTest: remove non-determinism in size of generated test records #9274

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a22dead
bugfix: generate objects fixed size
grubberr Jan 4, 2022
3871c96
generated records 40 bytes of size
grubberr Jan 4, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .../io/airbyte/integrations/destination/buffered_stream_consumer/BufferedStreamConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,7 @@ protected void acceptTracked(final AirbyteMessage message) throws Exception {
// are serialized again when writing to
// the destination
long messageSizeInBytes = ByteUtils.getSizeInBytesForUTF8CharSet(Jsons.serialize(recordMessage.getData()));
if (bufferSizeInBytes + messageSizeInBytes >= maxQueueSizeInBytes) {
if (bufferSizeInBytes + messageSizeInBytes > maxQueueSizeInBytes) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why is this changed too? should this also be =?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As for me it's real bug in code itself which I discovered fixing tests.
We need to FLUSH buffer only if
bufferSizeInBytes + messageSizeInBytes > maxQueueSizeInBytes

because if

bufferSizeInBytes + messageSizeInBytes = maxQueueSizeInBytes

we still has to keep buffer not flushed

it's not my logic
it's what I see according to tests

LOGGER.info("Flushing buffer...");
flushQueueToDestination();
bufferSizeInBytes = 0;
Expand Down
2 changes: 1 addition & 1 deletion ...airbyte/integrations/destination/buffered_stream_consumer/BufferedStreamConsumerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,7 +315,7 @@ private static List<AirbyteMessage> generateRecords(final long targetSizeInBytes
List<AirbyteMessage> output = Lists.newArrayList();
long bytesCounter = 0;
for (int i = 0;; i++) {
JsonNode payload = Jsons.jsonNode(ImmutableMap.of("id", RandomStringUtils.randomAlphabetic(7), "name", "human " + String.format("%5d", i)));
JsonNode payload = Jsons.jsonNode(ImmutableMap.of("id", RandomStringUtils.randomAlphabetic(7), "name", "human " + String.format("%8d", i)));
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

how does this make the generated size always be 40?

Copy link
Contributor Author

@grubberr grubberr Jan 4, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

now it generates object of fixed size:

{"id":"dumplEw","name":"human        0"}
{"id":"tnJatrn","name":"human       25"}

I have added more spaces before number (to make it 40 bytes)

I have also changed randomAscii -> randomAlphabetic
to avoid generation DOULD-QUOTE and BACK-SLACH which can generated 2 bytes after json-encoding

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, thanks for the explanation!

long sizeInBytes = ByteUtils.getSizeInBytesForUTF8CharSet(Jsons.serialize(payload));
bytesCounter += sizeInBytes;
AirbyteMessage airbyteMessage = new AirbyteMessage()
Expand Down