You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2024/05/03 12:32:25 UTC
[PR] KAFKA-16659: KafkaConsumer#position() does not respect wakup when group protocol is CONSUMER [kafka]
FrankYang0529 opened a new pull request, #15853:
URL: https://github.com/apache/kafka/pull/15853
From KafkaConsumer#position comments, we should return WakeupException when KafkaConsumer#wakeup is called.
https://github.com/apache/kafka/blob/240243b91d69c2b65b5e456065fdcce90c121b04/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java#L1228-L1252
### 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-16659: KafkaConsumer#position() does not respect wakup when group protocol is CONSUMER [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #15853:
URL: https://github.com/apache/kafka/pull/15853#discussion_r1589362449
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -906,6 +906,7 @@ public long position(TopicPartition partition, Duration timeout) {
return position.offset;
updateFetchPositions(timer);
Review Comment:
We need to make sure `updateFetchPositions` also honour the `wakeup`. For example, the following test get failed with this PR.
```java
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
@MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
@Timeout(5)
def testPositionRespectsWakeup(quorum: String, groupProtocol: String): Unit = {
val topicPartition = new TopicPartition(topic, 15)
val properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:12345");
val consumer = createConsumer(configOverrides = properties)
consumer.assign(List(topicPartition).asJava)
CompletableFuture.runAsync { () =>
TimeUnit.SECONDS.sleep(1)
consumer.wakeup()
}
assertThrows(classOf[WakeupException], () => consumer.position(topicPartition, Duration.ofSeconds(100)))
}
```
--
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-16659: KafkaConsumer#position() does not respect wakup when group protocol is CONSUMER [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #15853:
URL: https://github.com/apache/kafka/pull/15853#issuecomment-2095025582
```
./gradlew cleanTest :connect:runtime:test --tests org.apache.kafka.connect.integration.OffsetsApiIntegrationTest.testAlterSinkConnectorOffsetsOverriddenConsumerGroupId :metadata:test --tests QuorumControllerTest.testFenceMultipleBrokers :trogdor:test --tests CoordinatorTest.testTaskRequestWithOldStartMsGetsUpdated :connect:mirror:test --tests MirrorConnectorsIntegrationExactlyOnceTest.testOffsetTranslationBehindReplicationFlow --tests MirrorConnectorsWithCustomForwardingAdminIntegrationTest.testReplicateSourceDefault :core:test --tests DelegationTokenEndToEndAuthorizationWithOwnerTest.testProduceConsumeWithWildcardAcls --tests ConsumerBounceTest.testConsumptionWithBrokerFailures --tests UserClientIdQuotaTest.testThrottledProducerConsumer --tests UserClientIdQuotaTest.testQuotaOverrideDelete --tests ReplicationQuotasTest.shouldThrottleOldSegments
```
unrelated error and they pass on my local.
--
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-16659: KafkaConsumer#position() does not respect wakup when group protocol is CONSUMER [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #15853:
URL: https://github.com/apache/kafka/pull/15853
--
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-16659: KafkaConsumer#position() does not respect wakup when group protocol is CONSUMER [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #15853:
URL: https://github.com/apache/kafka/pull/15853#discussion_r1589912438
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -906,6 +906,7 @@ public long position(TopicPartition partition, Duration timeout) {
return position.offset;
updateFetchPositions(timer);
Review Comment:
I add `FetchCommittedOffsetsEvent` to `wakeupTrigger.setActiveTask`. Thanks for the review.
--
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-16659: KafkaConsumer#position() does not respect wakup when group protocol is CONSUMER [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #15853:
URL: https://github.com/apache/kafka/pull/15853#discussion_r1589364098
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -906,6 +906,7 @@ public long position(TopicPartition partition, Duration timeout) {
return position.offset;
updateFetchPositions(timer);
Review Comment:
`properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:12345");` can cause the connection waiting, and that should be "wakeupable"
--
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-16659: KafkaConsumer#position() does not respect wakup when group protocol is CONSUMER [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #15853:
URL: https://github.com/apache/kafka/pull/15853#discussion_r1589913151
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1703,7 +1704,9 @@ private boolean initWithCommittedOffsetsIfNeeded(Timer timer) {
new FetchCommittedOffsetsEvent(
initializingPartitions,
timer);
+ wakeupTrigger.setActiveTask(event.future());
final Map<TopicPartition, OffsetAndMetadata> offsets = applicationEventHandler.addAndGet(event, timer);
+ wakeupTrigger.clearTask();
Review Comment:
Please use try-finally
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -872,4 +873,38 @@ class PlaintextConsumerTest extends BaseConsumerTest {
// local metadata. However, it should give up after the user-supplied timeout has past.
assertThrows(classOf[TimeoutException], () => consumer.position(topicPartition, Duration.ofSeconds(3)))
}
+
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ @Timeout(15)
+ def testPositionRespectsWakeup(quorum: String, groupProtocol: String): Unit = {
+ val topicPartition = new TopicPartition(topic, 15)
+ val consumer = createConsumer()
+ consumer.assign(List(topicPartition).asJava)
+
+ CompletableFuture.runAsync { () =>
+ TimeUnit.SECONDS.sleep(1)
+ consumer.wakeup()
+ }
+
+ assertThrows(classOf[WakeupException], () => consumer.position(topicPartition, Duration.ofSeconds(3)))
+ }
+
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ @Timeout(15)
+ def testPositionWithErrorConnectionRespectsWakeup(quorum: String, groupProtocol: String): Unit = {
+ val topicPartition = new TopicPartition(topic, 15)
+ val properties = new Properties()
+ properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:12345")
Review Comment:
Please add comments to say this connection in un-connectable.
--
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