KAFKA-15942: Implement ConsumerInterceptor

[Joker-5]

When invoking ConsumerInterceptor#onCommit method in ConsumerCoordinator, there're some redundant if-nonNull branches.

private final ConsumerInterceptors<?, ?> interceptors is a non-null field because it'll be instantiate in the constructor, and this field is non-null when delivering from LegacyKafkaConsumer.

Some proofs:
There're 2 constructors in LegacyKafkaConsumer, both of them will instantiate interceptors field:

• constructor1 in LegacyKafkaConsumer：this.interceptors = new ConsumerInterceptors<>(interceptorList);
• constructor2 in LegacyKafkaConsumer: this.interceptors = new ConsumerInterceptors<>(Collections.emptyList());

And this field can not set to null because it's private and no method to modify it.

Committer Checklist (excluded from commit message)

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

[vamossagar12]

Thanks for the PR @Joker-5. I looked at the ticket KAFKA-15492, the ask seems to be to implement ConsumerInterceptor in AsyncKafkaConsumer. I don't see those changes in this PR. Also, I see a test failure like testOnJoinPrepareWithOffsetCommitShouldKeepJoinAfterRebalanceTimeout() – org.apache.kafka.clients.consumer.internals.EagerConsumerCoordinatorTest having this stacktrace:

java.lang.NullPointerException: Cannot invoke "org.apache.kafka.clients.consumer.internals.ConsumerInterceptors.onCommit(java.util.Map)" because the return value of "org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.access$1000(org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)" is null
Stacktrace
java.lang.NullPointerException: Cannot invoke "org.apache.kafka.clients.consumer.internals.ConsumerInterceptors.onCommit(java.util.Map)" because the return value of "org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.access$1000(org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)" is null
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$2.onSuccess(ConsumerCoordinator.java:1062)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$2.onSuccess(ConsumerCoordinator.java:1057)
	at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:169)
	at org.apache.kafka.clients.consumer.internals.RequestFuture.addListener(RequestFuture.java:192)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.doCommitOffsetsAsync(ConsumerCoordinator.java:1057)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsAsync(ConsumerCoordinator.java:1008)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.autoCommitOffsetsAsync(ConsumerCoordinator.java:1185)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.maybeAutoCommitOffsetsAsync(ConsumerCoordinator.java:1202)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinPrepare(ConsumerCoordinator.java:720)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:472)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinatorTest.prepareCoordinatorForCloseTest(ConsumerCoordinatorTest.java:3761)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinatorTest.testOnJoinPrepareWithOffsetCommitShouldKeepJoinAfterRebalanceTimeout(ConsumerCoordinatorTest.java:1381)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
	at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:728)
	at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
	at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:156)
	at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:147)
	at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:86)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(InterceptingExecutableInvoker.java:103)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.lambda$invoke$0(InterceptingExecutableInvoker.java:93)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.invoke(InterceptingExecutableInvoker.java:92)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.invoke(InterceptingExecutableInvoker.java:86)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$7(TestMethodTestDescriptor.java:218)

Do you think, that is related to the changes in this PR?

[Joker-5]

@vamossagar12 Thanks for reviewing the PR!

At the beginning I misunderstood this ticket, now I understood and the code has already updated.
When commit request returns successfully from the broker in AsyncKafkaConsumer on both sync and async methods, I added the ConsumerInterceptor#onCommit.

Would you have time to take a look at this PR?

[vamossagar12]

hmm, the JDK21 build failed with this error =>

> Task :examples:spotbugsMain

Cannot contact jenkins-shared-ubuntu-3: java.lang.InterruptedException

> Task :core:compileScala

> Task :group-coordinator:classes

> Task :metadata:classes

> Task :examples:spotbugsTest SKIPPED

> Task :examples:check

> Task :group-coordinator:checkstyleMain

> Task :clients:testClasses

> Task :clients:checkstyleTest

> Task :clients:spotbugsMain



The message received from the daemon indicates that the daemon has disappeared.

Build request sent: Build{id=f24da9a8-313c-42d5-bc18-9f578f44c4a3, currentDir=/home/jenkins/jenkins-agent/712657a4/workspace/Kafka_kafka-pr_PR-14963}

Attempting to read last messages from the daemon log...

Daemon pid: 2004981

  log file: /home/jenkins/.gradle/daemon/8.5/daemon-2004981.out.log

----- Last  20 lines from daemon log file - daemon-2004981.out.log -----

2023-12-11T15:25:02.911+0000 [DEBUG] [org.gradle.launcher.daemon.server.SynchronizedDispatchConnection] thread 25: received class org.gradle.launcher.daemon.protocol.CloseInput

2023-12-11T15:25:02.911+0000 [DEBUG] [org.gradle.launcher.daemon.server.DefaultDaemonConnection] thread 25: Received IO message from client: org.gradle.launcher.daemon.protocol.CloseInput@41f596f2

2023-12-11T15:25:02.945+0000 [DEBUG] [org.gradle.launcher.daemon.server.exec.RequestStopIfSingleUsedDaemon] Requesting daemon stop after processing Build{id=f24da9a8-313c-42d5-bc18-9f578f44c4a3, currentDir=/home/jenkins/jenkins-agent/712657a4/workspace/Kafka_kafka-pr_PR-14963}

2023-12-11T15:25:02.965+0000 [LIFECYCLE] [org.gradle.launcher.daemon.server.DaemonStateCoordinator] Daemon will be stopped at the end of the build 

2023-12-11T15:25:02.968+0000 [DEBUG] [org.gradle.launcher.daemon.server.DaemonStateCoordinator] Stop as soon as idle requested. The daemon is busy: true

2023-12-11T15:25:02.969+0000 [DEBUG] [org.gradle.launcher.daemon.server.DaemonStateCoordinator] daemon stop has been requested. Sleeping until state changes.

2023-12-11T15:25:02.970+0000 [DEBUG] [org.gradle.launcher.daemon.server.exec.ExecuteBuild] The daemon has started executing the build.

2023-12-11T15:25:02.971+0000 [DEBUG] [org.gradle.launcher.daemon.server.exec.ExecuteBuild] Executing build with daemon context: DefaultDaemonContext[uid=c8e3ee8e-33ef-46b7-a5a1-3d91a0e54ca4,javaHome=/usr/local/asfpackages/java/adoptium-jdk-21.0.1+12,daemonRegistryDir=/home/jenkins/.gradle/daemon,pid=2004981,idleTimeout=120000,priority=NORMAL,applyInstrumentationAgent=true,daemonOpts=-Xss4m,-XX:+UseParallelGC,--add-opens=java.base/java.util=ALL-UNNAMED,--add-opens=java.base/java.lang=ALL-UNNAMED,--add-opens=java.base/java.lang.invoke=ALL-UNNAMED,--add-opens=java.prefs/java.util.prefs=ALL-UNNAMED,--add-opens=java.base/java.nio.charset=ALL-UNNAMED,--add-opens=java.base/java.net=ALL-UNNAMED,--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED,-Xmx2g,-Dfile.encoding=UTF-8,-Duser.country,-Duser.language=en,-Duser.variant]

2023-12-11T15:25:02.972+0000 [INFO] [org.gradle.launcher.daemon.server.exec.ForwardClientInput] Closing daemon's stdin at end of input.

2023-12-11T15:25:02.973+0000 [INFO] [org.gradle.launcher.daemon.server.exec.ForwardClientInput] The daemon will no longer process any standard input.

Starting build with version 3.7.0-SNAPSHOT (commit id 22d2b462) using Gradle 8.5, Java 21 and Scala 2.13.12

Build properties: maxParallelForks=4, maxScalacThreads=4, maxTestRetries=0

[vamossagar12]

Thanks for the changes @Joker-5 . I will take a look this week.

[vamossagar12]

@Joker-5 , thanks for the update. There are some missing pieces in this PR from what I can tell. Having said that, I see another PR created for this here. You might want to check with @lucasbru about this.

[lucasbru]

Hey @Joker-5, I took the ticket since your original PR seemed to only change the legacy consumer, so I thought it was just linked to the wrong ticket.

I think there are some things missing here

• enable unit / integration tests
• the way you implemented it, I think the interceptors will run as part of the background thread, but I think they should not interfere with the background thread and run as part of the application thread instead.

How about we merge my PR which has the two changes and I add you in a Co-authored-by tag? Sorry again for the confusion.

[Joker-5]

Hey @Joker-5, I took the ticket since your original PR seemed to only change the legacy consumer, so I thought it was just linked to the wrong ticket.

I think there are some things missing here

• enable unit / integration tests
• the way you implemented it, I think the interceptors will run as part of the background thread, but I think they should not interfere with the background thread and run as part of the application thread instead.

How about we merge my PR which has the two changes and I add you in a Co-authored-by tag? Sorry again for the confusion.

Hi @lucasbru, I understand. This is the second PR which i committed to Kafka so it seems a bit confusing. Now I learned a lot from your PR, so just do it.

There're some information for Co-authored-by tag:

• my username: john_doe
• my email: zh2725284321@gmail.com

[Joker-5]

@Joker-5 , thanks for the update. There are some missing pieces in this PR from what I can tell. Having said that, I see another PR created for this here. You might want to check with @lucasbru about this.

Thanks so much for the review!
I had checked with @lucasbru. We will merge @lucasbru's PR which has the two changes and @lucasbru will add me in a Co-authored-by tag.

[vamossagar12]

@Joker-5 , makes sense and thanks for letting me know. We can close this PR then?

[Joker-5]

@Joker-5 , makes sense and thanks for letting me know. We can close this PR then?

Sure, I'll close this PR soon.

[Joker-5]

Hey, @lucasbru, I saw this PR is ready to merge, but I don't see the Co-authord-by tag in the commits log.
I'm a bit worried that you forgot or maybe there's some reason for it. I would be extremely grateful if you would tell me about it if you have time.
So sorry to trouble you.

[lucasbru]

@Joker-5 I can add it in the PR description so that I don't forget. But the tag needs to added to the final commit, once we merge it, so we cannot really add it yet

[lucasbru]

There is one here: 1a54c25

[Joker-5]

There is one here: 1a54c25

Got it, feel so sorry to trouble you again.
Have a nice day!