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/16 14:28:49 UTC
[PR] KAFKA-16009: Fix PlaintextConsumerTest. testMaxPollIntervalMsDelayInRevocation [kafka]
lucasbru opened a new pull request, #15383:
URL: https://github.com/apache/kafka/pull/15383
The wake-up mechanism in the new consumer is preventing from committing within a rebalance listener callback. The reason is that we are trying to register two wake-uppable actions at the same time.
The fix is to register the wake-uppable action more closely to where we are in fact blocking on it, so that the action is not registered when we execute rebalance listeneners and callback listeners.
### 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-16009: Fix PlaintextConsumerTest. testMaxPollIntervalMsDelayInRevocation [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #15383:
URL: https://github.com/apache/kafka/pull/15383#issuecomment-1948483411
@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-16009: Fix PlaintextConsumerTest. testMaxPollIntervalMsDelayInRevocation [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #15383:
URL: https://github.com/apache/kafka/pull/15383#discussion_r1498439703
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -419,6 +422,35 @@ public void testWakeupAfterNonEmptyFetch() {
assertThrows(WakeupException.class, () -> consumer.poll(Duration.ZERO));
}
+ @Test
+ public void testCommitInRebalanceCallback() {
+ consumer = newConsumer();
+ final String topicName = "foo";
+ final int partition = 3;
+ final TopicPartition tp = new TopicPartition(topicName, partition);
+ doAnswer(invocation -> Fetch.empty()).when(fetchCollector).collectFetch(Mockito.any(FetchBuffer.class));
+ SortedSet<TopicPartition> sortedPartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
+ sortedPartitions.add(tp);
+ CompletableBackgroundEvent<Void> e = new ConsumerRebalanceListenerCallbackNeededEvent(ON_PARTITIONS_REVOKED, sortedPartitions);
+ backgroundEventQueue.add(e);
+ completeCommitApplicationEventSuccessfully();
+
+ ConsumerRebalanceListener listener = new ConsumerRebalanceListener() {
+ @Override
+ public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {
+ assertDoesNotThrow(() -> consumer.commitSync(mkMap(mkEntry(tp, new OffsetAndMetadata(0)))));
Review Comment:
If this throws, should it bubble up from `consumer.poll()` call below?
Also wondering, if we should actually verify that this code was executed after all?
```
final AtomicBoolean callbackExecuted = new AtomicBoolean(false);
ConsumerRebalanceListener listener = new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {
assertDoesNotThrow(() -> consumer.commitSync(mkMap(mkEntry(tp, new OffsetAndMetadata(0)))));
callbackExecuted.set(true);
}
}
[...]
consumer.poll(Duration.ZERO);
asserTrue(callbackExecuted.get());
```
--
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-16009: Fix PlaintextConsumerTest. testMaxPollIntervalMsDelayInRevocation [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #15383:
URL: https://github.com/apache/kafka/pull/15383
--
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-16009: Fix PlaintextConsumerTest. testMaxPollIntervalMsDelayInRevocation [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15383:
URL: https://github.com/apache/kafka/pull/15383#discussion_r1492621621
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -419,6 +422,35 @@ public void testWakeupAfterNonEmptyFetch() {
assertThrows(WakeupException.class, () -> consumer.poll(Duration.ZERO));
}
+ @Test
+ public void testCommitInCommitCallback() {
Review Comment:
oops
--
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-16009: Fix PlaintextConsumerTest. testMaxPollIntervalMsDelayInRevocation [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #15383:
URL: https://github.com/apache/kafka/pull/15383#issuecomment-1948876369
Hey @mjsax . This PR is right now waiting for @cadonna's review. If you have time, you could take a look to get into the 848 work.
--
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-16009: Fix PlaintextConsumerTest. testMaxPollIntervalMsDelayInRevocation [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #15383:
URL: https://github.com/apache/kafka/pull/15383#discussion_r1492613876
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -419,6 +422,35 @@ public void testWakeupAfterNonEmptyFetch() {
assertThrows(WakeupException.class, () -> consumer.poll(Duration.ZERO));
}
+ @Test
+ public void testCommitInCommitCallback() {
Review Comment:
We're resting here the commit on rebalance callback, so name maybe `testCommitInRebalanceCallback` ?
--
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-16009: Fix PlaintextConsumerTest. testMaxPollIntervalMsDelayInRevocation [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15383:
URL: https://github.com/apache/kafka/pull/15383#discussion_r1499319626
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -419,6 +422,35 @@ public void testWakeupAfterNonEmptyFetch() {
assertThrows(WakeupException.class, () -> consumer.poll(Duration.ZERO));
}
+ @Test
+ public void testCommitInRebalanceCallback() {
+ consumer = newConsumer();
+ final String topicName = "foo";
+ final int partition = 3;
+ final TopicPartition tp = new TopicPartition(topicName, partition);
+ doAnswer(invocation -> Fetch.empty()).when(fetchCollector).collectFetch(Mockito.any(FetchBuffer.class));
+ SortedSet<TopicPartition> sortedPartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
+ sortedPartitions.add(tp);
+ CompletableBackgroundEvent<Void> e = new ConsumerRebalanceListenerCallbackNeededEvent(ON_PARTITIONS_REVOKED, sortedPartitions);
+ backgroundEventQueue.add(e);
+ completeCommitApplicationEventSuccessfully();
+
+ ConsumerRebalanceListener listener = new ConsumerRebalanceListener() {
+ @Override
+ public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {
+ assertDoesNotThrow(() -> consumer.commitSync(mkMap(mkEntry(tp, new OffsetAndMetadata(0)))));
Review Comment:
I started a thread about the bubbling up of exceptions in listeners - I noticed that as well and wanted to bring it up - good that you remind me.
As for checking that the listener is actually executed - that's a good idea, I added a small follow-up PR, could you please review it? https://github.com/apache/kafka/pull/15420
--
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-16009: Fix PlaintextConsumerTest. testMaxPollIntervalMsDelayInRevocation [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15383:
URL: https://github.com/apache/kafka/pull/15383#discussion_r1492547506
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -384,7 +387,7 @@ public void testWakeupAfterEmptyFetch() {
doAnswer(invocation -> {
consumer.wakeup();
return Fetch.empty();
- }).when(fetchCollector).collectFetch(any(FetchBuffer.class));
+ }).doAnswer(invocation -> Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
Review Comment:
It seems that this was just a bit of a lazy test setup. We only want to wake up the consumer once for this test.
--
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