Skip to content

KAFKA-14834: [2/N] Test coverage for out-of-order data in joins #13497

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

Merged
mjsax merged 3 commits into from
Apr 12, 2023
Merged

KAFKA-14834: [2/N] Test coverage for out-of-order data in joins #13497

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f286210
add out-of-order data to join tests
vcrfxia Mar 31, 2023
120d513
checkstyle suppression
vcrfxia Apr 4, 2023
2e5dbe5
remove redunant suppression
vcrfxia Apr 11, 2023
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 checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -224,7 +224,7 @@
files="(RecordCollectorTest|StreamsPartitionAssignorTest|StreamThreadTest|StreamTaskTest|TaskManagerTest|TopologyTestDriverTest).java"/>

<suppress checks="MethodLength"
files="(EosIntegrationTest|EosV2UpgradeIntegrationTest|KStreamKStreamJoinTest|RocksDBWindowStoreTest).java"/>
files="(EosIntegrationTest|EosV2UpgradeIntegrationTest|KStreamKStreamJoinTest|RocksDBWindowStoreTest|StreamStreamJoinIntegrationTest).java"/>

<suppress checks="ClassDataAbstractionCoupling"
files=".*[/\\]streams[/\\].*test[/\\].*.java"/>
Expand Down
66 changes: 35 additions & 31 deletions streams/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,31 +88,35 @@ public static Collection<Object[]> data() {
StreamsBuilder builder;

private final List<Input<String>> input = Arrays.asList(
new Input<>(INPUT_TOPIC_LEFT, null),
new Input<>(INPUT_TOPIC_RIGHT, null),
new Input<>(INPUT_TOPIC_LEFT, "A"),
new Input<>(INPUT_TOPIC_RIGHT, "a"),
new Input<>(INPUT_TOPIC_LEFT, "B"),
new Input<>(INPUT_TOPIC_RIGHT, "b"),
new Input<>(INPUT_TOPIC_LEFT, null),
new Input<>(INPUT_TOPIC_RIGHT, null),
new Input<>(INPUT_TOPIC_LEFT, "C"),
new Input<>(INPUT_TOPIC_RIGHT, "c"),
new Input<>(INPUT_TOPIC_RIGHT, null),
new Input<>(INPUT_TOPIC_LEFT, null),
new Input<>(INPUT_TOPIC_RIGHT, null),
new Input<>(INPUT_TOPIC_RIGHT, "d"),
new Input<>(INPUT_TOPIC_LEFT, "D")
new Input<>(INPUT_TOPIC_LEFT, null, 1),
new Input<>(INPUT_TOPIC_RIGHT, null, 2),
new Input<>(INPUT_TOPIC_LEFT, "A", 3),
new Input<>(INPUT_TOPIC_RIGHT, "a", 4),
new Input<>(INPUT_TOPIC_LEFT, "B", 5),
new Input<>(INPUT_TOPIC_RIGHT, "b", 6),
new Input<>(INPUT_TOPIC_LEFT, null, 7),
new Input<>(INPUT_TOPIC_RIGHT, null, 8),
new Input<>(INPUT_TOPIC_LEFT, "C", 9),
new Input<>(INPUT_TOPIC_RIGHT, "c", 10),
new Input<>(INPUT_TOPIC_RIGHT, null, 11),
new Input<>(INPUT_TOPIC_LEFT, null, 12),
new Input<>(INPUT_TOPIC_RIGHT, null, 13),
new Input<>(INPUT_TOPIC_RIGHT, "d", 14),
new Input<>(INPUT_TOPIC_LEFT, "D", 15),
new Input<>(INPUT_TOPIC_LEFT, "E", 4), // out-of-order data

@mjsax mjsax Apr 11, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Seems we only need this for table-table join tests, but it only blows up stream-stream join results? I am ok to merge as is, but seems it's adding "some noise" the stream-stream-join case (not too bad for binary joins, but for 3-way join it's a lot).

Not sure if we could have different inputs for stream vs table case?

@vcrfxia vcrfxia Apr 11, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Yeah, I noticed this as well. If you don't think having the out-of-order data in stream-stream join tests is valuable, I can open a follow-up PR to remove this (to reduce noise). Just LMK.

@mjsax mjsax Apr 12, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Don't think it does really harm, but it makes it very hard to actually read the test.

@vcrfxia vcrfxia Apr 18, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Fixed in #13609.

new Input<>(INPUT_TOPIC_RIGHT, "e", 3),
new Input<>(INPUT_TOPIC_RIGHT, "f", 7),
new Input<>(INPUT_TOPIC_LEFT, "F", 8)
);

private final List<Input<String>> leftInput = Arrays.asList(
new Input<>(INPUT_TOPIC_LEFT, null),
new Input<>(INPUT_TOPIC_LEFT, "A"),
new Input<>(INPUT_TOPIC_LEFT, "B"),
new Input<>(INPUT_TOPIC_LEFT, null),
new Input<>(INPUT_TOPIC_LEFT, "C"),
new Input<>(INPUT_TOPIC_LEFT, null),
new Input<>(INPUT_TOPIC_LEFT, "D")
new Input<>(INPUT_TOPIC_LEFT, null, 1),
new Input<>(INPUT_TOPIC_LEFT, "A", 2),
new Input<>(INPUT_TOPIC_LEFT, "B", 3),
new Input<>(INPUT_TOPIC_LEFT, null, 4),
new Input<>(INPUT_TOPIC_LEFT, "C", 5),
new Input<>(INPUT_TOPIC_LEFT, null, 6),
new Input<>(INPUT_TOPIC_LEFT, "D", 7)
);


Expand Down Expand Up @@ -156,17 +160,16 @@ void runTestWithDriver(final List<List<TestRecord<Long, String>>> expectedResult

TestRecord<Long, String> expectedFinalResult = null;

final long firstTimestamp = time.milliseconds();
long eventTimestamp = firstTimestamp;
final long baseTimestamp = time.milliseconds();
final Iterator<List<TestRecord<Long, String>>> resultIterator = expectedResult.iterator();
for (final Input<String> singleInputRecord : input) {
testInputTopicMap.get(singleInputRecord.topic).pipeInput(singleInputRecord.record.key, singleInputRecord.record.value, ++eventTimestamp);
testInputTopicMap.get(singleInputRecord.topic).pipeInput(singleInputRecord.record.key, singleInputRecord.record.value, baseTimestamp + singleInputRecord.timestamp);

final List<TestRecord<Long, String>> expected = resultIterator.next();
if (expected != null) {
final List<TestRecord<Long, String>> updatedExpected = new LinkedList<>();
for (final TestRecord<Long, String> record : expected) {
updatedExpected.add(new TestRecord<>(record.key(), record.value(), null, firstTimestamp + record.timestamp()));
updatedExpected.add(new TestRecord<>(record.key(), record.value(), null, baseTimestamp + record.timestamp()));
}

final List<TestRecord<Long, String>> output = outputTopic.readRecordsToList();
Expand All @@ -191,19 +194,18 @@ void runTestWithDriver(final TestRecord<Long, String> expectedFinalResult, final
testInputTopicMap.put(INPUT_TOPIC_RIGHT, right);
testInputTopicMap.put(INPUT_TOPIC_LEFT, left);

final long firstTimestamp = time.milliseconds();
long eventTimestamp = firstTimestamp;
final long baseTimestamp = time.milliseconds();

for (final Input<String> singleInputRecord : input) {
testInputTopicMap.get(singleInputRecord.topic).pipeInput(singleInputRecord.record.key, singleInputRecord.record.value, ++eventTimestamp);
testInputTopicMap.get(singleInputRecord.topic).pipeInput(singleInputRecord.record.key, singleInputRecord.record.value, baseTimestamp + singleInputRecord.timestamp);
}

final TestRecord<Long, String> updatedExpectedFinalResult =
new TestRecord<>(
expectedFinalResult.key(),
expectedFinalResult.value(),
null,
firstTimestamp + expectedFinalResult.timestamp());
baseTimestamp + expectedFinalResult.timestamp());

final List<TestRecord<Long, String>> output = outputTopic.readRecordsToList();

Expand Down Expand Up @@ -259,10 +261,12 @@ private void checkQueryableStore(final String queryableName, final TestRecord<Lo
private static final class Input<V> {
String topic;
KeyValue<Long, V> record;
long timestamp;

Input(final String topic, final V value) {
Input(final String topic, final V value, final long timestamp) {
this.topic = topic;
record = KeyValue.pair(ANY_UNIQUE_KEY, value);
this.timestamp = timestamp;
}
}
}
Loading