Processor (provided by a given ProcessorSupplier);
KStream#processValues: Process all records in a stream, one record at a time, by applying a
- FixedKeyProcessor (provided by a given FixedKeyProcessorSupplier);
+ FixedKeyProcessor (provided by a given FixedKeyProcessorSupplier)
+ [CAUTION: If you are deploying a new Kafka Streams application, and you are using the
+ "merge repartition topics" optimization, you should enable the fix for
+ KAFKA-19668 to avoid compatibility
+ issues for future upgrades to newer versions of Kafka Streams;
+ For more details, see the migration guide below];
Processor: A processor of key-value pair records;ContextualProcessor: An abstract implementation of Processor that manages the
- ProcessorContext instance.
+ ProcessorContext instance;
FixedKeyProcessor: A processor of key-value pair records where keys are immutable;ContextualFixedKeyProcessor: An abstract implementation of FixedKeyProcessor that
- manages the FixedKeyProcessorContext instance.
+ manages the FixedKeyProcessorContext instance;
ProcessorSupplier: A processor supplier that can create one or more Processor
instances; andThe 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.
+ +CAUTION: If you are using KStream.transformValues() and you have the "merge repartition topics"
+ optimization enabled, rewriting your program to KStream.processValues() might not be safe due to
+ KAFKA-19668. 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.
+ Note, that the fix is not enabled by default for backward compatibility reasons, and you would need to
+ enable the fix by setting config __enable.process.processValue.fix__ = true and pass it
+ into StreamsBuilder() constructor.
final Properties properties = new Properties();
+properties.put(StreamsConfig.APPLICATION_ID_CONFIG, ...);
+properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, ...);
+properties.put(TopologyConfig.InternalConfig.ENABLE_PROCESS_PROCESSVALUE_FIX, true);
+
+final StreamsBuilder builder = new StreamsBuilder(new TopologyConfig(new StreamsConfig(properties)));It is recommended, that you compare the output of Topology.describe() for the old and new topology,
+ to verify if the rewrite to processValues() is correct, and that it does not introduce any incompatibilities.
+ You should also test the upgrade in a non-production environment.
To migrate from the deprecated transform, transformValues, flatTransform, and
flatTransformValues methods to the Processor API (PAPI) in Kafka Streams, let's resume the
diff --git a/docs/streams/upgrade-guide.html b/docs/streams/upgrade-guide.html
index 2230ffb8aaa..d5c926923e0 100644
--- a/docs/streams/upgrade-guide.html
+++ b/docs/streams/upgrade-guide.html
@@ -262,7 +262,15 @@
KStreams#transformValues() to KStreams.processValues() might not be safe
+ due to KAFKA-19668.
+ Please refer to the migration guide for more details.
+
- You can now configure your topology with a ProcessorWrapper, which allows you to access and optionally wrap/replace
- any processor in the topology by injecting an alternative ProcessorSupplier in its place. This can be used to peek
- records and access the processor context even for DSL operators, for example to implement a logging or tracing framework, or to
- aid in testing or debugging scenarios. You must implement the ProcessorWrapper interface and then pass the class
- or class name into the configs via the new StreamsConfig#PROCESSOR_WRAPPER_CLASS_CONFIG config. NOTE: this config is
- applied during the topology building phase, and therefore will not take effect unless the config is passed in when creating
- the StreamsBuilder (DSL) or Topology(PAPI) objects. You MUST use the StreamsBuilder/Topology constructor overload that
- accepts a TopologyConfig parameter for the StreamsConfig#PROCESSOR_WRAPPER_CLASS_CONFIG to be picked up.
- See KIP-1112 for more details.
+ You can now configure your topology with a ProcessorWrapper, which allows you to access and optionally wrap/replace
+ any processor in the topology by injecting an alternative ProcessorSupplier in its place. This can be used to peek
+ records and access the processor context even for DSL operators, for example to implement a logging or tracing framework, or to
+ aid in testing or debugging scenarios. You must implement the ProcessorWrapper interface and then pass the class
+ or class name into the configs via the new StreamsConfig#PROCESSOR_WRAPPER_CLASS_CONFIG config. NOTE: this config is
+ applied during the topology building phase, and therefore will not take effect unless the config is passed in when creating
+ the StreamsBuilder (DSL) or Topology(PAPI) objects. You MUST use the StreamsBuilder/Topology constructor overload that
+ accepts a TopologyConfig parameter for the StreamsConfig#PROCESSOR_WRAPPER_CLASS_CONFIG to be picked up.
+ See KIP-1112 for more details.
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.
The org.rocksdb.AccessHint class, along with its associated methods, has been removed.
Several methods related to compressed block cache configuration in the BlockBasedTableConfig class have been removed, including blockCacheCompressedNumShardBits, blockCacheCompressedSize, and their corresponding setters. These functionalities are now consolidated under the cache option, and developers should configure their compressed block cache using the setCache method instead.
- The NO_FILE_CLOSES field has been removed from the org.rocksdb.TickerTypeenum as a result the number-open-files metrics does not work as expected. Metric number-open-files returns constant -1 from now on until it will officially be removed.
+ The NO_FILE_CLOSES field has been removed from the org.rocksdb.TickerTypeenum as a result the number-open-files metrics does not work as expected. Metric number-open-files returns constant -1 from now on until it will officially be removed.
The org.rocksdb.Options.setLogger() method now accepts a LoggerInterface as a parameter instead of the previous Logger.
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.
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 rocksdb.config.setter, 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 AccessHint class, the removed methods from the BlockBasedTableConfig class, the NO_FILE_CLOSES field from TickerType, or relying on the previous signature of setLogger() will need to update their implementations.
@@ -639,6 +647,11 @@
FixedKeyProcessorContext, and ContextualFixedKeyProcessor are introduced to guard against
disallowed key modification inside processValues(). Furthermore, ProcessingContext is
added for a better interface hierarchy.
+ CAUTION: The newly added KStream.processValues() method introduced a regression bug
+ (KAFKA-19668).
+ If you have "merge repartition topics" optimization enabled, it is not safe to migrate from transformValues()
+ to processValues() in 3.3.0 release. The bug is only fixed with Kafka Streams 4.0.1, 4.1.1, and 4.2.0.
+ For more details, please refer to the migration guide.
diff --git a/docs/upgrade.html b/docs/upgrade.html
index 257f14e40b0..de8fc394361 100644
--- a/docs/upgrade.html
+++ b/docs/upgrade.html
@@ -130,6 +130,11 @@
The filename for rotated state-change.log files has been updated from stage-change.log.[date] to state-change.log.[date] in the log4j2.yaml configuration file.
See KAFKA-19576 for details.
+
KStreams#transformValues() (remove with 4.0.0 release)
+ to KStreams#processValues().
+ For more details, see the migration guide.
+ KStream#transformValues() which was removed with Apache Kafka 4.0.0 release,
+ and you need to rewrite your program to use KStreams#processValues() instead,
+ pay close attention to the migration guide.