public class TopologyConfigStreamsConfig {

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

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 MUSTshould wrap the supplied {@code ProcessorSupplier} and the {@code Processor} it supplies.
 * Returning a new or completely different instance can have unexpected and undesirable affects 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 Noteorder thatto whileuse 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.  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#asWrapped(ProcessorSupplier)} method
     */  
    <KIn, VIn, KOut, VOut> ProcessorSupplier<KInWrappedProcessorSupplier<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#asWrappedFixedKey(FixedKeyProcessorSupplier)} method
     */
     <KIn, VIn,  VOut> FixedKeyProcessorSupplier<KInWrappedFixedKeyProcessorSupplier<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> asWrapped(
        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> asWrappedFixedKey(
        final FixedKeyProcessorSupplier<KIn, VIn, VOut> processorSupplier
    ) {
        return new WrappedFixedKeyProcessorSupplierImpl<>(processorSupplier);
    } }