apoorvmittal10 commented on code in PR #23499:
URL: https://github.com/apache/kafka/pull/23499#discussion_r4039214471
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -227,11 +236,17 @@ public boolean checkAllInFlightAreAcknowledged() {
*/
public Map<TopicIdPartition, NodeAcknowledgements>
takeAcknowledgedRecords() {
Map<TopicIdPartition, NodeAcknowledgements> acknowledgementMap = new
LinkedHashMap<>();
- batches.forEach((tip, batch) -> {
- int nodeId = batch.nodeId();
- Acknowledgements acknowledgements =
batch.takeAcknowledgedRecords();
- if (!acknowledgements.isEmpty())
- acknowledgementMap.put(tip, new NodeAcknowledgements(nodeId,
acknowledgements));
+ batches.forEach((tip, batchList) -> {
+ if (!batchList.isEmpty()) {
+ Acknowledgements acknowledgements = Acknowledgements.empty();
+ int nodeId = batchList.get(0).nodeId();
Review Comment:
Is there a possibility on leader change the node id in the first batch is
the older node id? If yes, then acks can go to incorrect broker.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareInFlightBatch.java:
##########
@@ -93,11 +93,8 @@ public void addGap(long offset) {
acknowledgements.addGap(offset);
}
- public void merge(ShareInFlightBatch<K, V> other) {
- inFlightRecords.putAll(other.inFlightRecords);
- if (other.checkForRenewAcknowledgements) {
- checkForRenewAcknowledgements = true;
- }
+ public boolean isInFlight(long offset) {
Review Comment:
Should it be package private? Seems being used in package and test.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -156,11 +160,15 @@ public void takeRenewedRecords() {
* @param type The acknowledge type which indicates whether it was
processed successfully
*/
public void acknowledge(final ConsumerRecord<K, V> record, final
AcknowledgeType type) {
- for (Map.Entry<TopicIdPartition, ShareInFlightBatch<K, V>> tipBatch :
batches.entrySet()) {
- TopicIdPartition tip = tipBatch.getKey();
+ for (Map.Entry<TopicIdPartition, List<ShareInFlightBatch<K, V>>> entry
: batches.entrySet()) {
+ TopicIdPartition tip = entry.getKey();
if (tip.topic().equals(record.topic()) && (tip.partition() ==
record.partition())) {
- tipBatch.getValue().acknowledge(record, type);
- return;
+ for (ShareInFlightBatch<K, V> batch : entry.getValue()) {
+ if (batch.isInFlight(record.offset())) {
Review Comment:
Earlier we used to check `checkForRenewAcknowledgements` before merging and
the only Surviving ShareInFlightBatch used to capture that. Is that not needed
to be checked anywhere now?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetchCollector.java:
##########
@@ -118,7 +118,10 @@ public ShareFetch<K, V> collect(final ShareFetchBuffer
fetchBuffer) {
}
}
}
- } catch (KafkaException e) {
+ } catch (RuntimeException e) {
Review Comment:
Query: Why do we need to broaden the exception types?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -40,37 +42,31 @@
* @param <V> The record value
*/
public class ShareFetch<K, V> {
- private final Map<TopicIdPartition, ShareInFlightBatch<K, V>> batches;
+ private final Map<TopicIdPartition, List<ShareInFlightBatch<K, V>>>
batches;
private Optional<Integer> acquisitionLockTimeoutMs;
private Optional<Integer> acquisitionLockTimeoutMsRenewed;
public static <K, V> ShareFetch<K, V> empty() {
return new ShareFetch<>(new HashMap<>(), Optional.empty());
}
- private ShareFetch(Map<TopicIdPartition, ShareInFlightBatch<K, V>>
batches, Optional<Integer> acquisitionLockTimeoutMs) {
+ private ShareFetch(Map<TopicIdPartition, List<ShareInFlightBatch<K, V>>>
batches, Optional<Integer> acquisitionLockTimeoutMs) {
this.batches = batches;
this.acquisitionLockTimeoutMs = acquisitionLockTimeoutMs;
this.acquisitionLockTimeoutMsRenewed = Optional.empty();
}
/**
* Add another {@link ShareInFlightBatch} to this one; all of its records
will be added to this object's
- * {@link #records() records}.
+ * {@link #records() records}. Generally, we will only have one {@link
ShareInFlightBatch} for a partition
+ * at a time, but in some rare cases (such as partition leader changes),
there might be more than one.
Review Comment:
Is it also true for the renew case?
--
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]
