...
JIRA: KAFKA-3856
Released: 0.1011.30
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
...
org.apache.kafka.streams.processor.TopologyBuilder
org.apache.kafka.streams.kstream.
KStreamBuilder
and add them again ( with new name and different package)
org.apache.kafka.streams.Topology
org.apache.kafka.streams.
KStreamBuilderStreamsTopologyBuilder
Furthermore, we add a new class 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 subclassesSubtopology
GlobalStores
Node
(interface)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 -> StreamsTopologyBuilder
: all methods from TopologyBuilder except addStateStore and addGlobalStore (as KStreamBuilder does not inherit from TopologyBuilder anymore) plus newNameexcept addStateStore and addGlobalStore (which are both added explicitly toStreamsTopologyBuilder
) plus newName- methods highlighted in methods highlighted in
TopologyBuilder
list, are methods that actually belong to DSL abstraction
...
New org.apache.kafka.streams.KStreamBuilderStreamTopologyBuilder
class:
Code Block | ||
---|---|---|
| ||
public class KStreamBuilderStreamsTopologyBuilder { // existing public API from current KStreamBuilder class that will not be changed // small change: adding `synchronized` to all methods to align with TopologyBuilder public KStreamBuilderStreamsTopologyBuilder(); 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> valSerde, final String... topics); public synchronized <K, V> KStream<K, V> stream(final Serde<K> keySerde, final Serde<V> valSerde, final Pattern topicPattern); public synchronized <K, V> KStream<K, V> stream(final AutoOffsetReset offsetReset, final Serde<K> keySerde, final Serde<V> valSerde, final String... topics); public synchronized <K, V> KStream<K, V> stream(final AutoOffsetReset offsetReset, final Serde<K> keySerde, final Serde<V> valSerde, 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> valSerde, final String topic, final String storeName); public synchronized <K, V> KTable<K, V> table(final AutoOffsetReset offsetReset, final Serde<K> keySerde, final Serde<V> valSerde, 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> valSerde, final String topic, final String storeName); public synchronized <K, V> KStream<K, V> merge(final KStream<K, V>... streams); // newly added method public synchronized KStreamBuilder addStateStore(StateStoreSupplier supplier, String... processorNames); public synchronized KStreamBuilder addGlobalStore(final StateStore store, final String sourceName, final Deserializer keyDeserializer, final Deserializer valueDeserializer, final String topic, final String processorName, final ProcessorSupplier stateUpdateSupplier); public synchronized Topology build(); } |
...
Code Block | ||
---|---|---|
| ||
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 to
Proposed Changes
We will add two new internal classes
org.apache.kafka.streams.
...
processor.internals.InternalTopologyBuilder
org.apache.kafka.streams.kstream.internals.InternalStreamsTopologyBuilder
Code Block | ||
---|---|---|
| ||
public class KafkaStreams {
// deprecating old constructors
@Deprecated
public KafkaStreams(final TopologyBuilder builder,
final Properties props);
@Deprecated
public KafkaStreams(final TopologyBuilder builder,
final StreamsConfig config);
@Deprecated
public KafkaStreams(final TopologyBuilder builder,
final StreamsConfig config,
final KafkaClientSupplier clientSupplier);
// adding new constructors
public KafkaStreams(final Topology builder,
final Properties props);
public KafkaStreams(final Topology builder,
final StreamsConfig config);
public KafkaStreams(final Topology 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.streams.kstream.internals.InternalKStreamBuilder
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 Topology
uses InternalTopologyBuilder
as member.
The newly added KStreamBuilder
used the new TopologyBuilder
StreamsTopologyBuilder
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 KStreamBuilder
StreamsTopologyBuilder
does not inherit from new TopologyBuilder
we need to add some more constructors to KafkaStreams
. Topology
we need to add StreamsTopologyBuilder#build()
that returns the actual Topology
to be passed into KafkaStreams
client.
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).
...
- 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 StreamsTopologyBuilder
. All tests using internal API, need to be rewritten to use InternalTopologyBuilder
and InternalKStreamBuilder
InternalStreamsTopologyBuilder
.
Rejected Alternatives
None.
...