KafkaConsumer#commitSync should copy its input
Author: Jan Lukavsky
Reviewers: Jason Gustafson
Closes from je-ik/KAFKA-4728
@@ -1088,7 +1088,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
public void commitSync() {
acquire();
try {
- commitSync(subscriptions.allConsumed());
+ coordinator.commitOffsetsSync(subscriptions.allConsumed(), Long.MAX_VALUE);
} finally {
release();
}
@@ -1122,7 +1122,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets) {
acquire();
try {
- coordinator.commitOffsetsSync(offsets, Long.MAX_VALUE);
+ coordinator.commitOffsetsSync(new HashMap<>(offsets), Long.MAX_VALUE);
} finally {
release();
}
