You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "AndrewJSchofield (via GitHub)" <gi...@apache.org> on 2026/01/07 15:15:56 UTC
[PR] KAFKA-20052: Share commitSync for deleted topic [kafka]
AndrewJSchofield opened a new pull request, #21265:
URL: https://github.com/apache/kafka/pull/21265
When a topic with outstanding acknowledgements is deleted, SHARE_ACKNOWLEDGE fails with a topic-level UNKNOWN_TOPIC_ID error code. This error code was not being handled as an immediate failure, which meant that the request would be retried until it timed out. As a result, `ShareConsumer.commitSync()` failed with a `TimeoutException` for the partition, rather than `UnknownTopicIdException`.
In addition, there was a mistake in `ShareInFlightBatch` which meant that an NPE occurred as the timeout was delivered.
--
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-20052: Share commitSync for deleted topic [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #21265:
URL: https://github.com/apache/kafka/pull/21265
--
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-20052: Share commitSync for deleted topic [kafka]
Posted by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org>.
chirag-wadhwa5 commented on code in PR #21265:
URL: https://github.com/apache/kafka/pull/21265#discussion_r2669367308
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java:
##########
@@ -2926,6 +2927,37 @@ public void testExplicitAcknowledgeReleaseAcceptInRecordLimitMode() {
}
}
+ @ClusterTest
+ public void testCommitSyncFailsForDeletedTopic() throws InterruptedException {
+ Uuid topicId = createTopic("baz", 1, 1);
+ alterShareAutoOffsetReset("group1", "earliest");
+ try (Producer<byte[], byte[]> producer = createProducer();
+ ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+ "group1",
+ Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, EXPLICIT))
+ ) {
+ for (int i = 0; i < 10; i++) {
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>("baz", 0, null, "key".getBytes(), ("Message " + i).getBytes());
+ producer.send(record);
+ }
+ producer.flush();
+
+ shareConsumer.subscribe(List.of("baz"));
+ ConsumerRecords<byte[], byte[]> records = waitedPoll(shareConsumer, 2500L, 10);
+ assertEquals(10, records.count());
+
+ records.forEach(shareConsumer::acknowledge);
+
+ // Topic deletion does not necessarily become apparent across the cluster immediately, so sleep a short while
+ deleteTopic("baz");
+ Thread.sleep(5000);
Review Comment:
Thanks for the PR
Nit: Can we use waitUntilTrue here instead of a sleep ? Maybe listing the topics and verifying that the topic in consideration is not part of the list can be the condition ?
--
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-20052: Share commitSync for deleted topic [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #21265:
URL: https://github.com/apache/kafka/pull/21265#discussion_r2669920788
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java:
##########
@@ -2926,6 +2927,37 @@ public void testExplicitAcknowledgeReleaseAcceptInRecordLimitMode() {
}
}
+ @ClusterTest
+ public void testCommitSyncFailsForDeletedTopic() throws InterruptedException {
+ Uuid topicId = createTopic("baz", 1, 1);
+ alterShareAutoOffsetReset("group1", "earliest");
+ try (Producer<byte[], byte[]> producer = createProducer();
+ ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+ "group1",
+ Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, EXPLICIT))
+ ) {
+ for (int i = 0; i < 10; i++) {
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>("baz", 0, null, "key".getBytes(), ("Message " + i).getBytes());
+ producer.send(record);
+ }
+ producer.flush();
+
+ shareConsumer.subscribe(List.of("baz"));
+ ConsumerRecords<byte[], byte[]> records = waitedPoll(shareConsumer, 2500L, 10);
+ assertEquals(10, records.count());
+
+ records.forEach(shareConsumer::acknowledge);
+
+ // Topic deletion does not necessarily become apparent across the cluster immediately, so sleep a short while
+ deleteTopic("baz");
+ Thread.sleep(5000);
Review Comment:
I don't think so because it's still non-deterministic. What we actually need is KIP-1243.
--
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