DL1231 commented on code in PR #20148:
URL: https://github.com/apache/kafka/pull/20148#discussion_r2206133813
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -123,6 +123,25 @@ public void acknowledge(final ConsumerRecord<K, V> record,
AcknowledgeType type)
throw new IllegalStateException("The record cannot be acknowledged.");
}
+ /**
+ * Acknowledge a single record by its topic, partition and offset in the
current batch.
+ *
+ * @param topic The topic of the record to acknowledge
+ * @param partition The partition of the record
+ * @param offset The offset of the record
+ * @param type The acknowledgment type which indicates whether it was
processed successfully
+ */
+ public void acknowledge(final String topic, final int partition, final
long offset, final AcknowledgeType type) {
+ for (Map.Entry<TopicIdPartition, ShareInFlightBatch<K, V>> tipBatch :
batches.entrySet()) {
+ TopicIdPartition tip = tipBatch.getKey();
+ if (tip.topic().equals(topic) && (tip.partition() == partition)) {
+ tipBatch.getValue().addAcknowledgement(offset, type);
Review Comment:
Good catch! Appreciate the suggestion.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
