Skip to content

Commit 255c612

Browse files
authored
KAFKA-19668: update upgrade docs (#20484)
Docs change for KAFKA-19668 bug fix. Reviewers: Bill Bejeck <[email protected]>, Lucas Brutschy <[email protected]>
1 parent d3b3aa5 commit 255c612

File tree

3 files changed

+65
-18
lines changed

3 files changed

+65
-18
lines changed

docs/streams/developer-guide/dsl-api.html

Lines changed: 27 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3130,15 +3130,20 @@ <h4>Operations and concepts</h4>
31303130
<code>Processor</code> (provided by a given <code>ProcessorSupplier</code>);
31313131
</li>
31323132
<li><code>KStream#processValues</code>: Process all records in a stream, one record at a time, by applying a
3133-
<code>FixedKeyProcessor</code> (provided by a given <code>FixedKeyProcessorSupplier</code>);
3133+
<code>FixedKeyProcessor</code> (provided by a given <code>FixedKeyProcessorSupplier</code>)
3134+
[<b>CAUTION:</b> If you are deploying a new Kafka Streams application, and you are using the
3135+
"merge repartition topics" optimization, you should enable the fix for
3136+
<a href="https://issues.apache.org/jira/browse/KAFKA-19668">KAFKA-19668</a> to avoid compatibility
3137+
issues for future upgrades to newer versions of Kafka Streams;
3138+
For more details, see the <a href="#transformers-removal-and-migration-to-processors">migration guide</a> below];
31343139
</li>
31353140
<li><code>Processor</code>: A processor of key-value pair records;</li>
31363141
<li><code>ContextualProcessor</code>: An abstract implementation of <code>Processor</code> that manages the
3137-
<code>ProcessorContext</code> instance.
3142+
<code>ProcessorContext</code> instance;
31383143
</li>
31393144
<li><code>FixedKeyProcessor</code>: A processor of key-value pair records where keys are immutable;</li>
31403145
<li><code>ContextualFixedKeyProcessor</code>: An abstract implementation of <code>FixedKeyProcessor</code> that
3141-
manages the <code>FixedKeyProcessorContext</code> instance.
3146+
manages the <code>FixedKeyProcessorContext</code> instance;
31423147
</li>
31433148
<li><code>ProcessorSupplier</code>: A processor supplier that can create one or more <code>Processor</code>
31443149
instances; and</li>
@@ -3456,6 +3461,25 @@ <h3>
34563461
</ul>
34573462
<p>The Processor API now serves as a unified replacement for all these methods. It simplifies the API surface
34583463
while maintaining support for both stateless and stateful operations.</p>
3464+
3465+
<p><b>CAUTION:</b> If you are using <code>KStream.transformValues()</code> and you have the "merge repartition topics"
3466+
optimization enabled, rewriting your program to <code>KStream.processValues()</code> might not be safe due to
3467+
<a href="https://issues.apache.org/jira/browse/KAFKA-19668">KAFKA-19668</a>. For this case, you should not upgrade
3468+
to Kafka Streams 4.0.0 or 4.1.0, but use Kafka Streams 4.0.1 instead, which contains a fix.
3469+
Note, that the fix is not enabled by default for backward compatibility reasons, and you would need to
3470+
enable the fix by setting config <code>__enable.process.processValue.fix__ = true</code> and pass it
3471+
into <code>StreamsBuilder()</code> constructor.</p>
3472+
<pre class="line-numbers"><code class="language-java">final Properties properties = new Properties();
3473+
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, ...);
3474+
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, ...);
3475+
properties.put(TopologyConfig.InternalConfig.ENABLE_PROCESS_PROCESSVALUE_FIX, true);
3476+
3477+
final StreamsBuilder builder = new StreamsBuilder(new TopologyConfig(new StreamsConfig(properties)));</code></pre>
3478+
3479+
<p>It is recommended, that you compare the output of <code>Topology.describe()</code> for the old and new topology,
3480+
to verify if the rewrite to <code>processValues()</code> is correct, and that it does not introduce any incompatibilities.
3481+
You should also test the upgrade in a non-production environment.</p>
3482+
34593483
<h4>Migration Examples</h4>
34603484
<p>To migrate from the deprecated <code>transform</code>, <code>transformValues</code>, <code>flatTransform</code>, and
34613485
<code>flatTransformValues</code> methods to the Processor API (PAPI) in Kafka Streams, let&#39;s resume the

docs/streams/upgrade-guide.html

Lines changed: 24 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -148,7 +148,15 @@ <h3><a id="streams_api_changes_400" href="#streams_api_changes_400">Streams API
148148
<ul>
149149
<li><a href="https://issues.apache.org/jira/browse/KAFKA-12829">Old processor APIs</a></li>
150150
<li><a href="https://issues.apache.org/jira/browse/KAFKA-12823">KStream#through() in both Java and Scala</a></li>
151-
<li><a href="https://issues.apache.org/jira/browse/KAFKA-16339">"transformer" methods and classes in both Java and Scala</a></li>
151+
<li>
152+
<a href="https://issues.apache.org/jira/browse/KAFKA-16339">"transformer" methods and classes in both Java and Scala</a>
153+
<ul>
154+
<li>migrating from <code>KStreams#transformValues()</code> to <code>KStreams.processValues()</code> might not be safe
155+
due to <a href="https://issues.apache.org/jira/browse/KAFKA-19668">KAFKA-19668</a>.
156+
Please refer to the <a href="/{{version}}/documentation/streams/developer-guide/dsl-api.html#transformers-removal-and-migration-to-processors">migration guide</a> for more details.
157+
</li>
158+
</ul>
159+
</li>
152160
<li><a href="https://issues.apache.org/jira/browse/KAFKA-12824">kstream.KStream#branch in both Java and Scala</a></li>
153161
<li><a href="https://issues.apache.org/jira/browse/KAFKA-16332">builder methods for Time/Session/Join/SlidingWindows</a></li>
154162
<li><a href="https://issues.apache.org/jira/browse/KAFKA-12827">KafkaStreams#setUncaughtExceptionHandler()</a></li>
@@ -231,22 +239,22 @@ <h3><a id="streams_api_changes_400" href="#streams_api_changes_400">Streams API
231239
</p>
232240

233241
<p>
234-
You can now configure your topology with a <code>ProcessorWrapper</code>, which allows you to access and optionally wrap/replace
235-
any processor in the topology by injecting an alternative <code>ProcessorSupplier</code> in its place. This can be used to peek
236-
records and access the processor context even for DSL operators, for example to implement a logging or tracing framework, or to
237-
aid in testing or debugging scenarios. You must implement the <code>ProcessorWrapper</code> interface and then pass the class
238-
or class name into the configs via the new <code>StreamsConfig#PROCESSOR_WRAPPER_CLASS_CONFIG</code> config. NOTE: this config is
239-
applied during the topology building phase, and therefore will not take effect unless the config is passed in when creating
240-
the StreamsBuilder (DSL) or Topology(PAPI) objects. You MUST use the StreamsBuilder/Topology constructor overload that
241-
accepts a TopologyConfig parameter for the <code>StreamsConfig#PROCESSOR_WRAPPER_CLASS_CONFIG</code> to be picked up.
242-
See <a href="https://cwiki.apache.org/confluence/x/TZCMEw">KIP-1112</a> for more details.
242+
You can now configure your topology with a <code>ProcessorWrapper</code>, which allows you to access and optionally wrap/replace
243+
any processor in the topology by injecting an alternative <code>ProcessorSupplier</code> in its place. This can be used to peek
244+
records and access the processor context even for DSL operators, for example to implement a logging or tracing framework, or to
245+
aid in testing or debugging scenarios. You must implement the <code>ProcessorWrapper</code> interface and then pass the class
246+
or class name into the configs via the new <code>StreamsConfig#PROCESSOR_WRAPPER_CLASS_CONFIG</code> config. NOTE: this config is
247+
applied during the topology building phase, and therefore will not take effect unless the config is passed in when creating
248+
the StreamsBuilder (DSL) or Topology(PAPI) objects. You MUST use the StreamsBuilder/Topology constructor overload that
249+
accepts a TopologyConfig parameter for the <code>StreamsConfig#PROCESSOR_WRAPPER_CLASS_CONFIG</code> to be picked up.
250+
See <a href="https://cwiki.apache.org/confluence/x/TZCMEw">KIP-1112</a> for more details.
243251
</p>
244252

245253
<p>
246254
Upgraded RocksDB dependency to version 9.7.3 (from 7.9.2). This upgrade incorporates various improvements and optimizations within RocksDB. However, it also introduces some API changes.
247255
The <code>org.rocksdb.AccessHint</code> class, along with its associated methods, has been removed.
248256
Several methods related to compressed block cache configuration in the <code>BlockBasedTableConfig</code> class have been removed, including <code>blockCacheCompressedNumShardBits</code>, <code>blockCacheCompressedSize</code>, and their corresponding setters. These functionalities are now consolidated under the <code>cache</code> option, and developers should configure their compressed block cache using the <code>setCache</code> method instead.
249-
The <code>NO_FILE_CLOSES</code> field has been removed from the <code>org.rocksdb.TickerTypeenum</code> as a result the <code>number-open-files</code> metrics does not work as expected. Metric <code>number-open-files</code> returns constant -1 from now on until it will officially be removed.
257+
The <code>NO_FILE_CLOSES</code> field has been removed from the <code>org.rocksdb.TickerTypeenum</code> as a result the <code>number-open-files</code> metrics does not work as expected. Metric <code>number-open-files</code> returns constant -1 from now on until it will officially be removed.
250258
The <code>org.rocksdb.Options.setLogger()</code> method now accepts a <code>LoggerInterface</code> as a parameter instead of the previous <code>Logger</code>.
251259
Some data types used in RocksDB's Java API have been modified. These changes, along with the removed class, field, and new methods, are primarily relevant to users implementing custom RocksDB configurations.
252260
These changes are expected to be largely transparent to most Kafka Streams users. However, those employing advanced RocksDB customizations within their Streams applications, particularly through the <code>rocksdb.config.setter</code>, are advised to consult the detailed RocksDB 9.7.3 changelog to ensure a smooth transition and adapt their configurations as needed. Specifically, users leveraging the removed <code>AccessHint</code> class, the removed methods from the <code>BlockBasedTableConfig</code> class, the <code>NO_FILE_CLOSES</code> field from <code>TickerType</code>, or relying on the previous signature of <code>setLogger()</code> will need to update their implementations.
@@ -525,6 +533,11 @@ <h3><a id="streams_api_changes_330" href="#streams_api_changes_330">Streams API
525533
<code>FixedKeyProcessorContext</code>, and <code>ContextualFixedKeyProcessor</code> are introduced to guard against
526534
disallowed key modification inside <code>processValues()</code>. Furthermore, <code>ProcessingContext</code> is
527535
added for a better interface hierarchy.
536+
<b>CAUTION:</b> The newly added <code>KStream.processValues()</code> method introduced a regression bug
537+
(<a href="https://issues.apache.org/jira/browse/KAFKA-19668">KAFKA-19668</a>).
538+
If you have "merge repartition topics" optimization enabled, it is not safe to migrate from <code>transformValues()</code>
539+
to <code>processValues()</code> in 3.3.0 release. The bug is only fixed with Kafka Streams 4.0.1, 4.1.1, and 4.2.0.
540+
For more details, please refer to the <a href="/{{version}}/documentation/streams/developer-guide/dsl-api.html#transformers-removal-and-migration-to-processors">migration guide</a>.
528541
</p>
529542

530543
<p>

docs/upgrade.html

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,11 @@ <h5><a id="upgrade_servers_401_notable" href="#upgrade_servers_401_notable">Nota
6262
The filename for rotated <code>state-change.log</code> files has been updated from <code>stage-change.log.[date]</code> to <code>state-change.log.[date]</code> in the log4j2.yaml configuration file.
6363
See <a href="https://issues.apache.org/jira/browse/KAFKA-19576">KAFKA-19576</a> for details.
6464
</li>
65+
<li>
66+
Kafka Streams include a critical fix to upgrade from <code>KStreams#transformValues()</code> (remove with 4.0.0 release)
67+
to <code>KStreams#processValues()</code>.
68+
For more details, see the <a href="/{{version}}/documentation/streams/developer-guide/dsl-api.html#transformers-removal-and-migration-to-processors">migration guide</a>.
69+
</li>
6570
</ul>
6671
<h5><a id="upgrade_servers_400_notable" href="#upgrade_servers_400_notable">Notable changes in 4.0.0</a></h5>
6772
<ul>
@@ -341,14 +346,19 @@ <h5><a id="upgrade_servers_400_notable" href="#upgrade_servers_400_notable">Nota
341346
<li><a id="upgrade_400_notable_kafka_streams" href="#upgrade_400_notable_kafka_streams"><b>Kafka Streams</b></a>
342347
<ul>
343348
<li>
344-
All public API, deprecated in Apache Kafka 3.6 or an earlier release, have been removed, with the exception of <code>JoinWindows.of()</code> and <code>JoinWindows#grace()</code>.
345-
See <a href="https://issues.apache.org/jira/browse/KAFKA-17531">KAFKA-17531</a> for details.
349+
All public API, deprecated in Apache Kafka 3.6 or an earlier release, have been removed, with the exception of <code>JoinWindows.of()</code> and <code>JoinWindows#grace()</code>.
350+
See <a href="https://issues.apache.org/jira/browse/KAFKA-17531">KAFKA-17531</a> for details.
351+
</li>
352+
<li>
353+
The most important changes are highlighted in the <a href="/{{version}}/documentation/streams/upgrade-guide.html#streams_api_changes_400">Kafka Streams upgrade guide</a>.
346354
</li>
347355
<li>
348-
The most important changes are highlighted in the <a href="/{{version}}/documentation/streams/upgrade-guide.html#streams_api_changes_400">Kafka Streams upgrade guide</a>.
356+
For a full list of changes, see <a href="https://issues.apache.org/jira/browse/KAFKA-12822">KAFKA-12822</a>.
349357
</li>
350358
<li>
351-
For a full list of changes, see <a href="https://issues.apache.org/jira/browse/KAFKA-12822">KAFKA-12822</a>.
359+
If you are using <code>KStream#transformValues()</code> which was removed with Apache Kafka 4.0.0 release,
360+
and you need to rewrite your program to use <code>KStreams#processValues()</code> instead,
361+
pay close attention to the <a href="/{{version}}/documentation/streams/developer-guide/dsl-api.html#transformers-removal-and-migration-to-processors">migration guide</a>.
352362
</li>
353363
</ul>
354364
</li>

0 commit comments

Comments
 (0)