Status

Current state: Voting

Discussion thread: here 

JIRA: KAFKA-18026 - Getting issue details... STATUS

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

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

Public Interfaces

The following public config definition would be added to the StreamsConfig  class. Note that the config itself will be defined in both the StreamsConfig and the TopologyConfig

public class StreamsConfig {

    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. Must be passed in to the StreamsBuilder or Topology constructor in order to take effect";
}

/**
 * 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>
 * Can be configured, if desired, by implementing the {@link #configure(Map)} method, which will be invoked
 * with the Streams application configs once the {@code ProcessorWrapper} is instantiated.
 */
public interface ProcessorWrapper extends Configurable {

    @Override
    default void configure(final Map<String, ?> configs) {
        // do nothing
    }

    <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);
}

Proposed Changes

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.

Compatibility, Deprecation, and Migration Plan

• N/A

Test Plan

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

Rejected Alternatives

N/A