chia7712 commented on code in PR #23499:
URL: https://github.com/apache/kafka/pull/23499#discussion_r4059262626
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -91,17 +94,17 @@ public Map<TopicPartition, List<ConsumerRecord<K, V>>>
records() {
public int numRecords() {
int numRecords = 0;
if (!batches.isEmpty()) {
- Iterator<Map.Entry<TopicIdPartition, ShareInFlightBatch<K, V>>>
iterator = batches.entrySet().iterator();
+ Iterator<Map.Entry<TopicIdPartition, List<ShareInFlightBatch<K,
V>>>> iterator = batches.entrySet().iterator();
while (iterator.hasNext()) {
- Map.Entry<TopicIdPartition, ShareInFlightBatch<K, V>> entry =
iterator.next();
- ShareInFlightBatch<K, V> batch = entry.getValue();
- if (batch.isEmpty()) {
- if (!batch.hasRenewals()) {
- iterator.remove();
- }
- } else {
+ Map.Entry<TopicIdPartition, List<ShareInFlightBatch<K, V>>>
entry = iterator.next();
+ List<ShareInFlightBatch<K, V>> batchList = entry.getValue();
+ batchList.removeIf(batch -> batch.isEmpty() &&
!batch.hasRenewals());
Review Comment:
The `removeIf` does not impact the return value of `numRecords`, so why do
we run it in this method?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -81,7 +78,13 @@ public void add(TopicIdPartition partition,
ShareInFlightBatch<K, V> batch) {
*/
public Map<TopicPartition, List<ConsumerRecord<K, V>>> records() {
final LinkedHashMap<TopicPartition, List<ConsumerRecord<K, V>>> result
= new LinkedHashMap<>();
- batches.forEach((tip, batch) -> result.put(tip.topicPartition(),
batch.getInFlightRecords()));
+ batches.forEach((tip, batchList) -> {
+ List<ConsumerRecord<K, V>> records = new ArrayList<>();
+ for (ShareInFlightBatch<K, V> batch : batchList) {
+ records.addAll(batch.getInFlightRecords());
+ }
+ result.put(tip.topicPartition(), records);
+ });
return Map.copyOf(result);
Review Comment:
I'm a bit confused about this copy. It breaks the order from
`LinkedHashMap`, and copying the map is more expensive than creating an
unmodifiable wrapper.
If the order is not a thing, we could use `HashMap` and then return an
unmodifiable wrapper instead.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -227,11 +237,16 @@ 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:
Should we run `removeIf` before calling `get(0)`? It won't fix the issue
raised by @apoorvmittal10, but it makes the node id selection more robust.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -40,37 +42,32 @@
* @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 cases (such as a repeated request after an empty
response, or partition leader
+ * changes), there might be more than one.
*
* @param partition the topic-partition
* @param batch the batch to add; may not be null
*/
public void add(TopicIdPartition partition, ShareInFlightBatch<K, V>
batch) {
Objects.requireNonNull(batch);
- ShareInFlightBatch<K, V> currentBatch = this.batches.get(partition);
- if (currentBatch == null) {
- this.batches.put(partition, batch);
- } else {
- // This case shouldn't usually happen because we only send one
fetch at a time per partition,
- // but it might conceivably happen in some rare cases (such as
partition leader changes).
- currentBatch.merge(batch);
- }
+ batches.computeIfAbsent(partition, k -> new LinkedList<>()).add(batch);
Review Comment:
It seems most usages are foreach, so `ArrayList` should be the better
choice. The cost of `removeIf` should be almost equal between both
implementations
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -81,7 +78,13 @@ public void add(TopicIdPartition partition,
ShareInFlightBatch<K, V> batch) {
*/
public Map<TopicPartition, List<ConsumerRecord<K, V>>> records() {
final LinkedHashMap<TopicPartition, List<ConsumerRecord<K, V>>> result
= new LinkedHashMap<>();
- batches.forEach((tip, batch) -> result.put(tip.topicPartition(),
batch.getInFlightRecords()));
+ batches.forEach((tip, batchList) -> {
+ List<ConsumerRecord<K, V>> records = new ArrayList<>();
+ for (ShareInFlightBatch<K, V> batch : batchList) {
+ records.addAll(batch.getInFlightRecords());
+ }
+ result.put(tip.topicPartition(), records);
Review Comment:
ditto: should we remove the empty batch?
--
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]
