From 3220797c894010ac5162aee6ae68541749ef3276 Mon Sep 17 00:00:00 2001
From: "Matthias J. Sax" <matthias@confluent.io>
Date: Wed, 26 Nov 2025 20:25:05 -0800
Subject: [PATCH] KAFKA-19688: add new KStreams.process() overlaod

Part of KIP-1128.

We introduced a backward compatiblity issue with 4.0 release,
which we fix with the newly added overload. Cf KAFKA-19668.
---
 docs/streams/developer-guide/dsl-api.html     | 78 +++++++++++--------
 .../apache/kafka/streams/kstream/KStream.java | 53 +++++++++++--
 .../kstream/internals/KStreamImpl.java        | 44 ++++++++++-
 3 files changed, 134 insertions(+), 41 deletions(-)

diff --git a/docs/streams/developer-guide/dsl-api.html b/docs/streams/developer-guide/dsl-api.html
index 2c7c959a637ec..f7261048168ed 100644
--- a/docs/streams/developer-guide/dsl-api.html
+++ b/docs/streams/developer-guide/dsl-api.html
@@ -3127,15 +3127,12 @@ <h3>
                 <h4>Operations and concepts</h4>
                 <ul>
                     <li><code>KStream#process</code>: Process all records in a stream, one record at a time, by applying a
-                        <code>Processor</code> (provided by a given <code>ProcessorSupplier</code>);
+                        <code>Processor</code/<code>FixedKeyProcessor</code> (provided by a given <code>ProcessorSupplier</code> or <code>FixedKeyProcessorSupplier</code>);
                     </li>
-                    <li><code>KStream#processValues</code>: Process all records in a stream, one record at a time, by applying a
-                        <code>FixedKeyProcessor</code> (provided by a given <code>FixedKeyProcessorSupplier</code>)
-                        [<b>CAUTION:</b> If you are deploying a new Kafka Streams application, and you are using the
-                        "merge repartition topics" optimization, you should enable the fix for
-                        <a href="https://issues.apache.org/jira/browse/KAFKA-19668">KAFKA-19668</a> to avoid compatibility
-                        issues for future upgrades to newer versions of Kafka Streams;
-                        For more details, see the <a href="#transformers-removal-and-migration-to-processors">migration guide</a> below];
+                    <li>If you are upgrading from an older version of Kafka Streams, and migrate your code from
+                        <code>[flat]TransformValues()</code> or <code>processValues()</code>, see the
+                        <a href="#transformers-removal-and-migration-to-processors">migration guide</a> below,
+                        to ensure backward compatibility.
                     </li>
                     <li><code>Processor</code>: A processor of key-value pair records;</li>
                     <li><code>ContextualProcessor</code>: An abstract implementation of <code>Processor</code> that manages the
@@ -3166,23 +3163,23 @@ <h4>Examples</h4>
                     <tbody>
                         <tr>
                             <td><a href="#categorizing-logs-by-severity">Categorizing Logs by Severity</a></td>
-                            <td><code>process</code></td>
+                            <td><code>process</code> with <code>Processor</code></td>
                             <td>Stateless</td>
                         </tr>
                         <tr>
                             <td><a href="#replacing-slang-in-text-messages">Replacing Slang in Text Messages</a></td>
-                            <td><code>processValues</code></td>
+                            <td><code>process</code> with <code>FixedKeyProcessor</code></td>
                             <td>Stateless</td>
                         </tr>
                         <tr>
                             <td><a href="#cumulative-discounts-for-a-loyalty-program">Cumulative Discounts for a Loyalty Program</a>
                             </td>
-                            <td><code>process</code></td>
+                            <td><code>process with <code>Processor</code></code></td>
                             <td>Stateful</td>
                         </tr>
                         <tr>
                             <td><a href="#traffic-radar-monitoring-car-count">Traffic Radar Monitoring Car Count</a></td>
-                            <td><code>processValues</code></td>
+                            <td><code>process</code> with <code>FixedKeyProcessor</code></td>
                             <td>Stateful</td>
                         </tr>
                     </tbody>
@@ -3205,7 +3202,7 @@ <h5 id="categorizing-logs-by-severity">Categorizing Logs by Severity</h5>
     private static final String UNKNOWN_LOGS_TOPIC = &quot;unknown-logs-topic&quot;;
     private static final String WARN_LOGS_TOPIC = &quot;warn-logs-topic&quot;;
 
-    public static void categorizeWithProcess(final StreamsBuilder builder) {
+    public static void categorizeWithProcessor(final StreamsBuilder builder) {
         final KStream&lt;String, String&gt; logStream = builder.stream(INPUT_LOGS_TOPIC);
         logStream.process(LogSeverityProcessor::new)
                 .to((key, value, recordContext) -&gt; {
@@ -3267,9 +3264,9 @@ <h5 id="replacing-slang-in-text-messages">Replacing Slang in Text Messages</h5>
     private static final String INPUT_MESSAGES_TOPIC = &quot;input-messages-topic&quot;;
     private static final String OUTPUT_MESSAGES_TOPIC = &quot;output-messages-topic&quot;;
 
-    public static void replaceWithProcessValues(final StreamsBuilder builder) {
+    public static void replaceWithFixedKeyProcessor(final StreamsBuilder builder) {
         KStream&lt;String, String&gt; messageStream = builder.stream(INPUT_MESSAGES_TOPIC);
-        messageStream.processValues(SlangReplacementProcessor::new).to(OUTPUT_MESSAGES_TOPIC);
+        messageStream.process(SlangReplacementProcessor::new).to(OUTPUT_MESSAGES_TOPIC);
     }
 
     private static class SlangReplacementProcessor extends ContextualFixedKeyProcessor&lt;String, String, String&gt; {
@@ -3307,7 +3304,7 @@ <h5 id="cumulative-discounts-for-a-loyalty-program">Cumulative Discounts for a L
     private static final String DISCOUNT_NOTIFICATIONS_TOPIC = &quot;discount-notifications-topic&quot;;
     private static final String PURCHASE_EVENTS_TOPIC = &quot;purchase-events-topic&quot;;
 
-    public static void applyDiscountWithProcess(final StreamsBuilder builder) {
+    public static void applyDiscountWithStatefulProcessor(final StreamsBuilder builder) {
         // Define the state store for tracking cumulative spending
         builder.addStateStore(
                 Stores.keyValueStoreBuilder(
@@ -3375,7 +3372,7 @@ <h5 id="traffic-radar-monitoring-car-count">Traffic Radar Monitoring Car Count</
     private static final String DAILY_COUNT_TOPIC = &quot;price-state-topic&quot;;
     private static final String RADAR_COUNT_TOPIC = &quot;car-radar-topic&quot;;
 
-    public static void countWithProcessValues(final StreamsBuilder builder) {
+    public static void countWithStatefulFixedKeyProcessor(final StreamsBuilder builder) {
         // Define a state store for tracking daily car counts
         builder.addStateStore(
                 Stores.keyValueStoreBuilder(
@@ -3386,7 +3383,7 @@ <h5 id="traffic-radar-monitoring-car-count">Traffic Radar Monitoring Car Count</
         );
         final KStream&lt;Void, String&gt; radarStream = builder.stream(RADAR_COUNT_TOPIC);
         // Apply the FixedKeyProcessor with the state store
-        radarStream.processValues(DailyCarCountProcessor::new, DAILY_COUNT_STORE)
+        radarStream.process(DailyCarCountProcessor::new, DAILY_COUNT_STORE)
                 .to(DAILY_COUNT_TOPIC);
     }
 
@@ -3428,7 +3425,7 @@ <h5 id="traffic-radar-monitoring-car-count">Traffic Radar Monitoring Car Count</
                 
                 <h4>Keynotes</h4>
                 <ul>
-                    <li><strong>Type Safety and Flexibility:</strong> The process and processValues APIs utilize
+                    <li><strong>Type Safety and Flexibility:</strong> The process APIs utilize
                         <code>ProcessorContext</code> and <code>Record</code> or <code>FixedKeyRecord</code> objects for better type
                         safety and flexibility of custom processing logic.
                     </li>
@@ -3462,7 +3459,14 @@ <h3>
                 <p>The Processor API now serves as a unified replacement for all these methods. It simplifies the API surface
                     while maintaining support for both stateless and stateful operations.</p>
 
-                <p><b>CAUTION:</b> If you are using <code>KStream.transformValues()</code> and you have the "merge repartition topics"
+                <p><b>Notes for 4.3 and newer releases:</b>
+                    If you are using <code>KStream.transformValues()</code> or <code>KStream.flatTransformValues()</code>
+                    and migrate to <code>KStream.process(FixedKeyProcessorSupplier)</code> the code is fully backward compatible,
+                    and no further action is needed.
+                    If you are using <code>KStream.processValues()</code> and migrate to <code>KStream.process(FixedKeyProcessorSupplier)</code>
+                    might <b>not</b> be backward compatible; compare the below paragraph for more details.
+                </p>
+                <p><b>CAUTION (for 4.0, 4.1) releases:</b> If you are using <code>KStream.transformValues()</code> and you have the "merge repartition topics"
                     optimization enabled, rewriting your program to <code>KStream.processValues()</code> might not be safe due to
                     <a href="https://issues.apache.org/jira/browse/KAFKA-19668">KAFKA-19668</a>. For this case, you should not upgrade
                     to Kafka Streams 4.0.0 or 4.1.0, but use Kafka Streams 4.0.1 instead, which contains a fix.
@@ -3477,13 +3481,14 @@ <h3>
 final StreamsBuilder builder = new StreamsBuilder(new TopologyConfig(new StreamsConfig(properties)));</code></pre>
 
                 <p>It is recommended, that you compare the output of <code>Topology.describe()</code> for the old and new topology,
-                    to verify if the rewrite to <code>processValues()</code> is correct, and that it does not introduce any incompatibilities.
+                    to verify if the rewrite to <code>processValues()</code> or <code>process(FixedKeyProcessorSupplier)</code> is correct,
+                    and that it does not introduce any incompatibilities.
                     You should also test the upgrade in a non-production environment.</p>
 
                 <h4>Migration Examples</h4>
                 <p>To migrate from the deprecated <code>transform</code>, <code>transformValues</code>, <code>flatTransform</code>, and
                     <code>flatTransformValues</code> methods to the Processor API (PAPI) in Kafka Streams, let&#39;s resume the
-                    previouss examples. The new <code>process</code> and <code>processValues</code> methods enable a more flexible
+                    previous examples. The new <code>process</code> method enable a more flexible
                     and reusable approach by requiring implementations of the <code>Processor</code>
                     or <code>FixedKeyProcessor</code> interfaces.</p>
                 <table>
@@ -3499,26 +3504,26 @@ <h4>Migration Examples</h4>
                         <tr>
                             <td><a href="#categorizing-logs-by-severity-removal">Categorizing Logs by Severity</a></td>
                             <td><code>flatTransform</code></td>
-                            <td><code>process</code></td>
+                            <td><code>process</code> with <code>Processor</code></td>
                             <td>Stateless</td>
                         </tr>
                         <tr>
                             <td><a href="#replacing-slang-in-text-messages-removal">Replacing Slang in Text Messages</a></td>
                             <td><code>flatTransformValues</code></td>
-                            <td><code>processValues</code></td>
+                            <td><code>process</code> with <code>FixedKeyProcessor</code></td>
                             <td>Stateless</td>
                         </tr>
                         <tr>
                             <td><a href="#cumulative-discounts-for-a-loyalty-program-removal">Cumulative Discounts for a Loyalty Program</a>
                             </td>
                             <td><code>transform</code></td>
-                            <td><code>process</code></td>
+                            <td><code>process</code> with <code>Processor</code></td>
                             <td>Stateful</td>
                         </tr>
                         <tr>
                             <td><a href="#traffic-radar-monitoring-car-count-removal">Traffic Radar Monitoring Car Count</a></td>
                             <td><code>transformValues</code></td>
-                            <td><code>processValues</code></td>
+                            <td><code>process</code> with <code>FixedKeyProcessor</code></td>
                             <td>Stateful</td>
                         </tr>
                     </tbody>
@@ -3615,8 +3620,8 @@ <h5 id="categorizing-logs-by-severity-removal">Categorizing Logs by Severity</h5
     }
 }</code></pre>
                 <h5 id="replacing-slang-in-text-messages-removal">Replacing Slang in Text Messages</h5>
-                <p>Below, methods <code>replaceWithFlatTransformValues</code> and <code>replaceWithProcessValues</code> show how you can
-                    migrate from <code>flatTransformValues</code> to <code>processValues</code>.</p>
+                <p>Below, methods <code>replaceWithFlatTransformValues</code> and <code>replaceWithProcess</code> show how you can
+                    migrate from <code>flatTransformValues</code> to <code>process</code>.</p>
                 <pre class="line-numbers"><code class="language-java">public class ReplacingSlangTextInMessagesExample {
     private static final Map&lt;String, String&gt; SLANG_DICTIONARY = Map.of(
             &quot;u&quot;, &quot;you&quot;,
@@ -3632,9 +3637,9 @@ <h5 id="replacing-slang-in-text-messages-removal">Replacing Slang in Text Messag
         messageStream.flatTransformValues(SlangReplacementTransformer::new).to(OUTPUT_MESSAGES_TOPIC);
     }
 
-    public static void replaceWithProcessValues(final StreamsBuilder builder) {
+    public static void replaceWithProcess(final StreamsBuilder builder) {
         KStream&lt;String, String&gt; messageStream = builder.stream(INPUT_MESSAGES_TOPIC);
-        messageStream.processValues(SlangReplacementProcessor::new).to(OUTPUT_MESSAGES_TOPIC);
+        messageStream.process(SlangReplacementProcessor::new).to(OUTPUT_MESSAGES_TOPIC);
     }
 
     private static class SlangReplacementTransformer implements ValueTransformer&lt;String, Iterable&lt;String&gt;&gt; {
@@ -3680,6 +3685,8 @@ <h5 id="replacing-slang-in-text-messages-removal">Replacing Slang in Text Messag
     }
 }</code></pre>
                 <h5 id="cumulative-discounts-for-a-loyalty-program-removal">Cumulative Discounts for a Loyalty Program</h5>
+                <p>Below, methods <code>applyDiscountWithTransform</code> and <code>applyDiscountWithProcess</code> show how you can
+                    migrate from <code>transform</code> to <code>process</code>.</p>
                 <pre class="line-numbers"><code class="language-java">public class CumulativeDiscountsForALoyaltyProgramExample {
     private static final double DISCOUNT_THRESHOLD = 100.0;
     private static final String CUSTOMER_SPENDING_STORE = &quot;customer-spending-store&quot;;
@@ -3798,8 +3805,8 @@ <h5 id="cumulative-discounts-for-a-loyalty-program-removal">Cumulative Discounts
     }
 }</code></pre>
                 <h5 id="traffic-radar-monitoring-car-count-removal">Traffic Radar Monitoring Car Count</h5>
-                <p>Below, methods <code>countWithTransformValues</code> and <code>countWithProcessValues</code> show how you can migrate
-                    from <code>transformValues</code> to <code>processValues</code>.</p>
+                <p>Below, methods <code>countWithTransformValues</code> and <code>countWithProcess</code> show how you can migrate
+                    from <code>transformValues</code> to <code>process</code>.</p>
                 <pre class="line-numbers"><code class="language-java">public class TrafficRadarMonitoringCarCountExample {
     private static final String DAILY_COUNT_STORE = &quot;price-state-store&quot;;
     private static final String DAILY_COUNT_TOPIC = &quot;price-state-topic&quot;;
@@ -3831,7 +3838,7 @@ <h5 id="traffic-radar-monitoring-car-count-removal">Traffic Radar Monitoring Car
         );
         final KStream&lt;Void, String&gt; radarStream = builder.stream(RADAR_COUNT_TOPIC);
         // Apply the FixedKeyProcessor with the state store
-        radarStream.processValues(DailyCarCountProcessor::new, DAILY_COUNT_STORE)
+        radarStream.process(DailyCarCountProcessor::new, DAILY_COUNT_STORE)
                 .to(DAILY_COUNT_TOPIC);
     }
 
@@ -3919,6 +3926,11 @@ <h4>Keynotes</h4>
                     <li><strong>Unified API:</strong> Consolidates multiple methods into a single, versatile API.</li>
                     <li><strong>Future-Proof:</strong> Ensures compatibility with the latest Kafka Streams releases.</li>
                 </ul>
+                <h4>Deprecation of <code>processValues</code> Method</h4>
+                <p>Note that <code>processValues</code> was deprecated in Kafka Streams 4.3.0 release, in favor of a new
+                    overload <code>process(FixedKeyProcessorSupplier)</code> to provide a backward compatible upgrade
+                    for Kafka Streams 3.x for users of <code>[flat]TransformValues</code> methods
+                    (cf. <a href="https://issues.apache.org/jira/browse/KAFKA-19668">KAFKA-19668</a>).</p>
                 <h4>Removal of Old <code>process</code> Method</h4>
                 <p>It is worth mentioning that, in addition to the methods mentioned above, the <code>process</code> method, which
                     integrated the &#39;old&#39; Processor API (i.e., <code>Processor</code> as opposed to the new
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
index acbe1a6b7e251..2410be398344c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
@@ -52,8 +52,7 @@
  * {@link GlobalKTable}, or can be aggregated into a {@link KTable}.
  * A {@link KStream} can also be directly {@link KStream#toTable() converted} into a {@code KTable}.
  * Kafka Streams DSL can be mixed-and-matched with the Processor API (PAPI) (cf. {@link Topology}) via
- * {@link #process(ProcessorSupplier, String...) process(...)} and {@link #processValues(FixedKeyProcessorSupplier,
- * String...) processValues(...)}.
+ * {@link #process(ProcessorSupplier, String...) process(...)}.
  *
  * @param <K> the key type of this stream
  * @param <V> the value type of this stream
@@ -63,7 +62,7 @@ public interface KStream<K, V> {
     /**
      * Create a new {@code KStream} that consists of all records of this stream which satisfy the given predicate.
      * All records that do not satisfy the predicate are dropped.
-     * This is a stateless record-by-record operation (cf. {@link #processValues(FixedKeyProcessorSupplier, String...)}
+     * This is a stateless record-by-record operation (cf. {@link #process(FixedKeyProcessorSupplier, String...)}
      * for stateful record processing or if you need access to the record's timestamp, headers, or other metadata).
      *
      * @param predicate
@@ -86,7 +85,7 @@ public interface KStream<K, V> {
      * Create a new {@code KStream} that consists all records of this stream which do <em>not</em> satisfy the given
      * predicate.
      * All records that <em>do</em> satisfy the predicate are dropped.
-     * This is a stateless record-by-record operation (cf. {@link #processValues(FixedKeyProcessorSupplier, String...)}
+     * This is a stateless record-by-record operation (cf. {@link #process(FixedKeyProcessorSupplier, String...)}
      * for stateful record processing or if you need access to the record's timestamp, headers, or other metadata).
      *
      * @param predicate
@@ -156,7 +155,7 @@ <KOut> KStream<KOut, V> selectKey(final KeyValueMapper<? super K, ? super V, ? e
      * Thus, an input record {@code <K,V>} can be transformed into an output record {@code <K:V'>}.
      * If you need read access to the input record key, use {@link #mapValues(ValueMapperWithKey)}.
      * This is a stateless record-by-record operation (cf.
-     * {@link #processValues(FixedKeyProcessorSupplier, String...)} for stateful value processing or if you need access
+     * {@link #process(FixedKeyProcessorSupplier, String...)} for stateful value processing or if you need access
      * to the record's timestamp, headers, or other metadata).
      *
      * <p>The example below counts the number of token of the value string.
@@ -320,7 +319,7 @@ <KR, VOut> KStream<KR, VOut> flatMap(final KeyValueMapper<? super K, ? super V,
      * (possibly of a different type) for it.
      * Thus, an input record {@code <K,V>} can be transformed into output records {@code <K:V'>, <K:V'>, ...}.
      * If you need read access to the input record key, use {@link #flatMapValues(ValueMapperWithKey)}.
-     * This is a stateless record-by-record operation (cf. {@link #processValues(FixedKeyProcessorSupplier, String...)}
+     * This is a stateless record-by-record operation (cf. {@link #process(FixedKeyProcessorSupplier, String...)}
      * for stateful record processing or if you need access to the record's timestamp, headers, or other metadata).
      *
      * <p>The example below splits input records {@code <null:String>} containing sentences as values into their words.
@@ -1540,7 +1539,7 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final GlobalKTable<Glob
      * If repartitioning is required, a call to {@link #repartition()} should be performed before {@code process()}.
      * At the same time, this method is considered a key changing operation by itself, and might result in an internal
      * data redistribution if a key-based operator (like an aggregation or join) is applied to the result
-     * {@code KStream} (cf. {@link #processValues(FixedKeyProcessorSupplier, String...)}).
+     * {@code KStream} (cf. {@link #process(FixedKeyProcessorSupplier, String...)}).
      *
      * @param processorSupplier
      *        the supplier used to obtain {@link Processor} instances
@@ -1577,6 +1576,41 @@ <KOut, VOut> KStream<KOut, VOut> process(
      * <p>However, because the key cannot be modified, some restrictions apply to a {@link FixedKeyProcessor} compared
      * to a {@link Processor}: for example, forwarding result records from a {@link Punctuator} is not possible.
      */
+    <VOut> KStream<K, VOut> process(
+        final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
+        final String... stateStoreNames
+    );
+
+    /**
+     * See {@link #process(FixedKeyProcessorSupplier, String...)}.
+     *
+     * <p>Takes an additional {@link Named} parameter that is used to name the processor in the topology.
+     */
+    <VOut> KStream<K, VOut> process(
+        final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
+        final Named named,
+        final String... stateStoreNames
+    );
+
+    /**
+     * Process all records in this stream, one record at a time, by applying a {@link FixedKeyProcessor} (provided by
+     * the given {@link FixedKeyProcessorSupplier}) to each input record.
+     * This method is similar to {@link #process(ProcessorSupplier, String...)}, however the key of the input
+     * {@link Record} cannot be modified.
+     *
+     * <p>Because the key cannot be modified, this method is <em>not</em> a key changing operation and preserves data
+     * co-location with respect to the key (cf. {@link #flatMapValues(ValueMapper)}).
+     * Thus, <em>no</em> internal data redistribution is required if a key-based operator (like an aggregation or join)
+     * is applied to the result {@code KStream}.
+     *
+     * <p>However, because the key cannot be modified, some restrictions apply to a {@link FixedKeyProcessor} compared
+     * to a {@link Processor}: for example, forwarding result records from a {@link Punctuator} is not possible.
+     *
+     * @deprecated Since 4.3. Use {@link #process(FixedKeyProcessorSupplier, String...)} instead.
+     * Note, that upgrading from {@code processValues()} to {@code process()} is not always backward compatibly.
+     * Please consult the Kafka Streams upgrade documentation for more details.
+     */
+    @Deprecated
     <VOut> KStream<K, VOut> processValues(
         final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
         final String... stateStoreNames
@@ -1586,7 +1620,12 @@ <VOut> KStream<K, VOut> processValues(
      * See {@link #processValues(FixedKeyProcessorSupplier, String...)}.
      *
      * <p>Takes an additional {@link Named} parameter that is used to name the processor in the topology.
+     *
+     * @deprecated Since 4.3. Use {@link #process(FixedKeyProcessorSupplier, Named, String...)} instead.
+     * Note, that upgrading from {@code processValues()} to {@code process()} is not always backward compatibly.
+     * Please consult the Kafka Streams upgrade documentation for more details.
      */
+    @Deprecated
     <VOut> KStream<K, VOut> processValues(
         final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
         final Named named,
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
index 02342a9de4a75..1db3d46d5fae9 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
@@ -1326,6 +1326,33 @@ public <KOut, VOut> KStream<KOut, VOut> process(
             builder);
     }
 
+    @Override
+    public <VOut> KStream<K, VOut> process(
+        final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
+        final String... stateStoreNames
+    ) {
+        return process(
+            processorSupplier,
+            Named.as(builder.newProcessorName(PROCESSVALUES_NAME)),
+            stateStoreNames
+        );
+    }
+
+    @Override
+    public <VOut> KStream<K, VOut> process(
+        final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
+        final Named named,
+        final String... stateStoreNames
+    ) {
+        return processValuesInternal(
+            true,
+            processorSupplier,
+            named,
+            stateStoreNames
+        );
+    }
+
+    @Deprecated
     @Override
     public <VOut> KStream<K, VOut> processValues(
         final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
@@ -1338,11 +1365,26 @@ public <VOut> KStream<K, VOut> processValues(
         );
     }
 
+    @Deprecated
     @Override
     public <VOut> KStream<K, VOut> processValues(
         final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
         final Named named,
         final String... stateStoreNames
+    ) {
+        return processValuesInternal(
+            builder.processProcessValueFixEnabled(),
+            processorSupplier,
+            named,
+            stateStoreNames
+        );
+    }
+
+    private <VOut> KStream<K, VOut> processValuesInternal(
+        final boolean enableProcessProcessValueFix,
+        final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
+        final Named named,
+        final String... stateStoreNames
     ) {
         ApiUtils.checkSupplier(processorSupplier);
         Objects.requireNonNull(named, "named cannot be null");
@@ -1357,7 +1399,7 @@ public <VOut> KStream<K, VOut> processValues(
             new ProcessorParameters<>(processorSupplier, name),
             stateStoreNames
         );
-        if (builder.processProcessValueFixEnabled()) {
+        if (enableProcessProcessValueFix) {
             processNode.setValueChangingOperation(true);
         }