Page History

package org.apache.kafka.streams.processor.api;

/**
 * Wrapper class that can be used to inject custom wrappers around the processors of their application topology.
 * The returned instance should wrap the supplied {@code ProcessorSupplier} and the {@code Processor} it supplies
 * to avoid disrupting the regular processing of the application, although this is not required and any processor
 * implementation can be substituted in to replace the original processor entirely (which may be useful for example
 * while testing or debugging an application topology).
 * <p>
 * NOTE: in order to use this feature, you must set the {@link StreamsConfig#PROCESSOR_WRAPPER} config and pass it
 * in as a {@link TopologyConfig} when creating the {@link StreamsBuilder} or {@link Topology} by using the
 * appropriate constructor (ie {@link StreamsBuilder#StreamsBuilder(TopologyConfig)} or {@link Topology#Topology(TopologyConfig)})
 * <p>
 * Can be configured, if desired, by implementing the {@link #configure(Map)} method. This will be invoked when
 * the {@code ProcessorWrapper} is instantiated, and will provide it with the TopologyConfigs that were passed in 
 * to the {@link StreamsBuilder} or {@link Topology} constructor.
 */
public interface ProcessorWrapper extends Configurable {

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

    /**
     * Wrap or replace the provided {@link ProcessorSupplier} and return a {@link WrappedProcessorSupplier}
     * To convert a {@link ProcessorSupplier} instance into a {@link WrappedProcessorSupplier},
     * use the {@link ProcessorWrapper#wrapProcessorSupplier(String, ProcessorSupplier)} method
     */     <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wrapProcessorSupplier(final String processorName,
                                                                                                        final ProcessorSupplier<KIn, VIn, KOut, VOut> processorSupplier);

    /**
     * Wrap or replace the provided {@link FixedKeyProcessorSupplier} and return a {@link WrappedFixedKeyProcessorSupplier}
     * To convert a {@link FixedKeyProcessorSupplier} instance into a {@link WrappedFixedKeyProcessorSupplier},
     * use the {@link ProcessorWrapper#wrapFixedKeyProcessorSupplier(String, FixedKeyProcessorSupplier)} method
     */
     <KIn, VIn,  VOut> WrappedFixedKeyProcessorSupplier<KIn, VIn,  VOut> wrapFixedKeyProcessorSupplier(final String processorName,
                                                                                                      final FixedKeyProcessorSupplier<KIn, VIn, VOut> processorSupplier);

     /**
     * Use to convert a {@link ProcessorSupplier} instance into a {@link WrappedProcessorSupplier}
     */
     static <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wrap(
        final ProcessorSupplier<KIn, VIn, KOut, VOut> processorSupplier
    ) {
        return new WrappedProcessorSupplierImpl<>(processorSupplier);
    }

    /**
     * Use to convert a {@link FixedKeyProcessorSupplier} instance into a {@link WrappedFixedKeyProcessorSupplier}
     */
    static <KIn, VIn,  VOut> WrappedFixedKeyProcessorSupplier<KIn, VIn,  VOut> wrapFixedKeyProcessorSupplier(
        final FixedKeyProcessorSupplier<KIn, VIn, VOut> processorSupplier
    ) {
        return new WrappedFixedKeyProcessorSupplierImpl<>(processorSupplier);
    } }