KAFKA-10049: Fixed FKJ bug where wrapped serdes are set incorrectly when using default StreamsConfig serdes by bellemare · Pull Request #8764 · apache/kafka

bellemare · 2020-05-31T14:48:35Z

Bug Details:
Mistakenly setting the value serde to the key serde for an internal wrapped serde in the FKJ workflow.

Testing:
Added integration test to use a non-primitive Serde, in this case the JSONSerde that the original bug finder reported using. Expanded integration test to ensure that the default Serdes work for the entire happy path of the FKJ.

Introduces a testing dependency on com.fasterxml.jackson, though this is already the case in other modules so I suspect it won't be a big issue.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

vvcephei

Thanks, @bellemare , just a high-level question about the testing strategy.

guozhangwang

@bellemare could you point to me which line is the actual fix?

guozhangwang · 2020-06-01T20:07:25Z

test this please

bellemare · 2020-06-01T20:58:40Z

@guozhangwang This is 1/2 of the areas that needed the fix. The valueSerde was being passed into the underlying Serde, despite it really needing the keySerde.

bellemare · 2020-06-06T15:50:45Z

Updated KTableKTableForeignKeyJoinScenarioTest from (String,String) to (Integer,String). This would have caught this particular bug. Reverted the JSON-related test since it's not essential for our coverage.

vvcephei · 2020-06-08T15:02:32Z

Test this please

vvcephei · 2020-06-08T15:02:49Z

Thanks, @bellemare ! I'll give it a look today.

vvcephei · 2020-06-08T15:21:36Z

Looks like checkstyle failed. I may submit a patch, just to get the tests running.

vvcephei

Thanks for the update, @bellemare ! Just a couple of minor comments on the tests, but it looks great overall.

vvcephei · 2020-06-08T17:03:13Z

Might be nice for demonstration purposes if the two records actually have different keys. Maybe:

Suggested change

aTopic.pipeInput(1, "1-alpha");

bTopic.pipeInput(1, "beta");

aTopic.pipeInput(1, "999-alpha");

bTopic.pipeInput(999, "beta");

Agreed. Good catch.

bump (it looks like this didn't get changed)

I swear I know how to read... T_T

vvcephei · 2020-06-08T17:10:56Z

Ok, since I had a couple of small requests, I won't push a fix for the checkstyle errors. In case they're gone by the time you come back to this, it was:

00:11:48.980 > Task :streams:checkstyleTest
00:11:48.981 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:71: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:48.981 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:72: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:48.981 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:78: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:48.982 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:94: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:48.982 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:95: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:49.132 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:101: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:49.132 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:117: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:49.132 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:118: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:49.133 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:126: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:49.133 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:142: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:49.133 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:143: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:49.133 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:149: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:49.133 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:166: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:49.133 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:167: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:49.133 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:173: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:49.134 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:206: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]
00:11:49.134 [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java:207: 'lambda arguments' has incorrect indentation level 16, expected level should be 12. [Indentation]

By the way, if you run ./gradlew :streams:test, it includes the checkstyle rules. It takes a while to run the tests, but it's faster than waiting for Jenkins. Often, if I don't want to wait before pushing to the branch, I'll run it on my machine after pushing, just so I'll catch stuff like this.

…hen using default StreamsConfig serdes

bellemare · 2020-06-08T22:47:05Z

I swear it ran checkstyle when I compiled it... bah! I guess not. I did run ./gradlew :streams:test it had one unrelated error (I think) so lets see how it does now.

vvcephei · 2020-06-09T15:29:39Z

Test this please

vvcephei · 2020-06-09T15:29:49Z

Test this please

vvcephei · 2020-06-09T15:30:00Z

retest this please

… join mechanisms

bellemare · 2020-06-11T15:05:08Z

retest this please

vvcephei · 2020-06-11T20:05:08Z

Test this please

vvcephei · 2020-06-11T20:05:18Z

Test this please

vvcephei · 2020-06-11T20:06:09Z

Test this please

vvcephei · 2020-06-11T20:06:21Z

Test this please

vvcephei · 2020-06-11T20:07:14Z

Hey @bellemare , Thanks so much for the update! Jenkins has been a bit lazy recently, so I'll probably be pinging it for a while...

vvcephei · 2020-06-11T20:08:30Z

Ah, there we go :)

vvcephei · 2020-06-11T20:08:45Z

I've noticed that if you complain, you tend to get results...

vvcephei · 2020-06-12T14:58:51Z

All the test failures were kafka.admin.ReassignPartitionsUnitTest.testModifyBrokerThrottles

bellemare · 2020-06-12T15:25:53Z

Thanks John!

vvcephei · 2020-06-12T16:09:31Z

Np, Thanks for the fix, @bellemare !

…hen using default StreamsConfig serdes (#8764) Bug Details: Mistakenly setting the value serde to the key serde for an internal wrapped serde in the FKJ workflow. Testing: Modified the existing test to reproduce the issue, then verified that the test passes. Reviewers: Guozhang Wang <wangguoz@gmail.com>, John Roesler <vvcephei@apache.org>

vvcephei · 2020-06-12T16:55:27Z

Cherry-picked to 2.6 and 2.5

* 'trunk' of github.com:apache/kafka: (42 commits) HOTFIX: Fix compile error in TopicAdminTest (apache#8866) KAFKA-10144: clean up corrupted standby tasks before attempting a commit (apache#8849) KAFKA-10157: Fix broken tests due to InterruptedException from FinalizedFeatureChangeListener (apache#8857) KAFKA-9432: automated protocol for DescribeConfigs (apache#8312) KAFKA-10049: Fixed FKJ bug where wrapped serdes are set incorrectly when using default StreamsConfig serdes (apache#8764) KAFKA-10027: Implement read path for feature versioning system (KIP-584) (apache#8680) KAFKA-10085: correctly compute lag for optimized source changelogs (apache#8787) KAFKA-10086: Integration test for ensuring warmups are effective (apache#8818) KAFKA-9374: Make connector interactions asynchronous (apache#8069) MINOR: reduce sizeInBytes for percentiles metrics (apache#8835) KAFKA-10115: Incorporate errors.tolerance with the Errant Record Reporter (apache#8829) KAFKA-9216: Enforce that Connect’s internal topics use `compact` cleanup policy (apache#8828) KAFKA-9845: Warn users about using config providers with plugin.path property (apache#8455) KAFKA-7833: Add missing test (apache#8847) KAFKA-9066: Retain metrics for failed tasks (apache#8502) KAFKA-9841: Revoke duplicate connectors and tasks when zombie workers return with an outdated assignment (apache#8453) KAFKA-9985: Sink connector may exhaust broker when writing in DLQ (apache#8663) KAFKA-9441: remove prepareClose() to simplify task management (apache#8833) KAFKA-7833: Add Global/StateStore name conflict check (apache#8825) KAFKA-9969: Exclude ConnectorClientConfigRequest from class loading isolation (apache#8630) ...

…t-for-generated-requests * apache-github/trunk: (248 commits) KAFKA-10049: Fixed FKJ bug where wrapped serdes are set incorrectly when using default StreamsConfig serdes (apache#8764) KAFKA-10027: Implement read path for feature versioning system (KIP-584) (apache#8680) KAFKA-10085: correctly compute lag for optimized source changelogs (apache#8787) KAFKA-10086: Integration test for ensuring warmups are effective (apache#8818) KAFKA-9374: Make connector interactions asynchronous (apache#8069) MINOR: reduce sizeInBytes for percentiles metrics (apache#8835) KAFKA-10115: Incorporate errors.tolerance with the Errant Record Reporter (apache#8829) KAFKA-9216: Enforce that Connect’s internal topics use `compact` cleanup policy (apache#8828) KAFKA-9845: Warn users about using config providers with plugin.path property (apache#8455) KAFKA-7833: Add missing test (apache#8847) KAFKA-9066: Retain metrics for failed tasks (apache#8502) KAFKA-9841: Revoke duplicate connectors and tasks when zombie workers return with an outdated assignment (apache#8453) KAFKA-9985: Sink connector may exhaust broker when writing in DLQ (apache#8663) KAFKA-9441: remove prepareClose() to simplify task management (apache#8833) KAFKA-7833: Add Global/StateStore name conflict check (apache#8825) KAFKA-9969: Exclude ConnectorClientConfigRequest from class loading isolation (apache#8630) KAFKA-9991: Fix flaky unit tests (apache#8843) KAFKA-10014; Always try to close all channels in Selector#close (apache#8685) KAFKA-10079: improve thread-level stickiness (apache#8775) MINOR: Print all removed dynamic members during join complete (apache#8816) ...

vvcephei added the streams label Jun 1, 2020

vvcephei self-requested a review June 1, 2020 14:52

vvcephei requested changes Jun 1, 2020

View reviewed changes

Comment thread ...ache/kafka/streams/integration/KTableKTableForeignKeyJoinMaterializationIntegrationTest.java Outdated

Comment thread ...ache/kafka/streams/integration/KTableKTableForeignKeyJoinMaterializationIntegrationTest.java Outdated

guozhangwang reviewed Jun 1, 2020

View reviewed changes

bellemare commented Jun 1, 2020

View reviewed changes

vvcephei reviewed Jun 1, 2020

View reviewed changes

Comment thread ...java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapperSerde.java Outdated

bellemare force-pushed the 10049 branch from b6ffb6d to 46ed050 Compare June 6, 2020 15:48

vvcephei reviewed Jun 8, 2020

View reviewed changes

KAFKA-10049: Fixed FKJ bug where wrapped serdes are set incorrectly w…

29a5395

…hen using default StreamsConfig serdes

bellemare force-pushed the 10049 branch from 46ed050 to 29a5395 Compare June 8, 2020 22:44

KAFKA-10049: Changed primary key from 1 to 999 to further clarify the…

e47f194

… join mechanisms

vvcephei approved these changes Jun 11, 2020

View reviewed changes

vvcephei merged commit bcf45b0 into apache:trunk Jun 12, 2020

bellemare deleted the 10049 branch June 12, 2020 15:26

-            aTopic.pipeInput(1, "1-alpha");
-            bTopic.pipeInput(1, "beta");
+            aTopic.pipeInput(1, "999-alpha");
+            bTopic.pipeInput(999, "beta");

Conversation

bellemare commented May 31, 2020

Committer Checklist (excluded from commit message)

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Jun 1, 2020

Uh oh!

bellemare Jun 1, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

bellemare commented Jun 6, 2020

Uh oh!

vvcephei commented Jun 8, 2020

Uh oh!

vvcephei commented Jun 8, 2020

Uh oh!

vvcephei commented Jun 8, 2020

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

vvcephei Jun 8, 2020

Choose a reason for hiding this comment

Uh oh!

bellemare Jun 8, 2020

Choose a reason for hiding this comment

Uh oh!

vvcephei Jun 9, 2020

Choose a reason for hiding this comment

Uh oh!

bellemare Jun 9, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

vvcephei commented Jun 8, 2020

Uh oh!

bellemare commented Jun 8, 2020

Uh oh!

vvcephei commented Jun 9, 2020

Uh oh!

vvcephei commented Jun 9, 2020

Uh oh!

vvcephei commented Jun 9, 2020

Uh oh!

bellemare commented Jun 11, 2020

Uh oh!

vvcephei commented Jun 11, 2020

Uh oh!

vvcephei commented Jun 11, 2020

Uh oh!

vvcephei commented Jun 11, 2020

Uh oh!

vvcephei commented Jun 11, 2020

Uh oh!

vvcephei commented Jun 11, 2020

Uh oh!

vvcephei commented Jun 11, 2020

Uh oh!

vvcephei commented Jun 11, 2020

Uh oh!

vvcephei commented Jun 12, 2020

Uh oh!

bellemare commented Jun 12, 2020

Uh oh!

vvcephei commented Jun 12, 2020

Uh oh!

vvcephei commented Jun 12, 2020

Uh oh!

Reviewers

Assignees

Labels