KAFKA-8147: Add changelog topic configuration to KTable suppress

[highluck]

Committer Checklist (excluded from commit message)

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

[highluck]

@vvcephei @bbejeck
Can I ask for your opinion? thanks !

[bbejeck]

Ok to test

[highluck]

@bbejeck
retest this please

[bbejeck]

Both Java 11 and 8 failed. Test results already cleaned up.

Retest this please.

[highluck]

@bbejeck @vvcephei
I'm sorry, but can you test it again?

[vvcephei]

Hi @highluck , sorry for the slow review cycle. I was waiting for the opportunity to really give you a good review.

I have three high level comments:

1. The javadocs are not quite right, see the suggestions below.
2. If we're going to start using BufferConfigInternal as the internally-facing interface for BufferConfig, I'd like to keep it more like an interface by maintaining all the state in the implementation classes, just for future maintainability. I wasn't 100% sure how clean this would be, so I tried it out myself. Here's my suggestion: SUGGESTION: KAFKA-8147: Add changelog topic configuration to KTable suppress highluck/kafka#1
3. We need at least one more test to demonstrate that the desired log configuration (or disablement) would actually be used. This should probably take the form of an integration test, in which we start a Streams application and then inspect the resulting topic to verify it got configured properly. Otherwise, we'd have no regression test that the KTableImpl logging code actually takes effect and continues to work after future modifications to the code base.

[highluck]

@vvcephei
thank you for your suggestion and review
It was better than my code!

i'm code update and add test code
Please confirm!
thank you

[vvcephei]

Hi @highluck , thanks for the update!

Thanks for adding the integration test. It doesn't look like we are actually verifying the changelog topic. I think you can query the EmbeddedKafkaCluster to get the actual log configuration. I'd consider the following two tests sufficient:

1. Add a non-default configuration, such as setting the retention time to some magic number, then query CLUSTER to see if that configuration did indeed take effect
2. Disable logging and then query CLUSTER to verify that the topic is in fact absent.

Additionally, we should verify (either in a new or just in an existing test) that the changelog is present by default.

Thanks!
-John

[highluck]

@vvcephei
While writing the test code, I realized that I was wrong
Your review was great!

I added two cases of test code
Please comment again!
Thank you!

[vvcephei]

Thanks for the update, @highluck !

I suggested a few tweaks.

Also, we'd need one more test like shouldAllowDisablingChangelog() that does withLoggingDisabled(), and then verifies assertThat(CLUSTER.getAllTopicsInCluster(), not(contains(changeLog)));. Although that verification might be brittle. Maybe it would be safer to just verify that there is no changelog in the cluster that contains the word "KTABLE-SUPPRESS".

[vvcephei]

Ok to test

[highluck]

@vvcephei
i'm test update and add
Please comment again
thankyou!

[highluck]

Retest this please.

[vvcephei]

Look good! Thanks, @highluck

[vvcephei]

Ok to test.

[vvcephei]

Retest this, please.

[vvcephei]

Hey @highluck , it looks like the tests can't build because of:

00:05:02.608 > Task :streams:test-utils:jar
00:05:05.208 
00:05:05.208 > Task :streams:compileTestJava
00:05:05.208 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13@2/streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java:349: warning: [deprecation] <A>setAsJavaSet(scala.collection.Set<A>) in JavaConverters has been deprecated
00:05:05.208         return JavaConverters.setAsJavaSet(brokers[0].kafkaServer().zkClient().getAllTopicsInCluster());
00:05:05.208                              ^
00:05:05.208   where A is a type-variable:
00:05:05.209     A extends Object declared in method <A>setAsJavaSet(scala.collection.Set<A>)
00:05:05.209 error: warnings found and -Werror specified
00:05:17.007 1 error
00:05:17.007 1 warning

[vvcephei]

You should be able to reproduce this with :

./gradlew clean :sterams:test :streams:test-utils:test

[highluck]

@vvcephei
oh! thankyou
i'm fixed!

Retest this, please.

[vvcephei]

test this please

[vvcephei]

ok to test

[highluck]

@vvcephei
retest this please

[vvcephei]

It looks like the Java 11 build was terminated (maybe timed out, or maybe terminated due to some ops on the host). But, since the java 8 build passed, and the java 11 build was so close to passing, I'm going to run the Streams tests locally and then merge.

[vvcephei]

Local build passed:

$ ./gradlew clean systemTestLibs :streams:test :streams:test-utils:test
...
BUILD SUCCESSFUL in 8m 51s

[vvcephei]

Merged to trunk, updated Jira and KIP to reflect the completed status. Thanks for the contribution, @highluck !

[highluck]

@vvcephei
thank you!!

[mjsax]

@highluck @vvcephei I just going over merged KIP PR -- seems we did not update any docs with this PR...

@highluck Would you like to do a follow up PR to update the docs?

[highluck]

@mjsax
thank you
I want to proceed!
Could you explain if it's okay?

[mjsax]

Cool. Just open a new PR using the same Jira number in the title and ping for review.

Let us know if you have any questions.

[highluck]

@mjsax
Could you please provide feedback on how to update the docs?

[mjsax]

There is a directory docs -- with sub-directory streams (cf https://github.com/apache/kafka/tree/trunk/docs).

As this PR implements a KIP, you should update docs/streams/upgrade-guide.html (there is a section about "Public API changes in 2.6"). Not sure atm, but maybe developer-guide/dsl-api.html would also need an update.

\cc @vvcephei

[highluck]

@mjsax thank you for comment!!