Revert "KAFKA-13722: remove usage of old ProcessorContext (#18292)" (#20398)

mjsax · web-flow · commit 35e594274342 · 2025-08-25T12:43:39.000+02:00
This reverts commit f13a22a. Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Eduwer Camacaro <eduwerc@gmail.com>, Mickael Maison <mickael.maison@gmail.com>,
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java
@@ -213,7 +213,7 @@ public void close() {
         private ValueAndTimestamp<KeyValue<? extends K1, ? extends V1>> mapValue(final K key, final ValueAndTimestamp<V> valueAndTimestamp) {
             return ValueAndTimestamp.make(
                 mapper.apply(key, getValueOrNull(valueAndTimestamp)),
-                valueAndTimestamp == null ? context.recordContext().timestamp() : valueAndTimestamp.timestamp()
+                valueAndTimestamp == null ? context.timestamp() : valueAndTimestamp.timestamp()
             );
         }
     }
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalProcessorContextImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalProcessorContextImpl.java
@@ -84,7 +84,7 @@ public <K, V> void forward(final Record<K, V> record, final String childName) {
 
     @Override
     public <KIn, VIn> void forward(final KIn key, final VIn value) {
-        forward(new Record<>(key, value, recordContext().timestamp(), headers()));
+        forward(new Record<>(key, value, timestamp(), headers()));
     }
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
@@ -120,8 +120,8 @@ public void update(final ConsumerRecord<byte[], byte[]> record) {
             final Record<Object, Object> toProcess = new Record<>(
                 deserialized.key(),
                 deserialized.value(),
-                processorContext.recordContext().timestamp(),
-                processorContext.recordContext().headers()
+                processorContext.timestamp(),
+                processorContext.headers()
             );
             ((SourceNode<Object, Object>) sourceNodeAndDeserializer.sourceNode()).process(toProcess);
         }
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
@@ -190,7 +190,7 @@ public <K, V> void forward(final K key,
         final Record<K, V> toForward = new Record<>(
             key,
             value,
-            recordContext.timestamp(),
+            timestamp(),
             headers()
         );
         forward(toForward);
@@ -204,7 +204,7 @@ public <K, V> void forward(final K key,
         final Record<K, V> toForward = new Record<>(
             key,
             value,
-            toInternal.hasTimestamp() ? toInternal.timestamp() : recordContext.timestamp(),
+            toInternal.hasTimestamp() ? toInternal.timestamp() : timestamp(),
             headers()
         );
         forward(toForward, toInternal.child());
@@ -250,11 +250,11 @@ public <K, V> void forward(final Record<K, V> record, final String childName) {
             // old API processors wouldn't see the timestamps or headers of upstream
             // new API processors. But then again, from the perspective of those old-API
             // processors, even consulting the timestamp or headers when the record context
-            // is undefined is itself not well-defined. Plus, I don't think we need to worry
+            // is undefined is itself not well defined. Plus, I don't think we need to worry
             // too much about heterogeneous applications, in which the upstream processor is
             // implementing the new API and the downstream one is implementing the old API.
             // So, this seems like a fine compromise for now.
-            if (recordContext != null && (record.timestamp() != recordContext.timestamp() || record.headers() != recordContext.headers())) {
+            if (recordContext != null && (record.timestamp() != timestamp() || record.headers() != headers())) {
                 recordContext = new ProcessorRecordContext(
                     record.timestamp(),
                     recordContext.offset(),
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
@@ -209,13 +209,13 @@ public void process(final Record<KIn, VIn> record) {
             // (instead of `RuntimeException`) to work well with those languages
             final ErrorHandlerContext errorHandlerContext = new DefaultErrorHandlerContext(
                 null, // only required to pass for DeserializationExceptionHandler
-                internalProcessorContext.recordContext().topic(),
-                internalProcessorContext.recordContext().partition(),
-                internalProcessorContext.recordContext().offset(),
-                internalProcessorContext.recordContext().headers(),
+                internalProcessorContext.topic(),
+                internalProcessorContext.partition(),
+                internalProcessorContext.offset(),
+                internalProcessorContext.headers(),
                 internalProcessorContext.currentNode().name(),
                 internalProcessorContext.taskId(),
-                internalProcessorContext.recordContext().timestamp(),
+                internalProcessorContext.timestamp(),
                 internalProcessorContext.recordContext().sourceRawKey(),
                 internalProcessorContext.recordContext().sourceRawValue()
             );
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
@@ -85,9 +85,9 @@ public void process(final Record<KIn, VIn> record) {
         final ProcessorRecordContext contextForExtraction =
             new ProcessorRecordContext(
                 timestamp,
-                context.recordContext().offset(),
-                context.recordContext().partition(),
-                context.recordContext().topic(),
+                context.offset(),
+                context.partition(),
+                context.topic(),
                 record.headers()
             );
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -866,8 +866,8 @@ private void doProcess(final long wallClockTime) {
         final Record<Object, Object> toProcess = new Record<>(
             record.key(),
             record.value(),
-            processorContext.recordContext().timestamp(),
-            processorContext.recordContext().headers()
+            processorContext.timestamp(),
+            processorContext.headers()
         );
         maybeMeasureLatency(() -> currNode.process(toProcess), time, processLatencySensor);
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
@@ -272,12 +272,12 @@ private void putInternal(final Bytes key,
                 key,
                 new LRUCacheEntry(
                     value,
-                    internalContext.recordContext().headers(),
+                    internalContext.headers(),
                     true,
-                    internalContext.recordContext().offset(),
-                    internalContext.recordContext().timestamp(),
-                    internalContext.recordContext().partition(),
-                    internalContext.recordContext().topic(),
+                    internalContext.offset(),
+                    internalContext.timestamp(),
+                    internalContext.partition(),
+                    internalContext.topic(),
                     internalContext.recordContext().sourceRawKey(),
                     internalContext.recordContext().sourceRawValue()
                 )
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java
@@ -135,12 +135,12 @@ public void put(final Windowed<Bytes> key, final byte[] value) {
         final LRUCacheEntry entry =
             new LRUCacheEntry(
                 value,
-                internalContext.recordContext().headers(),
+                internalContext.headers(),
                 true,
-                internalContext.recordContext().offset(),
-                internalContext.recordContext().timestamp(),
-                internalContext.recordContext().partition(),
-                internalContext.recordContext().topic(),
+                internalContext.offset(),
+                internalContext.timestamp(),
+                internalContext.partition(),
+                internalContext.topic(),
                 internalContext.recordContext().sourceRawKey(),
                 internalContext.recordContext().sourceRawValue()
             );
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java
@@ -153,12 +153,12 @@ public synchronized void put(final Bytes key,
         final LRUCacheEntry entry =
             new LRUCacheEntry(
                 value,
-                internalContext.recordContext().headers(),
+                internalContext.headers(),
                 true,
-                internalContext.recordContext().offset(),
-                internalContext.recordContext().timestamp(),
-                internalContext.recordContext().partition(),
-                internalContext.recordContext().topic(),
+                internalContext.offset(),
+                internalContext.timestamp(),
+                internalContext.partition(),
+                internalContext.topic(),
                 internalContext.recordContext().sourceRawKey(),
                 internalContext.recordContext().sourceRawValue()
             );
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStore.java
@@ -52,7 +52,7 @@ private void maybeSetEvictionListener() {
         if (wrapped() instanceof MemoryLRUCache) {
             ((MemoryLRUCache) wrapped()).setWhenEldestRemoved((key, value) -> {
                 // pass null to indicate removal
-                log(key, null, internalContext.recordContext().timestamp());
+                log(key, null, internalContext.timestamp());
             });
         }
     }
@@ -66,7 +66,7 @@ public long approximateNumEntries() {
     public void put(final Bytes key,
                     final byte[] value) {
         wrapped().put(key, value);
-        log(key, value, internalContext.recordContext().timestamp());
+        log(key, value, internalContext.timestamp());
     }
 
     @Override
@@ -75,7 +75,7 @@ public byte[] putIfAbsent(final Bytes key,
         final byte[] previous = wrapped().putIfAbsent(key, value);
         if (previous == null) {
             // then it was absent
-            log(key, value, internalContext.recordContext().timestamp());
+            log(key, value, internalContext.timestamp());
         }
         return previous;
     }
@@ -84,7 +84,7 @@ public byte[] putIfAbsent(final Bytes key,
     public void putAll(final List<KeyValue<Bytes, byte[]>> entries) {
         wrapped().putAll(entries);
         for (final KeyValue<Bytes, byte[]> entry : entries) {
-            log(entry.key, entry.value, internalContext.recordContext().timestamp());
+            log(entry.key, entry.value, internalContext.timestamp());
         }
     }
 
@@ -97,7 +97,7 @@ public <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]> prefixScan(
     @Override
     public byte[] delete(final Bytes key) {
         final byte[] oldValue = wrapped().delete(key);
-        log(key, null, internalContext.recordContext().timestamp());
+        log(key, null, internalContext.timestamp());
         return oldValue;
     }
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStore.java
@@ -32,9 +32,9 @@ public void put(final Bytes key, final byte[] value) {
         // we need to log the full new list and thus call get() on the inner store below
         // if the value is a tombstone, we delete the whole list and thus can save the get call
         if (value == null) {
-            log(key, null, internalContext.recordContext().timestamp());
+            log(key, null, internalContext.timestamp());
         } else {
-            log(key, wrapped().get(key), internalContext.recordContext().timestamp());
+            log(key, wrapped().get(key), internalContext.timestamp());
         }
     }
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingSessionBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingSessionBytesStore.java
@@ -73,13 +73,13 @@ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFindSessions(final Byte
     @Override
     public void remove(final Windowed<Bytes> sessionKey) {
         wrapped().remove(sessionKey);
-        internalContext.logChange(name(), SessionKeySchema.toBinary(sessionKey), null, internalContext.recordContext().timestamp(), wrapped().getPosition());
+        internalContext.logChange(name(), SessionKeySchema.toBinary(sessionKey), null, internalContext.timestamp(), wrapped().getPosition());
     }
 
     @Override
     public void put(final Windowed<Bytes> sessionKey, final byte[] aggregate) {
         wrapped().put(sessionKey, aggregate);
-        internalContext.logChange(name(), SessionKeySchema.toBinary(sessionKey), aggregate, internalContext.recordContext().timestamp(), wrapped().getPosition());
+        internalContext.logChange(name(), SessionKeySchema.toBinary(sessionKey), aggregate, internalContext.timestamp(), wrapped().getPosition());
     }
 
     @Override
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingTimestampedKeyValueBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingTimestampedKeyValueBytesStore.java
@@ -35,7 +35,7 @@ public class ChangeLoggingTimestampedKeyValueBytesStore extends ChangeLoggingKey
     public void put(final Bytes key,
                     final byte[] valueAndTimestamp) {
         wrapped().put(key, valueAndTimestamp);
-        log(key, rawValue(valueAndTimestamp), valueAndTimestamp == null ? internalContext.recordContext().timestamp() : timestamp(valueAndTimestamp));
+        log(key, rawValue(valueAndTimestamp), valueAndTimestamp == null ? internalContext.timestamp() : timestamp(valueAndTimestamp));
     }
 
     @Override
@@ -44,7 +44,7 @@ public byte[] putIfAbsent(final Bytes key,
         final byte[] previous = wrapped().putIfAbsent(key, valueAndTimestamp);
         if (previous == null) {
             // then it was absent
-            log(key, rawValue(valueAndTimestamp), valueAndTimestamp == null ? internalContext.recordContext().timestamp() : timestamp(valueAndTimestamp));
+            log(key, rawValue(valueAndTimestamp), valueAndTimestamp == null ? internalContext.timestamp() : timestamp(valueAndTimestamp));
         }
         return previous;
     }
@@ -54,7 +54,7 @@ public void putAll(final List<KeyValue<Bytes, byte[]>> entries) {
         wrapped().putAll(entries);
         for (final KeyValue<Bytes, byte[]> entry : entries) {
             final byte[] valueAndTimestamp = entry.value;
-            log(entry.key, rawValue(valueAndTimestamp), valueAndTimestamp == null ? internalContext.recordContext().timestamp() : timestamp(valueAndTimestamp));
+            log(entry.key, rawValue(valueAndTimestamp), valueAndTimestamp == null ? internalContext.timestamp() : timestamp(valueAndTimestamp));
         }
     }
 }
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingTimestampedWindowBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingTimestampedWindowBytesStore.java
@@ -36,7 +36,7 @@ void log(final Bytes key,
             name(),
             key,
             rawValue(valueAndTimestamp),
-            valueAndTimestamp != null ? timestamp(valueAndTimestamp) : internalContext.recordContext().timestamp(),
+            valueAndTimestamp != null ? timestamp(valueAndTimestamp) : internalContext.timestamp(),
             wrapped().getPosition()
         );
     }
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingWindowBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingWindowBytesStore.java
@@ -129,7 +129,7 @@ public void put(final Bytes key,
     }
 
     void log(final Bytes key, final byte[] value) {
-        internalContext.logChange(name(), key, value, internalContext.recordContext().timestamp(), wrapped().getPosition());
+        internalContext.logChange(name(), key, value, internalContext.timestamp(), wrapped().getPosition());
     }
 
     private int maybeUpdateSeqnumForDups() {
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
@@ -417,7 +417,7 @@ protected void maybeRecordE2ELatency() {
         // In that case, we _can't_ get the current timestamp, so we don't record anything.
         if (e2eLatencySensor.shouldRecord() && internalContext != null) {
             final long currentTime = time.milliseconds();
-            final long e2eLatency =  currentTime - internalContext.recordContext().timestamp();
+            final long e2eLatency =  currentTime - internalContext.timestamp();
             e2eLatencySensor.record(e2eLatency, currentTime);
         }
     }
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStore.java
@@ -494,7 +494,7 @@ private void maybeRecordE2ELatency() {
         // In that case, we _can't_ get the current timestamp, so we don't record anything.
         if (e2eLatencySensor.shouldRecord() && internalContext != null) {
             final long currentTime = time.milliseconds();
-            final long e2eLatency =  currentTime - internalContext.recordContext().timestamp();
+            final long e2eLatency =  currentTime - internalContext.timestamp();
             e2eLatencySensor.record(e2eLatency, currentTime);
         }
     }
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStore.java
@@ -510,7 +510,7 @@ private void maybeRecordE2ELatency() {
         // In that case, we _can't_ get the current timestamp, so we don't record anything.
         if (e2eLatencySensor.shouldRecord() && internalContext != null) {
             final long currentTime = time.milliseconds();
-            final long e2eLatency =  currentTime - internalContext.recordContext().timestamp();
+            final long e2eLatency =  currentTime - internalContext.timestamp();
             e2eLatencySensor.record(e2eLatency, currentTime);
         }
     }
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingWindowStore.java
@@ -97,13 +97,12 @@ private void enforceWrappedStore(final WindowStore<Bytes, byte[]> underlying) {
         hasIndex = timeOrderedWindowStore.hasIndex();
     }
 
-    @SuppressWarnings("unchecked")
     private RocksDBTimeOrderedWindowStore getWrappedStore(final StateStore wrapped) {
         if (wrapped instanceof RocksDBTimeOrderedWindowStore) {
             return (RocksDBTimeOrderedWindowStore) wrapped;
         }
         if (wrapped instanceof WrappedStateStore) {
-            return getWrappedStore(((WrappedStateStore<?, Bytes, byte[]>) wrapped).wrapped());
+            return getWrappedStore(((WrappedStateStore<?, ?, ?>) wrapped).wrapped());
         }
         return null;
     }
@@ -256,12 +255,12 @@ public synchronized void put(final Bytes key,
         final LRUCacheEntry entry =
             new LRUCacheEntry(
                 value,
-                internalContext.recordContext().headers(),
+                internalContext.headers(),
                 true,
-                internalContext.recordContext().offset(),
-                internalContext.recordContext().timestamp(),
-                internalContext.recordContext().partition(),
-                internalContext.recordContext().topic(),
+                internalContext.offset(),
+                internalContext.timestamp(),
+                internalContext.partition(),
+                internalContext.topic(),
                 internalContext.recordContext().sourceRawKey(),
                 internalContext.recordContext().sourceRawValue()
             );
@@ -278,9 +277,9 @@ public synchronized void put(final Bytes key,
                     new byte[0],
                     new RecordHeaders(),
                     true,
-                    internalContext.recordContext().offset(),
-                    internalContext.recordContext().timestamp(),
-                    internalContext.recordContext().partition(),
+                    internalContext.offset(),
+                    internalContext.timestamp(),
+                    internalContext.partition(),
                     "",
                     internalContext.recordContext().sourceRawKey(),
                     internalContext.recordContext().sourceRawValue()
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorContextImplTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorContextImplTest.java
@@ -593,8 +593,6 @@ public void shouldThrowUnsupportedOperationExceptionOnGetStateStore() {
     @Test
     public void shouldThrowUnsupportedOperationExceptionOnForward() {
         context = getStandbyContext();
-        context.recordContext = mock(ProcessorRecordContext.class);
-
         assertThrows(
             UnsupportedOperationException.class,
             () -> context.forward("key", "value")
@@ -604,8 +602,6 @@ public void shouldThrowUnsupportedOperationExceptionOnForward() {
     @Test
     public void shouldThrowUnsupportedOperationExceptionOnForwardWithTo() {
         context = getStandbyContext();
-        context.recordContext = mock(ProcessorRecordContext.class);
-
         assertThrows(
             UnsupportedOperationException.class,
             () -> context.forward("key", "value", To.child("child-name"))
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
@@ -362,7 +362,7 @@ public ProcessingExceptionHandler.ProcessingHandlerResponse handle(final ErrorHa
             assertEquals(internalProcessorContext.offset(), context.offset());
             assertEquals(internalProcessorContext.currentNode().name(), context.processorNodeId());
             assertEquals(internalProcessorContext.taskId(), context.taskId());
-            assertEquals(internalProcessorContext.recordContext().timestamp(), context.timestamp());
+            assertEquals(internalProcessorContext.timestamp(), context.timestamp());
             assertEquals(internalProcessorContext.recordContext().sourceRawKey(), context.sourceRawKey());
             assertEquals(internalProcessorContext.recordContext().sourceRawValue(), context.sourceRawValue());
 
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingSessionBytesStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingSessionBytesStoreTest.java
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingTimestampedWindowBytesStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingTimestampedWindowBytesStoreTest.java
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingWindowBytesStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingWindowBytesStoreTest.java
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreTest.java
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingPersistentWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingPersistentWindowStoreTest.java
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedWindowStoreTest.java

Original file line number	Diff line number	Diff line change
`@@ -213,7 +213,7 @@ public void close() {`
`213`	`213`	`private ValueAndTimestamp<KeyValue<? extends K1, ? extends V1>> mapValue(final K key, final ValueAndTimestamp<V> valueAndTimestamp) {`
`214`	`214`	`return ValueAndTimestamp.make(`
`215`	`215`	`mapper.apply(key, getValueOrNull(valueAndTimestamp)),`
`216`		`- valueAndTimestamp == null ? context.recordContext().timestamp() : valueAndTimestamp.timestamp()`
	`216`	`+ valueAndTimestamp == null ? context.timestamp() : valueAndTimestamp.timestamp()`
`217`	`217`	`);`
`218`	`218`	`}`
`219`	`219`	`}`
Original file line number	Diff line number	Diff line change
`@@ -84,7 +84,7 @@ public <K, V> void forward(final Record<K, V> record, final String childName) {`
`84`	`84`
`85`	`85`	`@Override`
`86`	`86`	`public <KIn, VIn> void forward(final KIn key, final VIn value) {`
`87`		`- forward(new Record<>(key, value, recordContext().timestamp(), headers()));`
	`87`	`+ forward(new Record<>(key, value, timestamp(), headers()));`
`88`	`88`	`}`
`89`	`89`
`90`	`90`	`/**`