You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2024/02/07 09:59:53 UTC
[PR] KAFKA-16155: Re-enable testAutoCommitIntercept [kafka]
lucasbru opened a new pull request, #15334:
URL: https://github.com/apache/kafka/pull/15334
The main bug causing this test to fail as described in the ticket was already fixed.
The test is still flaky if unchanged, because in the new consumer, the assignment can change in between two polls. Interceptors are only executed inside poll (and have to be, since they must run as part of the application thread), so we need to modify the integration test to call poll once after observing that the assignment changed.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16155: Re-enable testAutoCommitIntercept [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15334:
URL: https://github.com/apache/kafka/pull/15334#discussion_r1487471412
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -1378,6 +1377,14 @@ class PlaintextConsumerTest extends BaseConsumerTest {
// after rebalancing, we should have reset to the committed positions
assertEquals(10, testConsumer.committed(Set(tp).asJava).get(tp).offset)
assertEquals(20, testConsumer.committed(Set(tp2).asJava).get(tp2).offset)
+
+ // In both CLASSIC and CONSUMER protocols, interceptors are executed in poll and close.
+ // However, in the CONSUMER protocol, the assignment may be changed outside of a poll, so
+ // we need to poll once to ensure the interceptor is called.
+ if (groupProtocol.toUpperCase == GroupProtocol.CONSUMER.name) {
Review Comment:
Do we actually need to differentiate between the protocols here? Would it also be OK to call `poll()` irrespectively from the protocol?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16155: Re-enable testAutoCommitIntercept [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15334:
URL: https://github.com/apache/kafka/pull/15334#discussion_r1489269554
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -1378,6 +1377,14 @@ class PlaintextConsumerTest extends BaseConsumerTest {
// after rebalancing, we should have reset to the committed positions
assertEquals(10, testConsumer.committed(Set(tp).asJava).get(tp).offset)
assertEquals(20, testConsumer.committed(Set(tp2).asJava).get(tp2).offset)
+
+ // In both CLASSIC and CONSUMER protocols, interceptors are executed in poll and close.
+ // However, in the CONSUMER protocol, the assignment may be changed outside of a poll, so
+ // we need to poll once to ensure the interceptor is called.
+ if (groupProtocol.toUpperCase == GroupProtocol.CONSUMER.name) {
Review Comment:
We can call it irrespective of the protocol. This is just to make the minor difference in behavior explicit. But we can simplify if you prefer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16155: Re-enable testAutoCommitIntercept [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15334:
URL: https://github.com/apache/kafka/pull/15334#discussion_r1481228908
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -1378,6 +1377,14 @@ class PlaintextConsumerTest extends BaseConsumerTest {
// after rebalancing, we should have reset to the committed positions
assertEquals(10, testConsumer.committed(Set(tp).asJava).get(tp).offset)
assertEquals(20, testConsumer.committed(Set(tp2).asJava).get(tp2).offset)
+
+ // In both CLASSIC and CONSUMER protocols, interceptors are executed in poll and close.
+ // However, in the CONSUMER protocol, the assignment may be changed outside of a poll, so
+ // we need to poll once to ensure the interceptor is called.
+ if (groupProtocol.toUpperCase == GroupProtocol.CONSUMER.name) {
+ testConsumer.poll(Duration.ZERO);
Review Comment:
The alternatives to this are
- Write a KIP to allow ConsumerInterceptor to be called from other threads. This would require all implementors of interceptors to make their implementations thread-safe.
- Execute the interceptors inside `AsyncKafkaConsumer.assignment` specifically to work around the problem of observing the assignment without running the intereceptors.
I think both options are worse than keeping this slight difference in behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16155: Re-enable testAutoCommitIntercept [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #15334:
URL: https://github.com/apache/kafka/pull/15334#issuecomment-1938263362
@cadonna Could you please have a look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16155: Re-enable testAutoCommitIntercept [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #15334:
URL: https://github.com/apache/kafka/pull/15334
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16155: Re-enable testAutoCommitIntercept [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15334:
URL: https://github.com/apache/kafka/pull/15334#discussion_r1489442547
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -1378,6 +1377,14 @@ class PlaintextConsumerTest extends BaseConsumerTest {
// after rebalancing, we should have reset to the committed positions
assertEquals(10, testConsumer.committed(Set(tp).asJava).get(tp).offset)
assertEquals(20, testConsumer.committed(Set(tp2).asJava).get(tp2).offset)
+
+ // In both CLASSIC and CONSUMER protocols, interceptors are executed in poll and close.
+ // However, in the CONSUMER protocol, the assignment may be changed outside of a poll, so
+ // we need to poll once to ensure the interceptor is called.
+ if (groupProtocol.toUpperCase == GroupProtocol.CONSUMER.name) {
Review Comment:
I fine either way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org