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

Jump to bottom

KAFKA-6474: Rewrite tests to use new public TopologyTestDriver [part 4] #5433

Merged
merged 4 commits into from
Feb 14, 2019
Merged

KAFKA-6474: Rewrite tests to use new public TopologyTestDriver [part 4] #5433

merged 4 commits into from
Feb 14, 2019

Conversation

h314to
Copy link
Contributor

@h314to h314to commented Jul 29, 2018

This PR continues the work towards the removal of KStreamTest driver. As suggested in a previous PR, OutputVerifier is used instead of MockProcessor.

Refactor:

  • KTableKTableOuterJoinTest
  • KTableKTableLeftJoinTest
  • KTableKTableOuterJoinTest

@mjsax mjsax added the streams label Aug 5, 2018
@mjsax
Copy link
Member

mjsax commented Aug 6, 2018

Call for review @vvcephei

mjsax
mjsax reviewed Aug 6, 2018
View reviewed changes
Copy link
Member

@mjsax mjsax left a comment

Choose a reason for hiding this comment

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

Thanks for the PR. Some minor comments. Overall LGTM.

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoinTest.java Outdated
assertOutputKeyValue(driver, 0, "XX0+YY0");
assertOutputKeyValue(driver, 1, "XX1+YY1");
assertOutputKeyValue(driver, 2, "XX2+YY2");
assertOutputKeyValue(driver, 3, "XX3+YY3");
Copy link
Member

Choose a reason for hiding this comment

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

add assertNull(driver.readOutput(output)); ?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Yes, good idea. Done

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoinTest.java Outdated
assertOutputKeyValue(driver, 0, "X0+YY0");
assertOutputKeyValue(driver, 1, "X1+YY1");
assertOutputKeyValue(driver, 2, "X2+YY2");
assertOutputKeyValue(driver, 3, "X3+YY3");
Copy link
Member

Choose a reason for hiding this comment

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

add assertNull(driver.readOutput(output)); ?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I agree. Done

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoinTest.java Outdated
driver.pipeInput(recordFactory.create(topic2, expectedKeys[i], null));
}
assertOutputKeyValue(driver, 0, null);
assertOutputKeyValue(driver, 1, null);
Copy link
Member

Choose a reason for hiding this comment

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

as above?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Done

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoinTest.java Outdated
assertEquals(kv.value, value);
}
}
private void assertOutputKeyValue(TopologyTestDriver driver, Integer expectedKey, String expectedValue) {
Copy link
Member

Choose a reason for hiding this comment

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

nit: add final to parameters / reformat one parameter per line

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Yes, I forgot final. Thanks

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java
assertOutputKeyValue(driver, 0, "X0+Y0");
assertOutputKeyValue(driver, 1, "X1+Y1");
assertOutputKeyValue(driver, 2, "X2+null");
assertOutputKeyValue(driver, 3, "X3+null");
Copy link
Member

Choose a reason for hiding this comment

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

add assertNull(driver.readOutput(output));

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Done

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
assertOutputKeyValue(driver, 0, "X0+YY0");
assertOutputKeyValue(driver, 1, "X1+YY1");
assertOutputKeyValue(driver, 2, "X2+YY2");
assertOutputKeyValue(driver, 3, "X3+YY3");
Copy link
Member

Choose a reason for hiding this comment

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

add assertNull(driver.readOutput(output));

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Done

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
driver.pipeInput(recordFactory.create(topic2, expectedKeys[i], null));
}
assertOutputKeyValue(driver, 0, "X0+null");
assertOutputKeyValue(driver, 1, "X1+null");
Copy link
Member

Choose a reason for hiding this comment

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

add assertNull(driver.readOutput(output));

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Done

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
assertOutputKeyValue(driver, 0, "XX0+null");
assertOutputKeyValue(driver, 1, "XX1+null");
assertOutputKeyValue(driver, 2, "XX2+YY2");
assertOutputKeyValue(driver, 3, "XX3+YY3");
Copy link
Member

Choose a reason for hiding this comment

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

add assertNull(driver.readOutput(output));

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Done

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
driver.pipeInput(recordFactory.create(topic1, expectedKeys[i], null));
}
assertOutputKeyValue(driver, 1, null);
assertOutputKeyValue(driver, 2, "null+YY2");
Copy link
Member

Choose a reason for hiding this comment

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

add assertNull(driver.readOutput(output));

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Done

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java Outdated
@@ -376,19 +326,8 @@ public void shouldLogAndMeterSkippedRecordsDueToNullLeftKey() {
assertThat(appender.getMessages(), hasItem("Skipping record due to null key. change=[(new<-old)] topic=[left] partition=[-1] offset=[-2]"));
}

private KeyValue<Integer, String> kv(final Integer key, final String value) {
return new KeyValue<>(key, value);
private void assertOutputKeyValue(TopologyTestDriver driver, Integer expectedKey, String expectedValue) {
Copy link
Member

Choose a reason for hiding this comment

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

as above

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Done

@h314to h314to force-pushed the fix/KAFKA-6474-part4-KTableKTableJoins branch from ead1c7a to b43c51f Compare January 23, 2019 00:24
@h314to
Copy link
Contributor Author

h314to commented Jan 23, 2019

retest this please

@h314to
Copy link
Contributor Author

h314to commented Jan 24, 2019

There are a couple of failed tests with JDK11 and Scala 2.12, namely:

kafka.cluster.PartitionTest.testDelayedFetchAfterAppendRecords
org.apache.kafka.streams.KafkaStreamsTest.statefulTopologyShouldCreateStateDirectory

but they seem unrelated to the changes in this PR.

@vvcephei
Copy link
Contributor

vvcephei commented Feb 5, 2019

Hey @h314to ,

I'll try to take a look tomorrow.

Most likely, the failures are unrelated. Thank you for taking note of which ones failed. You can re-run them by saying:

Retest this, please.

vvcephei
vvcephei approved these changes Feb 6, 2019
View reviewed changes
Copy link
Contributor

@vvcephei vvcephei left a comment

Choose a reason for hiding this comment

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

Looks good to me.

Thanks, @h314to !

@mjsax
Copy link
Member

mjsax commented Feb 13, 2019

Retest this please.

1 similar comment
@mjsax
Copy link
Member

mjsax commented Feb 13, 2019

Retest this please.

@mjsax mjsax merged commit c1918a2 into apache:trunk Feb 14, 2019
@mjsax
Copy link
Member

mjsax commented Feb 14, 2019

Thanks for the PR @h314to! Merged to trunk.

@h314to h314to deleted the fix/KAFKA-6474-part4-KTableKTableJoins branch February 14, 2019 23:17
@h314to
Copy link
Contributor Author

h314to commented Feb 14, 2019

Great! Thanks. I'm working on a final PR to close this issue, but I'm running into some trouble. If I don't figure it out soon, and you're ok with that, I'll push it marked as work in progress to ask for some help.

@mjsax
Copy link
Member

mjsax commented Feb 14, 2019

Sure. Sounds good!

@mjsax
Copy link
Member

mjsax commented Feb 15, 2019

@h314to We got a bug report today: https://issues.apache.org/jira/browse/KAFKA-7933 -- I am wondering if this could be related to this PR? Any thoughts on this?

jarekr pushed a commit to confluentinc/kafka that referenced this pull request Apr 18, 2019
@lbradstreet
Merge remote-tracking branch 'ak/trunk' into cos-trunk
0eacf48 
* ak/trunk: (45 commits)
  KAFKA-7487: DumpLogSegments misreports offset mismatches (apache#5756)
  MINOR: improve JavaDocs about auto-repartitioning in Streams DSL (apache#6269)
  KAFKA-7935: UNSUPPORTED_COMPRESSION_TYPE if ReplicaManager.getLogConfig returns None (apache#6274)
  KAFKA-7895: Fix stream-time reckoning for suppress (apache#6278)
  KAFKA-6569: Move OffsetIndex/TimeIndex logger to companion object  (apache#4586)
  MINOR: add log indicating the suppression time (apache#6260)
  MINOR: Make info logs for KafkaConsumer a bit more verbose (apache#6279)
  KAFKA-7758: Reuse KGroupedStream/KGroupedTable with named repartition topics (apache#6265)
  KAFKA-7884; Docs for message.format.version should display valid values (apache#6209)
  MINOR: Save failed test output to build output directory
  MINOR: add test for StreamsSmokeTestDriver (apache#6231)
  MINOR: Fix bugs identified by compiler warnings (apache#6258)
  KAFKA-6474: Rewrite tests to use new public TopologyTestDriver [part 4] (apache#5433)
  MINOR: fix bypasses in ChangeLogging stores (apache#6266)
  MINOR: Make MockClient#poll() more thread-safe (apache#5942)
  MINOR: drop dbAccessor reference on close (apache#6254)
  KAFKA-7811: Avoid unnecessary lock acquire when KafkaConsumer commits offsets (apache#6119)
  KAFKA-7916: Unify store wrapping code for clarity (apache#6255)
  MINOR: Add missing Alter Operation to Topic supported operations list in AclCommand
  KAFKA-7921: log at error level for missing source topic (apache#6262)
  ...
pengxiaolong pushed a commit to pengxiaolong/kafka that referenced this pull request Jun 14, 2019
@h314to @pengxiaolong
KAFKA-6474: Rewrite tests to use new public TopologyTestDriver [part …
6884f34 
…4] (apache#5433)

Reviewer: John Roesler <john@confluent.io>, Matthias J. Sax <matthias@confluent.io>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@mjsax mjsax mjsax left review comments

@vvcephei vvcephei vvcephei approved these changes

Assignees
No one assigned
Labels
streams
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
3 participants
@h314to @mjsax @vvcephei