AndrewJSchofield commented on code in PR #23499:
URL: https://github.com/apache/kafka/pull/23499#discussion_r4046070003
##########
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:
Yes.
--
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]
