Skip to content

TopologyBuilder methods return the builder to allow methods to be chained together. #32

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ymatsuda merged 2 commits into from
Sep 4, 2015
Merged

TopologyBuilder methods return the builder to allow methods to be chained together. #32

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fe35dc6
TopologyBuilder methods return the builder to allow methods to be cha…
rhauch Sep 2, 2015
1a1bf8f
Added JavaDoc to the public methods in the TopologyBuilder class.
rhauch Sep 4, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
105 changes: 96 additions & 9 deletions stream/src/main/java/org/apache/kafka/streaming/processor/TopologyBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.streaming.KafkaStreaming;
import org.apache.kafka.streaming.StreamingConfig;
import org.apache.kafka.streaming.processor.internals.ProcessorNode;
import org.apache.kafka.streaming.processor.internals.ProcessorTopology;
import org.apache.kafka.streaming.processor.internals.SinkNode;
Expand All @@ -33,6 +35,15 @@
import java.util.Map;
import java.util.Set;

/**
* A component that is used to build a {@link ProcessorTopology}. A topology contains an acyclic graph of sources, processors,
* and sinks. A {@link SourceNode source} is a node in the graph that consumes one or more Kafka topics and forwards them to
* its child nodes. A {@link Processor processor} is a node in the graph that receives input messages from upstream nodes,
* processes that message, and optionally forwarding new messages to one or all of its children. Finally, a {@link SinkNode sink}
* is a node in the graph that receives messages from upstream nodes and writes them to a Kafka topic. This builder allows you
* to construct an acyclic graph of these nodes, and the builder is then passed into a new {@link KafkaStreaming} instance
* that will then {@link KafkaStreaming#start() begin consuming, processing, and producing messages}.
*/
public class TopologyBuilder {

// list of node factories in a topological order
Expand All @@ -56,6 +67,7 @@ public ProcessorNodeFactory(String name, String[] parents, ProcessorDef definiti
this.definition = definition;
}

@Override
public ProcessorNode build() {
Processor processor = definition.instance();
return new ProcessorNode(name, processor);
Expand All @@ -75,6 +87,7 @@ private SourceNodeFactory(String name, String[] topics, Deserializer keyDeserial
this.valDeserializer = valDeserializer;
}

@Override
public ProcessorNode build() {
return new SourceNode(name, keyDeserializer, valDeserializer);
}
Expand All @@ -94,18 +107,48 @@ private SinkNodeFactory(String name, String[] parents, String topic, Serializer
this.keySerializer = keySerializer;
this.valSerializer = valSerializer;
}
@Override
public ProcessorNode build() {
return new SinkNode(name, topic, keySerializer, valSerializer);
}
}

/**
* Create a new builder.
*/
public TopologyBuilder() {}

public final void addSource(String name, String... topics) {
addSource(name, (Deserializer) null, (Deserializer) null, topics);
/**
* Add a new source that consumes the named topics and forwards the messages to child processor and/or sink nodes.
* The source will use the {@link StreamingConfig#KEY_DESERIALIZER_CLASS_CONFIG default key deserializer} and
* {@link StreamingConfig#VALUE_DESERIALIZER_CLASS_CONFIG default value deserializer} specified in the
* {@link StreamingConfig streaming configuration}.
*
* @param name the unique name of the source used to reference this node when
* {@link #addProcessor(String, ProcessorDef, String...) adding processor children}.
* @param topics the name of one or more Kafka topics that this source is to consume
* @return this builder instance so methods can be chained together; never null
*/
public final TopologyBuilder addSource(String name, String... topics) {
return addSource(name, (Deserializer) null, (Deserializer) null, topics);
}

public final void addSource(String name, Deserializer keyDeserializer, Deserializer valDeserializer, String... topics) {
/**
* Add a new source that consumes the named topics and forwards the messages to child processor and/or sink nodes.
* The sink will use the specified key and value deserializers.
*
* @param name the unique name of the source used to reference this node when
* {@link #addProcessor(String, ProcessorDef, String...) adding processor children}.
* @param keyDeserializer the {@link Deserializer key deserializer} used when consuming messages; may be null if the source
* should use the {@link StreamingConfig#KEY_DESERIALIZER_CLASS_CONFIG default key deserializer} specified in the
* {@link StreamingConfig streaming configuration}
* @param valDeserializer the {@link Deserializer value deserializer} used when consuming messages; may be null if the source
* should use the {@link StreamingConfig#VALUE_DESERIALIZER_CLASS_CONFIG default value deserializer} specified in the
* {@link StreamingConfig streaming configuration}
* @param topics the name of one or more Kafka topics that this source is to consume
* @return this builder instance so methods can be chained together; never null
*/
public final TopologyBuilder addSource(String name, Deserializer keyDeserializer, Deserializer valDeserializer, String... topics) {
if (nodeNames.contains(name))
throw new IllegalArgumentException("Processor " + name + " is already added.");

Expand All @@ -118,14 +161,40 @@ public final void addSource(String name, Deserializer keyDeserializer, Deseriali

nodeNames.add(name);
nodeFactories.add(new SourceNodeFactory(name, topics, keyDeserializer, valDeserializer));
return this;
}

public final void addSink(String name, String topic, String... parentNames) {
addSink(name, topic, (Serializer) null, (Serializer) null, parentNames);
/**
* Add a new sink that forwards messages from upstream parent processor and/or source nodes to the named Kafka topic.
* The sink will use the {@link StreamingConfig#KEY_SERIALIZER_CLASS_CONFIG default key serializer} and
* {@link StreamingConfig#VALUE_SERIALIZER_CLASS_CONFIG default value serializer} specified in the
* {@link StreamingConfig streaming configuration}.
*
* @param name the unique name of the sink
* @param topic the name of the Kafka topic to which this sink should write its messages
* @return this builder instance so methods can be chained together; never null
*/
public final TopologyBuilder addSink(String name, String topic, String... parentNames) {
return addSink(name, topic, (Serializer) null, (Serializer) null, parentNames);
}

public final void addSink(String name, String topic, Serializer keySerializer, Serializer valSerializer, String... parentNames) {

/**
* Add a new sink that forwards messages from upstream parent processor and/or source nodes to the named Kafka topic.
* The sink will use the specified key and value serializers.
*
* @param name the unique name of the sink
* @param topic the name of the Kafka topic to which this sink should write its messages
* @param keySerializer the {@link Serializer key serializer} used when consuming messages; may be null if the sink
* should use the {@link StreamingConfig#KEY_SERIALIZER_CLASS_CONFIG default key serializer} specified in the
* {@link StreamingConfig streaming configuration}
* @param valSerializer the {@link Serializer value serializer} used when consuming messages; may be null if the sink
* should use the {@link StreamingConfig#VALUE_SERIALIZER_CLASS_CONFIG default value serializer} specified in the
* {@link StreamingConfig streaming configuration}
* @param parentNames the name of one or more source or processor nodes whose output message this sink should consume
* and write to its topic
* @return this builder instance so methods can be chained together; never null
*/
public final TopologyBuilder addSink(String name, String topic, Serializer keySerializer, Serializer valSerializer, String... parentNames) {
if (nodeNames.contains(name))
throw new IllegalArgumentException("Processor " + name + " is already added.");

Expand All @@ -142,9 +211,19 @@ public final void addSink(String name, String topic, Serializer keySerializer, S

nodeNames.add(name);
nodeFactories.add(new SinkNodeFactory(name, parentNames, topic, keySerializer, valSerializer));
return this;
}

public final void addProcessor(String name, ProcessorDef definition, String... parentNames) {
/**
* Add a new processor node that receives and processes messages output by one or more parent source or processor node.
* Any new messages output by this processor will be forwarded to its child processor or sink nodes.
* @param name the unique name of the processor node
* @param definition the supplier used to obtain this node's {@link Processor} instance
* @param parentNames the name of one or more source or processor nodes whose output messages this processor should receive
* and process
* @return this builder instance so methods can be chained together; never null
*/
public final TopologyBuilder addProcessor(String name, ProcessorDef definition, String... parentNames) {
if (nodeNames.contains(name))
throw new IllegalArgumentException("Processor " + name + " is already added.");

Expand All @@ -161,10 +240,14 @@ public final void addProcessor(String name, ProcessorDef definition, String... p

nodeNames.add(name);
nodeFactories.add(new ProcessorNodeFactory(name, parentNames, definition));
return this;
}

/**
* Build the topology by creating the processors
* Build the topology. This is typically called automatically when passing this builder into the
* {@link KafkaStreaming#KafkaStreaming(TopologyBuilder, StreamingConfig)} constructor.
*
* @see KafkaStreaming#KafkaStreaming(TopologyBuilder, StreamingConfig)
*/
@SuppressWarnings("unchecked")
public ProcessorTopology build() {
Expand Down Expand Up @@ -206,6 +289,10 @@ public ProcessorTopology build() {
return new ProcessorTopology(processorNodes, topicSourceMap, topicSinkMap);
}

/**
* Get the names of topics that are to be consumed by the source nodes created by this builder.
* @return the unmodifiable set of topic names used by source nodes, which changes as new sources are added; never null
*/
public Set<String> sourceTopics() {
return Collections.unmodifiableSet(sourceTopicNames);
}
Expand Down