...

Currently, Kafka Streams public API leaks a bunch of internal methods that should not be public. Furthermore, DSL and PAPI abstraction are not completely separated at the moment and TopologyBuilder offers methods that belong to DSL only. Additionally, we want to change the creation of KafkaStreams instances and remove the uncommon constructor pattern.

Public Interfaces

In order to get a clean refactoring, we will deprecate classes 

...

public final class TopologyDescription {
    public final List<Subtopology> subtopologies;
    public final List<GlobalStore> globalStores;
 
    public final class Subtopology {
        public final List<Node> nodes;
    }
 
    public final class GlobalStore {
        public final String name;
        public final String topic;
    }
 
    public interface Node {
        List<Node> getPredecessors();
        List<Node> getSuccessors();
    }
 
    public final class Source implements Node {
        public final String name;
        public final String topic; // can be topic name or pattern (as String)
    }
 
    public final class Processor implements Node {
        public final String name;
        public final List<String> stores;
    }
 
    public final class Sink implements Node {
        public final String name;
        public final String topic;
    }
    
}

...

Changes

...

Refactor the code to remove internal methods from public API.

We will add two new internal classes

org.apache.kafka.streams.KafkaStreams:

public class KafkaStreams {
 
    // deprecating old constructors

    @Deprecated
    public KafkaStreams(final TopologyBuilder builder, // old TopologyBuilder
                        final Properties props);

    @Deprecated
    public KafkaStreams(final TopologyBuilder builder, // old TopologyBuilder
                        final KStreamBuilder config);

    @Deprecated
    public KafkaStreams(final TopologyBuilder builder, // old TopologyBuilder
                        final StreamsConfig config,
                        final KafkaClientSupplier clientSupplier);

    // adding new constructors
 
    public KafkaStreams(final TopologyBuilder builder, // new TopologyBuilder
                        final Properties props);

    public KafkaStreams(final TopologyBuilder builder, // new TopologyBuilder
                        final KStreamBuilder config);

    public KafkaStreams(final TopologyBuilder builder, // new TopologyBuilder
                        final StreamsConfig config,
                        final KafkaClientSupplier clientSupplier);

    public KafkaStreams(final KStreamBuilder builder,
                        final Properties props);


    public KafkaStreams(final KStreamBuilder builder,
                        final KStreamBuilder config);

    public KafkaStreams(final KStreamBuilder builder,
                        final StreamsConfig config,
                        final KafkaClientSupplier clientSupplier);


}

Proposed Changes

Refactor the code to remove internal methods from public API.

We will add two new internal classes

• org.apache.kafka.streams.processor.internals.InternalTopologyBuilder
• org.apache.kafka.processor.internals.InternalTopologyBuilderorg.apache.kafka.streams.kstream.internals.InternalKStreamBuilder 

...

The newly added KStreamBuilder used the new TopologyBuilder as a member (no class hierarchy anymore – using it as member gives a clear separation between PAPI and DSL). 

Because the new {{T

Note: because of backward compatibility, removed DSL specific classes offered by old TopologyBuilder must be offered by InternalTopologyBuilder for now. However, after both deprecated classes got removed, this cleanup can be done (and does not require a KIP anymore, because it's internal refactoring -- we just need to create a JIRA for this). Thus, this KIP falls short of separating PAPI and DSL completely. But it's a necessary first step to do the separation in a backward compatible way (backward compatibility requires a two step approach).

...