Fix adding headers to log change method, some variables final, update return type

Author: bbejeck

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

@@ -17,6 +17,7 @@
 package org.apache.kafka.streams.processor.internals;
 
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.common.utils.Time;
@@ -805,7 +806,7 @@ private static class StartupContext extends AbstractProcessorContext<Object, Obj
         private final StateManager stateManager;
         final StreamsMetricsImpl metricsImpl;
 
-        public StartupContext(final TaskId taskId, final StreamsConfig config, final StateManager stateManager, final StreamsMetricsImpl metricsImpl, ThreadCache cache) {
+        public StartupContext(final TaskId taskId, final StreamsConfig config, final StateManager stateManager, final StreamsMetricsImpl metricsImpl, final ThreadCache cache) {
             super(taskId, config, metricsImpl, cache);
             this.stateManager = stateManager;
             this.metricsImpl = metricsImpl;
@@ -831,7 +832,7 @@ public void registerCacheFlushListener(final String namespace, final ThreadCache
         }
 
         @Override
-        public void logChange(final String storeName, final Bytes key, final byte[] value, final long timestamp, final Position position) {
+        public void logChange(final String storeName, final Bytes key, final byte[] value, final long timestamp, final Headers headers, final Position position) {
             throw new IllegalStateException("Should not be called");
         }
 

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

@@ -327,7 +327,7 @@ private Collection<Task> assignActiveTaskFromStartupState(final Map<TaskId, Set<
                     assignedTasks.put(taskId, entry.getValue());
                 }
             }
-            return activeTaskCreator.createTasks(mainConsumer, assignedTasks);
+            return new ArrayList<>(activeTaskCreator.createTasks(mainConsumer, assignedTasks));
         } else {
             return Collections.emptySet();
         }
@@ -343,7 +343,7 @@ private Collection<Task> assignStartupTasks(final Map<TaskId, Set<TopicPartition
                     assignedTasks.put(taskId, inputPartitions);
                 }
             }
-            return standbyTaskCreator.createTasks(assignedTasks);
+            return new ArrayList<>(standbyTaskCreator.createTasks(assignedTasks));
         } else {
             return Collections.emptySet();
         }

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

@@ -892,7 +892,7 @@ public void shouldCloseStartupStateOnAutoCleanUp() {
         // we need to set this because the auto-cleanup uses the last-modified time from the filesystem,
         // which can't be mocked
         time.setCurrentTimeMs(System.currentTimeMillis());
-        TaskId taskId = new TaskId(0, 0);
+        final TaskId taskId = new TaskId(0, 0);
 
         final StateStore store = initializeStartupStores(taskId, true);
 

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

@@ -4902,7 +4902,7 @@ public void shouldCreateActiveTaskFromStartupStateStore() {
         when(stateUpdater.tasks()).thenReturn(Collections.singleton(activeTask));
         when(stateUpdater.standbyTasks()).thenReturn(Collections.emptySet());
 
-        InOrder inOrder = inOrder(activeTaskCreator);
+        final InOrder inOrder = inOrder(activeTaskCreator);
         inOrder.verify(activeTaskCreator).createTasks(same(consumer), eq(Map.of(taskId00, taskId00Partitions)));
         inOrder.verify(activeTaskCreator).createTasks(consumer, emptyMap());
 
@@ -4941,7 +4941,7 @@ public void shouldCreateStandbyTaskFromStartupStateStore() {
 
         // ensure we didn't construct any new Tasks, or recycle an existing Task; we only used the one we already have
         verify(activeTaskCreator, times(2)).createTasks(any(), eq(Collections.emptyMap()));
-        InOrder inOrder = inOrder(standbyTaskCreator);
+        final InOrder inOrder = inOrder(standbyTaskCreator);
         inOrder.verify(standbyTaskCreator).createTasks(Map.of(taskId00, taskId00Partitions));
         inOrder.verify(standbyTaskCreator).createTasks(Collections.emptyMap());
         verifyNoMoreInteractions(activeTaskCreator);