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

This reverts commit f13a22a.

Author: mjsax

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()
             );
         }
     }

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()));
     }
 
     /**

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);
         }

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(),

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

@@ -34,6 +34,13 @@ public final class ProcessorContextUtils {
 
     private ProcessorContextUtils() {}
 
+    /**
+     * Should be removed as part of KAFKA-10217
+     */
+    public static StreamsMetricsImpl metricsImpl(final ProcessorContext context) {
+        return (StreamsMetricsImpl) context.metrics();
+    }
+
     /**
      * Should be removed as part of KAFKA-10217
      */
@@ -68,7 +75,7 @@ public static <K, V> InternalProcessorContext<K, V> asInternalProcessorContext(
         final ProcessorContext<K, V> context
     ) {
         if (context instanceof InternalProcessorContext) {
-            return (InternalProcessorContext<K, V>) context;
+            return (InternalProcessorContext) context;
         } else {
             throw new IllegalArgumentException(
                 "This component requires internal features of Kafka Streams and must be disabled for unit tests."

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()
             );

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()
             );
 

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);
 

streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStore.java

@@ -55,7 +55,7 @@ public abstract class AbstractDualSchemaRocksDBSegmentedBytesStore<S extends Seg
     protected final Optional<KeySchema> indexKeySchema;
     private final long retentionPeriod;
 
-    protected InternalProcessorContext<?, ?> internalProcessorContext;
+    protected InternalProcessorContext internalProcessorContext;
     private Sensor expiredRecordSensor;
     protected long observedStreamTime = ConsumerRecord.NO_TIMESTAMP;
     protected boolean consistencyEnabled = false;

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()
                 )