From aaf3fc05f8e7cb6b8c79f8471f68faa3a994e1e5 Mon Sep 17 00:00:00 2001
From: "Matthias J. Sax" <matthias@confluent.io>
Date: Mon, 16 Sep 2024 11:49:19 -0700
Subject: [PATCH] MINOR: fix rawtype warning in StandbyTask (#17203)

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
---
 .../streams/processor/internals/StandbyTask.java      | 11 ++---------
 .../processor/internals/StandbyTaskCreator.java       |  4 ++--
 .../streams/processor/internals/StandbyTaskTest.java  |  2 +-
 3 files changed, 5 insertions(+), 12 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
index e6d7001627a..417f754ca2c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
@@ -49,8 +49,7 @@ public class StandbyTask extends AbstractTask implements Task {
     private final Sensor updateSensor;
     private final StreamsMetricsImpl streamsMetrics;
 
-    @SuppressWarnings("rawtypes")
-    protected final InternalProcessorContext processorContext;
+    protected final InternalProcessorContext<?, ?> processorContext;
 
     /**
      * @param id              the ID of this task
@@ -61,7 +60,6 @@ public class StandbyTask extends AbstractTask implements Task {
      * @param stateMgr        the {@link ProcessorStateManager} for this task
      * @param stateDirectory  the {@link StateDirectory} created by the thread
      */
-    @SuppressWarnings("rawtypes")
     StandbyTask(final TaskId id,
                 final Set<TopicPartition> inputPartitions,
                 final ProcessorTopology topology,
@@ -70,7 +68,7 @@ public class StandbyTask extends AbstractTask implements Task {
                 final ProcessorStateManager stateMgr,
                 final StateDirectory stateDirectory,
                 final ThreadCache cache,
-                final InternalProcessorContext processorContext) {
+                final InternalProcessorContext<?, ?> processorContext) {
         super(
             id,
             topology,
@@ -333,11 +331,6 @@ public class StandbyTask extends AbstractTask implements Task {
         throw new IllegalStateException("Attempted to add records to task " + id() + " for invalid input partition " + partition);
     }
 
-    @SuppressWarnings("rawtypes")
-    InternalProcessorContext processorContext() {
-        return processorContext;
-    }
-
     /**
      * Produces a string representation containing useful information about a Task.
      * This is useful in debugging scenarios.
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTaskCreator.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTaskCreator.java
index de0c2fcdf62..668bd832c86 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTaskCreator.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTaskCreator.java
@@ -92,7 +92,7 @@ class StandbyTaskCreator {
                     partitions,
                     stateUpdaterEnabled);
 
-                final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
+                final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
                     taskId,
                     applicationConfig,
                     stateManager,
@@ -147,7 +147,7 @@ class StandbyTaskCreator {
                                   final Set<TopicPartition> inputPartitions,
                                   final ProcessorTopology topology,
                                   final ProcessorStateManager stateManager,
-                                  final InternalProcessorContext<Object, Object> context) {
+                                  final InternalProcessorContext<?, ?> context) {
         final StandbyTask task = new StandbyTask(
             taskId,
             inputPartitions,
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
index 4b8fd63eff0..56f7fb96c43 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
@@ -598,7 +598,7 @@ public class StandbyTaskTest {
             streamsMetrics
         );
 
-        final InternalProcessorContext context = new ProcessorContextImpl(
+        final InternalProcessorContext<?, ?> context = new ProcessorContextImpl(
             taskId,
             config,
             stateManager,