You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "ShivsundarR (via GitHub)" <gi...@apache.org> on 2024/10/18 06:35:08 UTC
[PR] KAFKA-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
ShivsundarR opened a new pull request, #17537:
URL: https://github.com/apache/kafka/pull/17537
*What*
Currently in `ShareConsumeRequestManager`, after we receive a `ShareFetchResponse`, if the subscription changes before we acknowledge(via `ShareFetch`), then we do not acknowledge the records which are not part of the updated subscription.
Instead we must acknowledge all the records that we had received irrespective of the current subscription.
This bug is only when we are acknowledging via `ShareFetch` where we use `SubscriptionState::fetchablePartitions` to obtain the partitions to fetch.
In `ShareAcknowledge`, as we are getting the partitions from the active share sessions, even if the subscription changed, the session would remain active.
In future, we would ideally refactor to make the `ShareFetch` also use the active share sessions instead of `SubscriptionState`.
--
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-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17537:
URL: https://github.com/apache/kafka/pull/17537#discussion_r1818602360
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -1214,6 +1217,38 @@ Tuple<AcknowledgeRequestState> requestStates(int nodeId) {
return acknowledgeRequestStates.get(nodeId);
}
+ static class IdAndPartition {
+ private final Uuid topicId;
+ private final int partitionIndex;
+
+ IdAndPartition(Uuid topicId, int partitionIndex) {
+ this.topicId = topicId;
+ this.partitionIndex = partitionIndex;
+ }
+
+ int getPartitionIndex() {
+ return partitionIndex;
+ }
+
+ Uuid getTopicId() {
+ return topicId;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(topicId, partitionIndex);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ IdAndPartition that = (IdAndPartition) o;
+ return Objects.equals(topicId, that.topicId) &&
+ Objects.equals(partitionIndex, that.partitionIndex);
Review Comment:
I'd prefer to see the primitive type just compared with `==`.
--
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-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on code in PR #17537:
URL: https://github.com/apache/kafka/pull/17537#discussion_r1827556947
##########
core/src/test/java/kafka/test/api/ShareConsumerTest.java:
##########
@@ -255,6 +257,44 @@ public void testSubscriptionAndPollMultiple(String persister) {
producer.close();
}
+ @ParameterizedTest(name = "{displayName}.persister={0}")
+ @ValueSource(strings = {NO_OP_PERSISTER, DEFAULT_STATE_PERSISTER})
+ public void testAcknowledgementSentOnSubscriptionChange(String persister) throws ExecutionException, InterruptedException {
+ Map<TopicPartition, Set<Long>> partitionOffsetsMap = new HashMap<>();
+ Map<TopicPartition, Exception> partitionExceptionMap = new HashMap<>();
+
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), "value".getBytes());
+ KafkaProducer<byte[], byte[]> producer = createProducer(new ByteArraySerializer(), new ByteArraySerializer());
+ producer.send(record);
+ ProducerRecord<byte[], byte[]> record2 = new ProducerRecord<>(tp2.topic(), tp2.partition(), null, "key".getBytes(), "value".getBytes());
+ producer.send(record2).get();
+ KafkaShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(new ByteArrayDeserializer(), new ByteArrayDeserializer(), "group1");
+ shareConsumer.setAcknowledgementCommitCallback(new TestableAcknowledgeCommitCallback(partitionOffsetsMap, partitionExceptionMap));
+
+ shareConsumer.subscribe(Collections.singleton(tp.topic()));
+
+ ConsumerRecords<byte[], byte[]> records = shareConsumer.poll(Duration.ofMillis(5000));
+ assertEquals(1, records.count());
+
+ shareConsumer.subscribe(Collections.singletonList(tp2.topic()));
+
+ // Waiting for heartbeat to propagate the subscription change.
+ TestUtils.waitForCondition(() -> shareConsumer.poll(Duration.ofMillis(2000)).count() == 1,
+ DEFAULT_MAX_WAIT_MS, 100L, () -> "Failed to consume records from the updated subscription");
+
+ producer.send(record2).get();
+
+ //Starting the 3rd poll to invoke the callback
+ shareConsumer.poll(Duration.ofMillis(500));
+
+ // Verifying if the callback was invoked for the partitions in the old subscription.
+ assertTrue(partitionExceptionMap.containsKey(tp));
+ assertNull(partitionExceptionMap.get(tp));
+
+ producer.close();
Review Comment:
nit: can we use try..finally to close the resources here and other tests
--
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-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on code in PR #17537:
URL: https://github.com/apache/kafka/pull/17537#discussion_r1827556947
##########
core/src/test/java/kafka/test/api/ShareConsumerTest.java:
##########
@@ -255,6 +257,44 @@ public void testSubscriptionAndPollMultiple(String persister) {
producer.close();
}
+ @ParameterizedTest(name = "{displayName}.persister={0}")
+ @ValueSource(strings = {NO_OP_PERSISTER, DEFAULT_STATE_PERSISTER})
+ public void testAcknowledgementSentOnSubscriptionChange(String persister) throws ExecutionException, InterruptedException {
+ Map<TopicPartition, Set<Long>> partitionOffsetsMap = new HashMap<>();
+ Map<TopicPartition, Exception> partitionExceptionMap = new HashMap<>();
+
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), "value".getBytes());
+ KafkaProducer<byte[], byte[]> producer = createProducer(new ByteArraySerializer(), new ByteArraySerializer());
+ producer.send(record);
+ ProducerRecord<byte[], byte[]> record2 = new ProducerRecord<>(tp2.topic(), tp2.partition(), null, "key".getBytes(), "value".getBytes());
+ producer.send(record2).get();
+ KafkaShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(new ByteArrayDeserializer(), new ByteArrayDeserializer(), "group1");
+ shareConsumer.setAcknowledgementCommitCallback(new TestableAcknowledgeCommitCallback(partitionOffsetsMap, partitionExceptionMap));
+
+ shareConsumer.subscribe(Collections.singleton(tp.topic()));
+
+ ConsumerRecords<byte[], byte[]> records = shareConsumer.poll(Duration.ofMillis(5000));
+ assertEquals(1, records.count());
+
+ shareConsumer.subscribe(Collections.singletonList(tp2.topic()));
+
+ // Waiting for heartbeat to propagate the subscription change.
+ TestUtils.waitForCondition(() -> shareConsumer.poll(Duration.ofMillis(2000)).count() == 1,
+ DEFAULT_MAX_WAIT_MS, 100L, () -> "Failed to consume records from the updated subscription");
+
+ producer.send(record2).get();
+
+ //Starting the 3rd poll to invoke the callback
+ shareConsumer.poll(Duration.ofMillis(500));
+
+ // Verifying if the callback was invoked for the partitions in the old subscription.
+ assertTrue(partitionExceptionMap.containsKey(tp));
+ assertNull(partitionExceptionMap.get(tp));
+
+ producer.close();
Review Comment:
nit: can we use try..finally to close the resources
--
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-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #17537:
URL: https://github.com/apache/kafka/pull/17537#discussion_r1827668994
##########
core/src/test/java/kafka/test/api/ShareConsumerTest.java:
##########
@@ -255,6 +257,44 @@ public void testSubscriptionAndPollMultiple(String persister) {
producer.close();
}
+ @ParameterizedTest(name = "{displayName}.persister={0}")
+ @ValueSource(strings = {NO_OP_PERSISTER, DEFAULT_STATE_PERSISTER})
+ public void testAcknowledgementSentOnSubscriptionChange(String persister) throws ExecutionException, InterruptedException {
+ Map<TopicPartition, Set<Long>> partitionOffsetsMap = new HashMap<>();
+ Map<TopicPartition, Exception> partitionExceptionMap = new HashMap<>();
+
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), "value".getBytes());
+ KafkaProducer<byte[], byte[]> producer = createProducer(new ByteArraySerializer(), new ByteArraySerializer());
+ producer.send(record);
+ ProducerRecord<byte[], byte[]> record2 = new ProducerRecord<>(tp2.topic(), tp2.partition(), null, "key".getBytes(), "value".getBytes());
+ producer.send(record2).get();
+ KafkaShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(new ByteArrayDeserializer(), new ByteArrayDeserializer(), "group1");
+ shareConsumer.setAcknowledgementCommitCallback(new TestableAcknowledgeCommitCallback(partitionOffsetsMap, partitionExceptionMap));
+
+ shareConsumer.subscribe(Collections.singleton(tp.topic()));
+
+ ConsumerRecords<byte[], byte[]> records = shareConsumer.poll(Duration.ofMillis(5000));
+ assertEquals(1, records.count());
+
+ shareConsumer.subscribe(Collections.singletonList(tp2.topic()));
+
+ // Waiting for heartbeat to propagate the subscription change.
+ TestUtils.waitForCondition(() -> shareConsumer.poll(Duration.ofMillis(2000)).count() == 1,
+ DEFAULT_MAX_WAIT_MS, 100L, () -> "Failed to consume records from the updated subscription");
+
+ producer.send(record2).get();
+
+ //Starting the 3rd poll to invoke the callback
+ shareConsumer.poll(Duration.ofMillis(500));
+
+ // Verifying if the callback was invoked for the partitions in the old subscription.
+ assertTrue(partitionExceptionMap.containsKey(tp));
+ assertNull(partitionExceptionMap.get(tp));
+
+ producer.close();
Review Comment:
Okay, as this change will be applied for all the existing tests, should I do it in a separate PR or making the change here is fine?
--
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-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17537:
URL: https://github.com/apache/kafka/pull/17537#discussion_r1812494263
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -172,14 +174,53 @@ public PollResult poll(long currentTimeMs) {
metricsManager.recordAcknowledgementSent(acknowledgementsToSend.size());
}
handler.addPartitionToFetch(tip, acknowledgementsToSend);
+ fetchedPartitions.add(tip);
- log.debug("Added fetch request for partition {} to node {}", partition, node.id());
+ log.debug("Added fetch request for partition {} to node {}", tip, node.id());
}
}
+ // Map storing the list of partitions to forget in the upcoming request.
+ Map<Node, List<TopicIdPartition>> partitionsToForgetMap = new HashMap<>();
+ Cluster cluster = metadata.fetch();
+ // Iterating over the session handlers to see if there are acknowledgements to be sent for partitions
+ // which are no longer part of the current subscription.
+ sessionHandlers.forEach((nodeId, sessionHandler) -> {
+ Node node = cluster.nodeById(nodeId);
+ if (node != null) {
+ if (nodesWithPendingRequests.contains(node.id())) {
+ log.trace("Skipping fetch because previous fetch request to {} has not been processed", node.id());
+ } else {
+ for (TopicIdPartition tip : sessionHandler.sessionPartitions()) {
+ if (!fetchedPartitions.contains(tip)) {
+ Acknowledgements acknowledgementsToSend = fetchAcknowledgementsMap.get(tip);
+ if (acknowledgementsToSend != null) {
+ metricsManager.recordAcknowledgementSent(acknowledgementsToSend.size());
+ }
+ sessionHandler.addPartitionToFetch(tip, acknowledgementsToSend);
+ partitionsToForgetMap.putIfAbsent(node, new ArrayList<>());
+ partitionsToForgetMap.get(node).add(tip);
+
+ forgottenTopicNames.putIfAbsent(tip.topicId(), tip.topic());
+ fetchedPartitions.add(tip);
+ log.debug("Added fetch request for partition {} to node {}", tip, node.id());
+ }
+ }
+ }
+ }
+ });
+
Map<Node, ShareFetchRequest.Builder> builderMap = new LinkedHashMap<>();
for (Map.Entry<Node, ShareSessionHandler> entry : handlerMap.entrySet()) {
builderMap.put(entry.getKey(), entry.getValue().newShareFetchBuilder(groupId, fetchConfig));
+ Node node = entry.getKey();
+ ShareFetchRequestData data = builderMap.get(entry.getKey()).data();
+ if (partitionsToForgetMap.containsKey(node)) {
+ if (data.forgottenTopicsData() == null) {
+ data.setForgottenTopicsData(new ArrayList<>());
+ }
+ ShareFetchRequest.Builder.updateForgottenData(partitionsToForgetMap.get(node), data);
Review Comment:
I'd prefer this to be a method on the `ShareFetchRequest.Builder` instance, rather than a static method. Look at `FetchRequest.Builder.addToForgottenTopicMap`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -594,7 +635,7 @@ private void handleShareFetchSuccess(Node fetchTarget,
topicResponse.partitions().forEach(partition ->
responseData.put(new TopicIdPartition(topicResponse.topicId(),
partition.partitionIndex(),
- metadata.topicNames().get(topicResponse.topicId())), partition)));
+ metadata.topicNames().getOrDefault(topicResponse.topicId(), forgottenTopicNames.remove(topicResponse.topicId()))), partition)));
Review Comment:
This seems wrong to me. Each partition for a forgotten topic will cause `forgottenTopicNames.remove()` to be called, and only the first call will be valid. The handling of the forgotten topic names needs to be a bit less clever.
--
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-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy merged PR #17537:
URL: https://github.com/apache/kafka/pull/17537
--
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-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17537:
URL: https://github.com/apache/kafka/pull/17537#discussion_r1818812830
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -172,14 +175,53 @@ public PollResult poll(long currentTimeMs) {
metricsManager.recordAcknowledgementSent(acknowledgementsToSend.size());
}
handler.addPartitionToFetch(tip, acknowledgementsToSend);
+ fetchedPartitions.add(tip);
- log.debug("Added fetch request for partition {} to node {}", partition, node.id());
+ log.debug("Added fetch request for partition {} to node {}", tip, node.id());
}
}
+ // Map storing the list of partitions to forget in the upcoming request.
+ Map<Node, List<TopicIdPartition>> partitionsToForgetMap = new HashMap<>();
+ Cluster cluster = metadata.fetch();
+ // Iterating over the session handlers to see if there are acknowledgements to be sent for partitions
+ // which are no longer part of the current subscription.
+ sessionHandlers.forEach((nodeId, sessionHandler) -> {
+ Node node = cluster.nodeById(nodeId);
+ if (node != null) {
+ if (nodesWithPendingRequests.contains(node.id())) {
+ log.trace("Skipping fetch because previous fetch request to {} has not been processed", node.id());
+ } else {
+ for (TopicIdPartition tip : sessionHandler.sessionPartitions()) {
+ if (!fetchedPartitions.contains(tip)) {
+ Acknowledgements acknowledgementsToSend = fetchAcknowledgementsMap.get(tip);
+ if (acknowledgementsToSend != null) {
+ metricsManager.recordAcknowledgementSent(acknowledgementsToSend.size());
+ }
+ sessionHandler.addPartitionToFetch(tip, acknowledgementsToSend);
+ partitionsToForgetMap.putIfAbsent(node, new ArrayList<>());
+ partitionsToForgetMap.get(node).add(tip);
+
+ forgottenTopicNames.putIfAbsent(new IdAndPartition(tip.topicId(), tip.partition()), tip.topic());
+ fetchedPartitions.add(tip);
+ log.debug("Added fetch request for partition {} to node {}", tip, node.id());
Review Comment:
This could say "for previously subscribed partition" in the log message to make it clear that this is essentially finishing off the processing for a partition which is no longer subscribed. It's not really a proper fetch now.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -172,14 +175,53 @@ public PollResult poll(long currentTimeMs) {
metricsManager.recordAcknowledgementSent(acknowledgementsToSend.size());
}
handler.addPartitionToFetch(tip, acknowledgementsToSend);
+ fetchedPartitions.add(tip);
- log.debug("Added fetch request for partition {} to node {}", partition, node.id());
+ log.debug("Added fetch request for partition {} to node {}", tip, node.id());
}
}
+ // Map storing the list of partitions to forget in the upcoming request.
+ Map<Node, List<TopicIdPartition>> partitionsToForgetMap = new HashMap<>();
+ Cluster cluster = metadata.fetch();
+ // Iterating over the session handlers to see if there are acknowledgements to be sent for partitions
+ // which are no longer part of the current subscription.
+ sessionHandlers.forEach((nodeId, sessionHandler) -> {
+ Node node = cluster.nodeById(nodeId);
+ if (node != null) {
+ if (nodesWithPendingRequests.contains(node.id())) {
+ log.trace("Skipping fetch because previous fetch request to {} has not been processed", node.id());
+ } else {
+ for (TopicIdPartition tip : sessionHandler.sessionPartitions()) {
+ if (!fetchedPartitions.contains(tip)) {
+ Acknowledgements acknowledgementsToSend = fetchAcknowledgementsMap.get(tip);
+ if (acknowledgementsToSend != null) {
+ metricsManager.recordAcknowledgementSent(acknowledgementsToSend.size());
+ }
+ sessionHandler.addPartitionToFetch(tip, acknowledgementsToSend);
+ partitionsToForgetMap.putIfAbsent(node, new ArrayList<>());
+ partitionsToForgetMap.get(node).add(tip);
+
+ forgottenTopicNames.putIfAbsent(new IdAndPartition(tip.topicId(), tip.partition()), tip.topic());
+ fetchedPartitions.add(tip);
+ log.debug("Added fetch request for partition {} to node {}", tip, node.id());
+ }
+ }
+ }
+ }
+ });
+
Map<Node, ShareFetchRequest.Builder> builderMap = new LinkedHashMap<>();
for (Map.Entry<Node, ShareSessionHandler> entry : handlerMap.entrySet()) {
builderMap.put(entry.getKey(), entry.getValue().newShareFetchBuilder(groupId, fetchConfig));
+ Node node = entry.getKey();
+ ShareFetchRequest.Builder builder = builderMap.get(entry.getKey());
Review Comment:
This is a bit odd. Surely it would be better to assign the `node` and `builder` first, using `entry.getKey/value()`, and then add the builder to the map, rather than adding the builder into the map, and then re-get it immediately after.
--
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-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #17537:
URL: https://github.com/apache/kafka/pull/17537#discussion_r1818734310
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -1244,8 +1244,8 @@ public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
IdAndPartition that = (IdAndPartition) o;
- return Objects.equals(topicId, that.topicId) &&
- Objects.equals(partitionIndex, that.partitionIndex);
+ return topicId == that.topicId &&
Review Comment:
Thanks yeah missed that. Updated the PR now.
--
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-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #17537:
URL: https://github.com/apache/kafka/pull/17537#discussion_r1815383443
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -594,7 +635,7 @@ private void handleShareFetchSuccess(Node fetchTarget,
topicResponse.partitions().forEach(partition ->
responseData.put(new TopicIdPartition(topicResponse.topicId(),
partition.partitionIndex(),
- metadata.topicNames().get(topicResponse.topicId())), partition)));
+ metadata.topicNames().getOrDefault(topicResponse.topicId(), forgottenTopicNames.remove(topicResponse.topicId()))), partition)));
Review Comment:
Right, thanks for catching this. I have now used `partitionIndex` along with `topicId` as the key for the map to ensure we have removed only the entry corresponding to that partition.
--
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-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #17537:
URL: https://github.com/apache/kafka/pull/17537#discussion_r1815384317
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -172,14 +174,53 @@ public PollResult poll(long currentTimeMs) {
metricsManager.recordAcknowledgementSent(acknowledgementsToSend.size());
}
handler.addPartitionToFetch(tip, acknowledgementsToSend);
+ fetchedPartitions.add(tip);
- log.debug("Added fetch request for partition {} to node {}", partition, node.id());
+ log.debug("Added fetch request for partition {} to node {}", tip, node.id());
}
}
+ // Map storing the list of partitions to forget in the upcoming request.
+ Map<Node, List<TopicIdPartition>> partitionsToForgetMap = new HashMap<>();
+ Cluster cluster = metadata.fetch();
+ // Iterating over the session handlers to see if there are acknowledgements to be sent for partitions
+ // which are no longer part of the current subscription.
+ sessionHandlers.forEach((nodeId, sessionHandler) -> {
+ Node node = cluster.nodeById(nodeId);
+ if (node != null) {
+ if (nodesWithPendingRequests.contains(node.id())) {
+ log.trace("Skipping fetch because previous fetch request to {} has not been processed", node.id());
+ } else {
+ for (TopicIdPartition tip : sessionHandler.sessionPartitions()) {
+ if (!fetchedPartitions.contains(tip)) {
+ Acknowledgements acknowledgementsToSend = fetchAcknowledgementsMap.get(tip);
+ if (acknowledgementsToSend != null) {
+ metricsManager.recordAcknowledgementSent(acknowledgementsToSend.size());
+ }
+ sessionHandler.addPartitionToFetch(tip, acknowledgementsToSend);
+ partitionsToForgetMap.putIfAbsent(node, new ArrayList<>());
+ partitionsToForgetMap.get(node).add(tip);
+
+ forgottenTopicNames.putIfAbsent(tip.topicId(), tip.topic());
+ fetchedPartitions.add(tip);
+ log.debug("Added fetch request for partition {} to node {}", tip, node.id());
+ }
+ }
+ }
+ }
+ });
+
Map<Node, ShareFetchRequest.Builder> builderMap = new LinkedHashMap<>();
for (Map.Entry<Node, ShareSessionHandler> entry : handlerMap.entrySet()) {
builderMap.put(entry.getKey(), entry.getValue().newShareFetchBuilder(groupId, fetchConfig));
+ Node node = entry.getKey();
+ ShareFetchRequestData data = builderMap.get(entry.getKey()).data();
+ if (partitionsToForgetMap.containsKey(node)) {
+ if (data.forgottenTopicsData() == null) {
+ data.setForgottenTopicsData(new ArrayList<>());
+ }
+ ShareFetchRequest.Builder.updateForgottenData(partitionsToForgetMap.get(node), data);
Review Comment:
Right okay, I have updated this now.
--
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-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #17537:
URL: https://github.com/apache/kafka/pull/17537#discussion_r1818854329
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -172,14 +175,53 @@ public PollResult poll(long currentTimeMs) {
metricsManager.recordAcknowledgementSent(acknowledgementsToSend.size());
}
handler.addPartitionToFetch(tip, acknowledgementsToSend);
+ fetchedPartitions.add(tip);
- log.debug("Added fetch request for partition {} to node {}", partition, node.id());
+ log.debug("Added fetch request for partition {} to node {}", tip, node.id());
}
}
+ // Map storing the list of partitions to forget in the upcoming request.
+ Map<Node, List<TopicIdPartition>> partitionsToForgetMap = new HashMap<>();
+ Cluster cluster = metadata.fetch();
+ // Iterating over the session handlers to see if there are acknowledgements to be sent for partitions
+ // which are no longer part of the current subscription.
+ sessionHandlers.forEach((nodeId, sessionHandler) -> {
+ Node node = cluster.nodeById(nodeId);
+ if (node != null) {
+ if (nodesWithPendingRequests.contains(node.id())) {
+ log.trace("Skipping fetch because previous fetch request to {} has not been processed", node.id());
+ } else {
+ for (TopicIdPartition tip : sessionHandler.sessionPartitions()) {
+ if (!fetchedPartitions.contains(tip)) {
+ Acknowledgements acknowledgementsToSend = fetchAcknowledgementsMap.get(tip);
+ if (acknowledgementsToSend != null) {
+ metricsManager.recordAcknowledgementSent(acknowledgementsToSend.size());
+ }
+ sessionHandler.addPartitionToFetch(tip, acknowledgementsToSend);
+ partitionsToForgetMap.putIfAbsent(node, new ArrayList<>());
+ partitionsToForgetMap.get(node).add(tip);
+
+ forgottenTopicNames.putIfAbsent(new IdAndPartition(tip.topicId(), tip.partition()), tip.topic());
+ fetchedPartitions.add(tip);
+ log.debug("Added fetch request for partition {} to node {}", tip, node.id());
+ }
+ }
+ }
+ }
+ });
+
Map<Node, ShareFetchRequest.Builder> builderMap = new LinkedHashMap<>();
for (Map.Entry<Node, ShareSessionHandler> entry : handlerMap.entrySet()) {
builderMap.put(entry.getKey(), entry.getValue().newShareFetchBuilder(groupId, fetchConfig));
+ Node node = entry.getKey();
+ ShareFetchRequest.Builder builder = builderMap.get(entry.getKey());
Review Comment:
Right yeah makes sense, thanks. I have updated the PR now.
--
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-17819 : Handle piggyback acknowledgements when subscription changes in ShareConsumeRequestManager. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17537:
URL: https://github.com/apache/kafka/pull/17537#discussion_r1818701642
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -1244,8 +1244,8 @@ public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
IdAndPartition that = (IdAndPartition) o;
- return Objects.equals(topicId, that.topicId) &&
- Objects.equals(partitionIndex, that.partitionIndex);
+ return topicId == that.topicId &&
Review Comment:
Only the primitive should be compared with `==`. Uuid is not a primitive.
--
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