Current state: Under Discussion
Discussion thread: here
JIRA: here [Change the link from KAFKA-1 to your own ticket]
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Sometimes users may have a common code module that they wish to apply across all processors of their application, such as a custom logger or debugging element. Right now users have to manually apply this module to every processor in their application, and can only do so for PAPI applications. DSL users are simply out of luck. To make things easier for PAPI users and possible for DSL users, we would like to introduce a mechanism for wrapping processors throughout a Kafka Streams topology.
The primary change here is the introduction of the ProcessorWrapper, but we will also use this as an opportunity to clean up the related TopologyConfig
The following config would be added to the TopologyConfig (NOTE: not StreamsConfig!!) class:
public class TopologyConfig {
public static final String PROCESSOR_WRAPPER_CLASS_CONFIG = "processor.wrapper.class";
private static final String PROCESSOR_WRAPPER_CLASS_DOC = "A processor wrapper class or class name that implements the <code>org.apache.kafka.streams.state.ProcessorWrapper</code> interface";
} |
/**
* Wrapper class that can be used to inject custom wrappers around the processors of their application topology.
* The returned instance MUST wrap the supplied {@code ProcessorSupplier} and the {@code Processor} it supplies.
* Returning a new or completely different instance can have unexpected and undesirable affects.
* <p>
* Note that while this API itself is stable, there is no guarantee to the ordering or type of processor suppliers passed
* in for a given Kafka Streams topology, nor for the names of processors received. T
*/
public interface ProcessorWrapper {
<KIn, VIn, KOut, VOut> ProcessorSupplier<KIn, VIn, KOut, VOut> wrapProcessorSupplier(final String processorName,
final ProcessorSupplier<KIn, VIn, KOut, VOut> processorSupplier);
<KIn, VIn, VOut> FixedKeyProcessorSupplier<KIn, VIn, VOut> wrapFixedKeyProcessorSupplier(final String processorName,
final FixedKeyProcessorSupplier<KIn, VIn, VOut> processorSupplier);
} |
As noted above, we will introduce a new ProcessorWrapper class and its associated config. Unfortunately we have to add this to the TopologyConfig, rather than the Streams config, because we need access to the ProcessorWrapper during the topology construction and by the time the Topology is built and handed in to the KafkaStreams app, it is too late. We would like to do a cleanup of the config handling at some point as well, but will save that for a followup KIP at this time (eg StreamsConfig vs TopologyConfig with their overlapping configs, also KafkaStreams::new vs StreamsBuilder::new vs StreamsBuilder::build vs Topology::new all of which take in various forms of configs)
If a user supplies a ProcessorWrapper, it will be applied immediately as processors are added to the topology, whether via the PAPI or the DSL. A user can decide whether or not to add a layer wrapping the passed-in process supplier, but cannot return null or a different/new ProcessorSupplier entirely. It is expected that the wrapper will be a pure wrapping layer and delegate any calls to the underlying processor for full functionality.
We will write a full integration test using both the PAPI and the DSL (with a processor/transformer) to ensure that the wrapping layer does not interfere with the underlying application
N/A