fix failing tests

apache · Jan 24, 2025 · 3a5d231 · 3a5d231
1 parent 07915d7
commit 3a5d231
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 3 deletions.
diff --git a/.../src/test/java/org/apache/kafka/streams/processor/internals/ProcessorContextImplTest.java b/.../src/test/java/org/apache/kafka/streams/processor/internals/ProcessorContextImplTest.java
@@ -645,6 +645,8 @@ public void shouldThrowUnsupportedOperationExceptionOnForward() {
         mockProcessorNodeWithLocalKeyValueStore();
 
         context = getStandbyContext();
+        context.recordContext = mock(ProcessorRecordContext.class);
+
         assertThrows(
             UnsupportedOperationException.class,
             () -> context.forward("key", "value")
@@ -665,6 +667,8 @@ public void shouldThrowUnsupportedOperationExceptionOnForwardWithTo() {
         mockProcessorNodeWithLocalKeyValueStore();
 
         context = getStandbyContext();
+        context.recordContext = mock(ProcessorRecordContext.class);
+
         assertThrows(
             UnsupportedOperationException.class,
             () -> context.forward("key", "value", To.child("child-name"))
@@ -787,7 +791,7 @@ public void shouldThrowUnsupportedOperationExceptionOnTimestamp() {
         context = getStandbyContext();
         assertThrows(
             UnsupportedOperationException.class,
-            () -> context.recordContext.timestamp()
+            () -> context.timestamp()
         );
     }
 

diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
@@ -31,6 +31,7 @@
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.apache.kafka.common.metrics.MetricConfig;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.metrics.Sensor;
@@ -61,6 +62,7 @@
 import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
 import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder;
 import org.apache.kafka.streams.processor.internals.ProcessorContextImpl;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
 import org.apache.kafka.streams.processor.internals.ProcessorStateManager;
 import org.apache.kafka.streams.processor.internals.ProcessorTopology;
 import org.apache.kafka.streams.processor.internals.RecordCollector;
@@ -447,7 +449,6 @@ private void setupGlobalTask(final Time mockWallClockTime,
                 streamsConfig.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG)
             );
             globalStateTask.initialize();
-            globalProcessorContext.setRecordContext(null);
         } else {
             globalStateManager = null;
             globalStateTask = null;
@@ -492,6 +493,7 @@ private void setupTask(final StreamsConfig streamsConfig,
                 streamsMetrics,
                 cache
             );
+            context.setRecordContext(new ProcessorRecordContext(0L, -1L, -1, null, new RecordHeaders()));
 
             task = new StreamTask(
                 TASK_ID,
@@ -511,7 +513,6 @@ private void setupTask(final StreamsConfig streamsConfig,
                 );
             task.initializeIfNeeded();
             task.completeRestoration(noOpResetter -> { });
-            task.processorContext().setRecordContext(null);
             for (final TopicPartition tp: task.inputPartitions()) {
                 task.updateNextOffsets(tp, new OffsetAndMetadata(0, Optional.empty(), ""));
             }