KAFKA-8816: Make offsets immutable to users of RecordCollector.offsets (apache#7223)

Make offsets immutable to users of RecordCollector.offsets. Fix up an
existing case where offsets could be modified in this way. Add a simple
test to verify offsets cannot be changed externally.

Reviewers: Bruno Cadonna <[email protected]>, Guozhang Wang <[email protected]>, Matthias J. Sax <[email protected]>

Author: cpettitt-confluent

streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java

@@ -63,7 +63,7 @@ <K, V> void send(final String topic,
     /**
      * The last acked offsets from the internal {@link Producer}.
      *
-     * @return the map from TopicPartition to offset
+     * @return an immutable map from TopicPartition to offset
      */
     Map<TopicPartition, Long> offsets();
 

streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams.processor.internals;
 
+import java.util.Collections;
 import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -267,7 +268,7 @@ public void close() {
 
     @Override
     public Map<TopicPartition, Long> offsets() {
-        return offsets;
+        return Collections.unmodifiableMap(offsets);
     }
 
     // for testing only

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

@@ -479,7 +479,8 @@ void commit(final boolean startNewTransaction) {
 
     @Override
     protected Map<TopicPartition, Long> activeTaskCheckpointableOffsets() {
-        final Map<TopicPartition, Long> checkpointableOffsets = recordCollector.offsets();
+        final Map<TopicPartition, Long> checkpointableOffsets =
+            new HashMap<>(recordCollector.offsets());
         for (final Map.Entry<TopicPartition, Long> entry : consumedOffsets.entrySet()) {
             checkpointableOffsets.putIfAbsent(entry.getKey(), entry.getValue());
         }

streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java

@@ -50,7 +50,10 @@
 import java.util.Map;
 import java.util.concurrent.Future;
 
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -145,6 +148,31 @@ public void testStreamPartitioner() {
         assertEquals((Long) 0L, offsets.get(new TopicPartition("topic1", 2)));
     }
 
+    @Test
+    public void shouldNotAllowOffsetsToBeUpdatedExternally() {
+        final String topic = "topic1";
+        final TopicPartition topicPartition = new TopicPartition(topic, 0);
+
+        final RecordCollectorImpl collector = new RecordCollectorImpl(
+            "RecordCollectorTest-TestSpecificPartition",
+            new LogContext("RecordCollectorTest-TestSpecificPartition "),
+            new DefaultProductionExceptionHandler(),
+            new Metrics().sensor("skipped-records")
+        );
+        collector.init(new MockProducer<>(cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer));
+
+        collector.send(topic, "999", "0", null, 0, null, stringSerializer, stringSerializer);
+        collector.send(topic, "999", "0", null, 0, null, stringSerializer, stringSerializer);
+        collector.send(topic, "999", "0", null, 0, null, stringSerializer, stringSerializer);
+
+        final Map<TopicPartition, Long> offsets = collector.offsets();
+
+        assertThat(offsets.get(topicPartition), equalTo(2L));
+        assertThrows(UnsupportedOperationException.class, () -> offsets.put(new TopicPartition(topic, 0), 50L));
+
+        assertThat(collector.offsets().get(topicPartition), equalTo(2L));
+    }
+
     @SuppressWarnings("unchecked")
     @Test(expected = StreamsException.class)
     public void shouldThrowStreamsExceptionOnAnyExceptionButProducerFencedException() {