Table of Contents |
---|
This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.
Status
Current state: Under Discussion Accepted [VOTE] KIP-120: Cleanup Kafka Streams builder API
Discussion thread: TODO [DISCUSS] KIP-120: Cleanup Kafka Streams builder API
JIRA: KAFKA-3856
Released: 1.0.10.30
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
...
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
org.apache.kafka.streams.processor.TopologyBuilder
org.apache.kafka.streams.kstream.
KStreamBuilder
and add them again ( with same new name but and different package)
org.apache.kafka.streams.TopologyBuilderTopology
org.apache.kafka.streams.
KStreamBuilderStreamsBuilder
Furthermore, we deprecate all constructors of class KafkaStreams
.add a new interface to get a full description the a topology (to compensate for some internal methods that get removed)
org.apache.kafka.streams.TopologyDescription
TopologyDescription
will have public interfaces:Subtopology
GlobalStores
Node
Source
Sink
Processor
The following methods will not be available in the newly added classes:
TopologyBuilder -> Topology
: setApplicationId, connectSourceStoreAndTopic, connectProcessors, addInternalTopic, copartitionSources, nodeGroups, build, buildGlobalStateTopology, globalStateStores, topicGroups, earliestResetTopicPattern, latestResetTopicPattern, stateStoreNamesToSourceTopics, copartitioneGroups, sourceTopicPattern, updateSubscriptionsKStreamBuilder -> StreamsBuilder
: all of above methods from TopologyBuilder (as KStreamBuilder does not inherit from TopologyBuilder anymore) plus except addStateStore and addGlobalStore (which are both added explicitly toStreamsBuilder
) plus newName- methods highlighted in
TopologyBuilder
list, are methods that actually belong to DSL abstraction
...
New org.apache.kafka.streams.TopologyBuilderTopology
class:
Code Block | ||
---|---|---|
| ||
public final class TopologyBuilderTopology { // make class final compared to old TopologyBuilder // existing public API from current TopologyBuilder class that will not be changed public enum AutoOffsetReset { EARLIEST , LATEST } public TopologyBuilderTopology(); public synchronized TopologyBuilderTopology addSource(String name, String... topics); public synchronized TopologyBuilderTopology addSource(String name, Pattern topicPattern); public synchronized TopologyBuilderTopology addSource(AutoOffsetReset offsetReset, String name, String... topics); public synchronized TopologyBuilderTopology addSource(AutoOffsetReset offsetReset, String name, Pattern topicPattern); public synchronized TopologyBuilderTopology addSource(String name, Deserializer keyDeserializer, Deserializer valDeserializervalueDeserializer, String... topics); public synchronized TopologyBuilderTopology addSource(String name, Deserializer keyDeserializer, Deserializer valDeserializervalueDeserializer, Pattern topicPattern); public synchronized TopologyBuilderTopology addSource(AutoOffsetReset offsetReset, String name, Deserializer keyDeserializer, Deserializer valDeserializervalueDeserializer, String... topics); public synchronized TopologyBuilderTopology addSource(AutoOffsetReset offsetReset, String name, Deserializer keyDeserializer, Deserializer valDeserializervalueDeserializer, Pattern topicPattern); public synchronized TopologyBuilderTopology addSink(String name, String topic, String... parentNames); public synchronized TopologyBuilderTopology addSink(String name, String topic, Serializer keySerializer, Serializer valSerializervalueSerializer, String... parentNames); public synchronized TopologyBuilderTopology addSink(String name, String topic, StreamPartitioner partitioner, String... parentNames); public synchronized <K, V> TopologyBuilderTopology addSink(String name, String topic, Serializer<K> keySerializer, Serializer<V> valSerializervalueSerializer, StreamPartitioner<? super K, ? super V> partitioner, String... parentNames); public synchronized TopologyBuilderTopology addProcessor(String name, ProcessorSupplier supplier, String... parentNames); public synchronized TopologyBuilderTopology addStateStore(StateStoreSupplier supplier, String... processorNames); public synchronized TopologyBuilderTopology addGlobalStore(final StateStore store, final String sourceName, final Deserializer keyDeserializer, final Deserializer valueDeserializer, final String topic, final String processorName, final ProcessorSupplier stateUpdateSupplier); public synchronized TopologyBuilderTopology connectProcessorAndStateStores(String processorName, String... stateStoreNames); // newly added method to supportreveal buildertopology pattern // should be used instead of KafkaStreams() constructorstructure // old: describes the current // TopologyBuilder builder = new TopologyBuilder(); // KafkaStreams streams = new KafkaStreams(builder, new StreamsConfig(...)); // new: // TopologyBuilder builder = new TopologyBuilder(); // KafkaStreams streams = builder.build(new StreamsConfig(...)); Topology, ie, TopologyDescription will not be updated if Topology is modified but #describe() must be called again public synchronized KafkaStreamsTopologyDescription builddescribe(); } |
New org.apache.kafka.streams.KStreamBuilderStreamsBuilder
class:
Code Block | ||
---|---|---|
| ||
public final class KStreamBuilderStreamsBuilder { // existing public API from current TopologyBuilderKStreamBuilder class that will not be changed // small change: adding `synchronized` to all methods to align with TopologyBuilder public KStreamBuilderStreamsBuilder(); public synchronized <K, V> KStream<K, V> stream(final String... topics); public synchronized <K, V> KStream<K, V> stream(final Pattern topicPattern); public synchronized <K, V> KStream<K, V> stream(final AutoOffsetReset offsetReset, final String... topics); public synchronized <K, V> KStream<K, V> stream(final AutoOffsetReset offsetReset, final Pattern topicPattern); public synchronized <K, V> KStream<K, V> stream(final Serde<K> keySerde, final Serde<V> valSerdevalueSerde, final String... topics); public synchronized <K, V> KStream<K, V> stream(final Serde<K> keySerde, final Serde<V> valSerdevalueSerde, final Pattern topicPattern); public synchronized <K, V> KStream<K, V> stream(final AutoOffsetReset offsetReset, final Serde<K> keySerde, final Serde<V> valSerdevalueSerde, final String... topics); public synchronized <K, V> KStream<K, V> stream(final AutoOffsetReset offsetReset, final Serde<K> keySerde, final Serde<V> valSerdevalueSerde, final Pattern topicPattern); public synchronized <K, V> KTable<K, V> table(final String topic, final String storeName); public synchronized <K, V> KTable<K, V> table(final AutoOffsetReset offsetReset, final String topic, final String storeName); public synchronized <K, V> KTable<K, V> table(final Serde<K> keySerde, final Serde<V> valSerdevalueSerde, final String topic, final String storeName); public synchronized <K, V> KTable<K, V> table(final AutoOffsetReset offsetReset, final Serde<K> keySerde, final Serde<V> valSerdevalueSerde, final String topic, final String storeName); public synchronized <K, V> GlobalKTable<K, V> globalTable(final String topic, final String storeName); public synchronized <K, V> GlobalKTable<K, V> globalTable(final Serde<K> keySerde, final Serde<V> valSerdevalueSerde, final String topic, final String storeName); public synchronized <K, V> KStream<K, V> merge(final KStream<K, V>... streams); // newly added method to support builder pattern // should be used instead of KafkaStreams() constructor // old: // KStreamBuilder builder = new KStreamBuilder public synchronized KStreamBuilder addStateStore(StateStoreSupplier supplier, String... processorNames); public synchronized KStreamBuilder addGlobalStore(StateStore store, String sourceName, Deserializer keyDeserializer, Deserializer valueDeserializer, String topic, String processorName, ProcessorSupplier stateUpdateSupplier); public synchronized Topology build(); } |
New org.apache.kafka.streams.TopologyDescription
class:
Code Block | ||
---|---|---|
| ||
public interface TopologyDescription { Set<Subtopology> subtopologies(); Set<GlobalStore> globalStores(); inferface Subtopology { int id(); Set<Node> nodes(); //} interface GlobalStore { KafkaStreams streams = newSource KafkaStreams(builder, new StreamsConfig(...)); source(); Processor processor(); } interface Node { // new: String name() // Set<Node> predecessors(); KStreamBuilder builder = newSet<Node> KStreamBuildersuccessors(); } interface Source extends Node { String topics(); // can be comma separated list of topic KafkaStreamsnames streamsor =pattern builder.build(new StreamsConfig(...)); (as String) } interface Processor extends Node { Set<String> stores(); } interface Sink extends Node { public synchronized KafkaStreamsString buildtopic(); } } |
Proposed Changes
...
We will add two new internal classes
org.apache.kafka.streams.processor.internals.InternalTopologyBuilder
org.apache.kafka.streams.kstream.internals.InternalKStreamBuilder InternalStreamsBuilder
that offer the methods remove from current API. Thus, both classes are the actual implementation. Old TopologyBuilder
and KStreamBuilder
are only proxy classes to both classes respectively, for backward compatibility.
The newly added TopologyBuilder
uses Topology
uses InternalTopologyBuilder
as member.
The newly added KStreamBuilder
used the new TopologyBuilder
StreamsBuilder
uses the new Topology
as a member (no class hierarchy anymore – using it as member gives a clear separation between PAPI and DSL).
Because the new StreamsBuilder
does not inherit from new Topology
we need to add StreamsBuilder#build()
that returns the actual Topology
to be passed into KafkaStreams
client.
Note: because of backward compatibility, removing 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 to separate 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).
...
- Because no classes/method will be removed but only deprecated, this change will be fully backward compatible
- We intend to remove all deprecated classes/methods in 0.1011.41, but we can keep them longer on user request
Test Plan
Test Tests need to be rewritten but no new tests are required. All tests for public API need to be updated to use new TopologyBuilder
and KStreamBuilder
, and thus will not call deprecated KafkaStreams
constructor anymoreTopology
and StreamsBuilder
. All tests using internal API, need to be rewritten to use InternalTopologyBuilder
and InternalKStreamBuilder
InternalStreamsBuilder
.
Rejected Alternatives
If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other wayNone.