diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/KStream.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/KStream.java index 8df2131624eea..2530b37d0d613 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/KStream.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/KStream.java @@ -19,7 +19,7 @@ import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.Serializer; -import org.apache.kafka.streaming.processor.ProcessorFactory; +import org.apache.kafka.streaming.processor.ProcessorDef; /** * KStream is an abstraction of a stream of key-value pairs. @@ -133,7 +133,7 @@ public interface KStream { /** * Processes all elements in this stream by applying a processor. * - * @param processorFactory the class of ProcessorFactory + * @param processorDef the class of ProcessorDef */ - KStream process(ProcessorFactory processorFactory); + KStream process(ProcessorDef processorDef); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindowDef.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindowDef.java index 4fc8ca41411c8..43c2ea7f5410c 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindowDef.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindowDef.java @@ -68,7 +68,7 @@ public String name() { } @Override - public Window build() { + public Window instance() { return new SlidingWindow(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/Window.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/Window.java index fc9582d21fc20..a968b1782e2a2 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/Window.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/Window.java @@ -33,4 +33,4 @@ public interface Window extends StateStore { Iterator findBefore(K key, long timestamp); void put(K key, V value, long timestamp); -} +} \ No newline at end of file diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/WindowDef.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/WindowDef.java index 26971937eccfe..6ad75528ebf15 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/WindowDef.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/WindowDef.java @@ -21,5 +21,5 @@ public interface WindowDef { String name(); - Window build(); -} + Window instance(); +} \ No newline at end of file diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamBranch.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamBranch.java index e9d6030401c58..e1407b1f44b57 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamBranch.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamBranch.java @@ -18,10 +18,10 @@ package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.streaming.processor.Processor; -import org.apache.kafka.streaming.processor.ProcessorFactory; +import org.apache.kafka.streaming.processor.ProcessorDef; import org.apache.kafka.streaming.kstream.Predicate; -class KStreamBranch implements ProcessorFactory { +class KStreamBranch implements ProcessorDef { private final Predicate[] predicates; @@ -31,7 +31,7 @@ public KStreamBranch(Predicate... predicates) { } @Override - public Processor build() { + public Processor instance() { return new KStreamBranchProcessor(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFilter.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFilter.java index a85e0181a7c76..8d62f7af85190 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFilter.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFilter.java @@ -19,9 +19,9 @@ import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.kstream.Predicate; -import org.apache.kafka.streaming.processor.ProcessorFactory; +import org.apache.kafka.streaming.processor.ProcessorDef; -class KStreamFilter implements ProcessorFactory { +class KStreamFilter implements ProcessorDef { private final Predicate predicate; private final boolean filterOut; @@ -32,7 +32,7 @@ public KStreamFilter(Predicate predicate, boolean filterOut) { } @Override - public Processor build() { + public Processor instance() { return new KStreamFilterProcessor(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMap.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMap.java index bc834a8f66ae8..b9a7f5dcd8285 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMap.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMap.java @@ -20,9 +20,9 @@ import org.apache.kafka.streaming.kstream.KeyValue; import org.apache.kafka.streaming.kstream.KeyValueFlatMap; import org.apache.kafka.streaming.processor.Processor; -import org.apache.kafka.streaming.processor.ProcessorFactory; +import org.apache.kafka.streaming.processor.ProcessorDef; -class KStreamFlatMap implements ProcessorFactory { +class KStreamFlatMap implements ProcessorDef { private final KeyValueFlatMap mapper; @@ -31,7 +31,7 @@ class KStreamFlatMap implements ProcessorFactory { } @Override - public Processor build() { + public Processor instance() { return new KStreamFlatMapProcessor(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValues.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValues.java index 5e18f188ffa08..241970c4a5323 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValues.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValues.java @@ -19,9 +19,9 @@ import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.kstream.ValueMapper; -import org.apache.kafka.streaming.processor.ProcessorFactory; +import org.apache.kafka.streaming.processor.ProcessorDef; -class KStreamFlatMapValues implements ProcessorFactory { +class KStreamFlatMapValues implements ProcessorDef { private final ValueMapper> mapper; @@ -31,7 +31,7 @@ class KStreamFlatMapValues implements ProcessorFactory { } @Override - public Processor build() { + public Processor instance() { return new KStreamFlatMapValuesProcessor(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamImpl.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamImpl.java index e317338b134e6..4cf99cf6ed842 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamImpl.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamImpl.java @@ -19,9 +19,9 @@ import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.Serializer; -import org.apache.kafka.streaming.kstream.KeyValueFlatMap; -import org.apache.kafka.streaming.processor.ProcessorFactory; +import org.apache.kafka.streaming.processor.ProcessorDef; import org.apache.kafka.streaming.processor.TopologyBuilder; +import org.apache.kafka.streaming.kstream.KeyValueFlatMap; import org.apache.kafka.streaming.kstream.KStreamWindowed; import org.apache.kafka.streaming.kstream.KeyValueMapper; import org.apache.kafka.streaming.kstream.Predicate; @@ -130,12 +130,12 @@ public KStream flatMapValues(ValueMapper> } @Override - public KStreamWindowed with(WindowDef windowDef) { + public KStreamWindowed with(WindowDef window) { String name = WINDOWED_NAME + INDEX.getAndIncrement(); - topology.addProcessor(name, new KStreamWindow<>(windowDef), this.name); + topology.addProcessor(name, new KStreamWindow<>(window), this.name); - return new KStreamWindowedImpl<>(topology, name, windowDef); + return new KStreamWindowedImpl<>(topology, name, window); } @Override @@ -185,10 +185,10 @@ public void sendTo(String topic, Serializer keySerializer, Serializer valS @SuppressWarnings("unchecked") @Override - public KStream process(final ProcessorFactory processorFactory) { + public KStream process(final ProcessorDef processorDef) { String name = PROCESSOR_NAME + INDEX.getAndIncrement(); - topology.addProcessor(name, processorFactory, this.name); + topology.addProcessor(name, processorDef, this.name); return new KStreamImpl<>(topology, name); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamJoin.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamJoin.java index a8452d5ff0b64..2c0020b12afb8 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamJoin.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamJoin.java @@ -21,11 +21,11 @@ import org.apache.kafka.streaming.processor.ProcessorContext; import org.apache.kafka.streaming.kstream.ValueJoiner; import org.apache.kafka.streaming.kstream.Window; -import org.apache.kafka.streaming.processor.ProcessorFactory; +import org.apache.kafka.streaming.processor.ProcessorDef; import java.util.Iterator; -class KStreamJoin implements ProcessorFactory { +class KStreamJoin implements ProcessorDef { private static abstract class Finder { abstract Iterator find(K key, long timestamp); @@ -37,9 +37,9 @@ private static abstract class Finder { private final boolean prior; private Processor processorForOtherStream = null; - public final ProcessorFactory processorFactoryForOtherStream = new ProcessorFactory() { + public final ProcessorDef processorDefForOtherStream = new ProcessorDef() { @Override - public Processor build() { + public Processor instance() { return processorForOtherStream; } }; @@ -52,7 +52,7 @@ public Processor build() { } @Override - public Processor build() { + public Processor instance() { return new KStreamJoinProcessor(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamMap.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamMap.java index 83fb4c11c99f6..2400e03e73ad8 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamMap.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamMap.java @@ -20,9 +20,9 @@ import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.kstream.KeyValue; import org.apache.kafka.streaming.kstream.KeyValueMapper; -import org.apache.kafka.streaming.processor.ProcessorFactory; +import org.apache.kafka.streaming.processor.ProcessorDef; -class KStreamMap implements ProcessorFactory { +class KStreamMap implements ProcessorDef { private final KeyValueMapper mapper; @@ -31,7 +31,7 @@ public KStreamMap(KeyValueMapper mapper) { } @Override - public Processor build() { + public Processor instance() { return new KStreamMapProcessor(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValues.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValues.java index 667c04386c381..dc8267a4fa91d 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValues.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValues.java @@ -19,9 +19,9 @@ import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.kstream.ValueMapper; -import org.apache.kafka.streaming.processor.ProcessorFactory; +import org.apache.kafka.streaming.processor.ProcessorDef; -class KStreamMapValues implements ProcessorFactory { +class KStreamMapValues implements ProcessorDef { private final ValueMapper mapper; @@ -30,7 +30,7 @@ public KStreamMapValues(ValueMapper mapper) { } @Override - public Processor build() { + public Processor instance() { return new KStreamMapProcessor(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamPassThrough.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamPassThrough.java index 98528f72c7c1b..89c3550998b12 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamPassThrough.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamPassThrough.java @@ -18,12 +18,12 @@ package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.streaming.processor.Processor; -import org.apache.kafka.streaming.processor.ProcessorFactory; +import org.apache.kafka.streaming.processor.ProcessorDef; -class KStreamPassThrough implements ProcessorFactory { +class KStreamPassThrough implements ProcessorDef { @Override - public Processor build() { + public Processor instance() { return new KStreamPassThroughProcessor(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindow.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindow.java index f357613d94ce4..980fb4e08e350 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindow.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindow.java @@ -19,11 +19,11 @@ import org.apache.kafka.streaming.kstream.Window; import org.apache.kafka.streaming.processor.Processor; -import org.apache.kafka.streaming.processor.ProcessorFactory; +import org.apache.kafka.streaming.processor.ProcessorDef; import org.apache.kafka.streaming.processor.ProcessorContext; import org.apache.kafka.streaming.kstream.WindowDef; -public class KStreamWindow implements ProcessorFactory { +public class KStreamWindow implements ProcessorDef { private final WindowDef windowDef; @@ -36,7 +36,7 @@ public WindowDef window() { } @Override - public Processor build() { + public Processor instance() { return new KStreamWindowProcessor(); } @@ -47,7 +47,7 @@ private class KStreamWindowProcessor extends KStreamProcessor { @Override public void init(ProcessorContext context) { super.init(context); - this.window = windowDef.build(); + this.window = windowDef.instance(); this.window.init(context); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedImpl.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedImpl.java index e269c3ad40bcc..62351a4e3607f 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedImpl.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedImpl.java @@ -37,7 +37,7 @@ private KStream join(KStreamWindowed other, boolean prior String joinName = JOIN_NAME + INDEX.getAndIncrement(); String joinOtherName = JOINOTHER_NAME + INDEX.getAndIncrement(); topology.addProcessor(joinName, join, this.name); - topology.addProcessor(joinOtherName, join.processorFactoryForOtherStream, ((KStreamImpl) other).name); + topology.addProcessor(joinOtherName, join.processorDefForOtherStream, ((KStreamImpl) other).name); return new KStreamImpl<>(topology, joinName); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorFactory.java b/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorDef.java similarity index 93% rename from stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorFactory.java rename to stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorDef.java index 1d1c4816d8cce..382c95733ebcd 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorFactory.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorDef.java @@ -17,7 +17,7 @@ package org.apache.kafka.streaming.processor; -public interface ProcessorFactory { +public interface ProcessorDef { - Processor build(); + Processor instance(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/TopologyBuilder.java b/stream/src/main/java/org/apache/kafka/streaming/processor/TopologyBuilder.java index 894e0e37f0244..596a4c6c49885 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/TopologyBuilder.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/TopologyBuilder.java @@ -47,16 +47,16 @@ private interface NodeFactory { private class ProcessorNodeFactory implements NodeFactory { public final String[] parents; private final String name; - private final ProcessorFactory factory; + private final ProcessorDef definition; - public ProcessorNodeFactory(String name, String[] parents, ProcessorFactory factory) { + public ProcessorNodeFactory(String name, String[] parents, ProcessorDef definition) { this.name = name; this.parents = parents.clone(); - this.factory = factory; + this.definition = definition; } public ProcessorNode build() { - Processor processor = factory.build(); + Processor processor = definition.instance(); return new ProcessorNode(name, processor); } } @@ -134,7 +134,7 @@ public final void addSink(String name, String topic, Serializer keySerializer, S nodeFactories.add(new SinkNodeFactory(name, parentNames, topic, keySerializer, valSerializer)); } - public final void addProcessor(String name, ProcessorFactory factory, String... parentNames) { + public final void addProcessor(String name, ProcessorDef definition, String... parentNames) { if (nodeNames.contains(name)) throw new IllegalArgumentException("Processor " + name + " is already added."); @@ -150,7 +150,7 @@ public final void addProcessor(String name, ProcessorFactory factory, String... } nodeNames.add(name); - nodeFactories.add(new ProcessorNodeFactory(name, parentNames, factory)); + nodeFactories.add(new ProcessorNodeFactory(name, parentNames, definition)); } /** @@ -186,7 +186,7 @@ public ProcessorTopology build() { processorMap.get(parent).addChild(node); } } else { - throw new IllegalStateException("unknown factory class: " + factory.getClass().getName()); + throw new IllegalStateException("unknown definition class: " + factory.getClass().getName()); } } } catch (Exception e) { diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/PartitionGroup.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/PartitionGroup.java index a473c4eff0891..2b708a3d933d2 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/PartitionGroup.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/PartitionGroup.java @@ -22,13 +22,13 @@ import org.apache.kafka.common.serialization.Deserializer; import java.util.Comparator; -import java.util.HashMap; import java.util.Map; import java.util.PriorityQueue; import java.util.Set; /** - * A PartitionGroup is composed from a set of partitions. + * A PartitionGroup is composed from a set of partitions. It also maintains the timestamp of this + * group, hence the associated task as the min timestamp across all partitions in the group. */ public class PartitionGroup { @@ -60,13 +60,14 @@ public int compare(RecordQueue queue1, RecordQueue queue2) { /** * Get one record from the specified partition queue + * + * @return StampedRecord */ public StampedRecord getRecord(RecordQueue queue) { // get the first record from this queue. StampedRecord record = queue.poll(); // update the partition's timestamp and re-order it against other partitions. - queuesByTime.remove(queue); if (queue.size() > 0) { @@ -80,19 +81,16 @@ public StampedRecord getRecord(RecordQueue queue) { /** * Get the next partition queue that has the lowest timestamp to process + * + * @return RecordQueue */ public RecordQueue nextQueue() { // get the partition with the lowest timestamp - RecordQueue recordQueue = queuesByTime.peek(); - - if (recordQueue == null) - throw new KafkaException("No records have ever been added to this partition group yet."); - - return recordQueue; + return queuesByTime.peek(); } /** - * Put a timestamped record associated into its corresponding partition's queues. + * Put a timestamped record associated into its corresponding partition's queues */ public void putRecord(StampedRecord record, TopicPartition partition) { if (record.partition() != partition.partition() || !record.topic().equals(partition.topic())) @@ -143,7 +141,7 @@ public Set partitions() { */ public long timestamp() { if (queuesByTime.isEmpty()) { - return -1L; + return TimestampTracker.NOT_KNOWN; } else { return queuesByTime.peek().timestamp(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorContextImpl.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorContextImpl.java index fcff4b0c1b9b8..03df3e3cf5a57 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorContextImpl.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorContextImpl.java @@ -89,6 +89,10 @@ public ProcessorContextImpl(int id, this.initialized = false; } + public ProcessorStateManager stateManager() { + return this.stateMgr; + } + public RecordCollector recordCollector() { return this.collector; } @@ -213,14 +217,18 @@ public void forward(K key, V value) { @Override @SuppressWarnings("unchecked") public void forward(K key, V value, int childIndex) { +<<<<<<< HEAD ProcessorNode childNode = (ProcessorNode) task.node().children().get(childIndex); +======= + ProcessorNode childNode = (ProcessorNode) task.node().children().get(childIndex); +>>>>>>> kstream refactored task.node(childNode); childNode.process(key, value); } @Override public void commit() { - task.commitOffset(); + task.needCommit(); } @Override diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorStateManager.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorStateManager.java index c659e0459d2e2..810966d83d8d9 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorStateManager.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorStateManager.java @@ -63,10 +63,6 @@ public File baseDir() { return this.baseDir; } - public Consumer restoreConsumer() { - return this.restoreConsumer; - } - public void register(StateStore store, RestoreFunc restoreFunc) { if (store.name().equals(CHECKPOINT_FILE_NAME)) throw new IllegalArgumentException("Illegal store name: " + CHECKPOINT_FILE_NAME); diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/RecordQueue.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/RecordQueue.java index 93e9420e5bd57..b628c1688e785 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/RecordQueue.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/RecordQueue.java @@ -31,22 +31,24 @@ public class RecordQueue { private final SourceNode source; private final TopicPartition partition; - private final ArrayDeque fifoQueue = new ArrayDeque<>(); - private final TimestampTracker> timeTracker = new MinTimestampTracker<>(); + private final ArrayDeque fifoQueue; + private final TimestampTracker> timeTracker; private long partitionTime = TimestampTracker.NOT_KNOWN; - /** - * Creates a new instance of RecordQueue - * - * @param partition partition - * @param source source node - */ public RecordQueue(TopicPartition partition, SourceNode source) { this.partition = partition; this.source = source; + + this.fifoQueue = new ArrayDeque<>(); + this.timeTracker = new MinTimestampTracker<>(); } + /** + * Returns the corresponding source node in the topology + * + * @return SourceNode + */ public SourceNode source() { return source; } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StampedRecord.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StampedRecord.java index e294c365f6346..bce29c2d954b0 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StampedRecord.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StampedRecord.java @@ -19,7 +19,6 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; -// TODO: making this class exposed to user in the lower-level Processor public class StampedRecord extends Stamped> { public StampedRecord(ConsumerRecord record, long timestamp) { diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamTask.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamTask.java index 751f504c8f778..7f48c1864c46b 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamTask.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamTask.java @@ -17,16 +17,21 @@ package org.apache.kafka.streaming.processor.internals; +import org.apache.kafka.clients.consumer.CommitType; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.serialization.Deserializer; -import org.apache.kafka.clients.consumer.Consumer; -import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.streaming.StreamingConfig; import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.ProcessorContext; import org.apache.kafka.streaming.processor.TimestampExtractor; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.Collection; @@ -40,6 +45,8 @@ */ public class StreamTask { + private static final Logger log = LoggerFactory.getLogger(StreamTask.class); + private final int id; private final int maxBufferedSize; @@ -50,8 +57,10 @@ public class StreamTask { private final TimestampExtractor timestampExtractor; private final Map consumedOffsets; + private final RecordCollector recordCollector; private boolean commitRequested = false; + private boolean commitOffsetNeeded = false; private StampedRecord currRecord = null; private ProcessorNode currNode = null; @@ -60,15 +69,17 @@ public class StreamTask { * * @param id the ID of this task * @param consumer the instance of {@link Consumer} - * @param topology the instance of {@link ProcessorTopology} + * @param producer the instance of {@link Producer} * @param partitions the collection of assigned {@link TopicPartition} + * @param topology the instance of {@link ProcessorTopology} * @param config the {@link StreamingConfig} specified by the user */ + @SuppressWarnings("unchecked") public StreamTask(int id, - Consumer consumer, - ProcessorTopology topology, + Consumer consumer, + Producer producer, Collection partitions, - RecordCollector collector, + ProcessorTopology topology, StreamingConfig config) { this.id = id; @@ -89,17 +100,22 @@ public StreamTask(int id, this.partitionGroup = new PartitionGroup(partitionQueues); + // initialize the consumed and produced offset cache + this.consumedOffsets = new HashMap<>(); + + // create the record recordCollector that maintains the produced offsets + this.recordCollector = new RecordCollector(producer, + (Serializer) config.getConfiguredInstance(StreamingConfig.KEY_DESERIALIZER_CLASS_CONFIG, Serializer.class), + (Serializer) config.getConfiguredInstance(StreamingConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Serializer.class)); + // initialize the topology with its own context try { - this.processorContext = new ProcessorContextImpl(id, this, config, collector, new Metrics()); + this.processorContext = new ProcessorContextImpl(id, this, config, recordCollector, new Metrics()); } catch (IOException e) { throw new KafkaException("Error while creating the state manager in processor context."); } topology.init(this.processorContext); - - // initialize the consumed offset cache - this.consumedOffsets = new HashMap<>(); } public int id() { @@ -136,49 +152,43 @@ public void addRecords(TopicPartition partition, Iterator 0) { - readyForNextExecution = true; + commit(); } // if after processing this record, its partition queue's buffered size has been @@ -192,7 +202,7 @@ public boolean process() { long timestamp = partitionGroup.timestamp(); punctuationQueue.mayPunctuate(timestamp); - return readyForNextExecution; + return partitionGroup.numbuffered(); } } @@ -209,18 +219,46 @@ public void node(ProcessorNode node) { } /** - * Request committing the current record's offset + * Commit the current task state */ - public void commitOffset() { + public void commit() { + // 1) flush local state + ((ProcessorContextImpl) processorContext).stateManager().flush(); + + // 2) commit consumed offsets if it is dirty already + if (commitOffsetNeeded) { + consumer.commit(consumedOffsets, CommitType.SYNC); + commitOffsetNeeded = false; + } + + // 3) flush produced records in the downstream + // TODO: this will actually block on all produced records across the tasks + recordCollector.flush(); + } + + /** + * Request committing the current task's state + */ + public void needCommit() { this.commitRequested = true; } + /** + * Schedules a punctuation for the processor + * + * @param processor the processor requesting scheduler + * @param interval the interval in milliseconds + */ + public void schedule(Processor processor, long interval) { + punctuationQueue.schedule(new PunctuationSchedule(processor, interval)); + } + public void close() { this.partitionGroup.close(); this.consumedOffsets.clear(); } - protected RecordQueue createRecordQueue(TopicPartition partition, SourceNode source) { + private RecordQueue createRecordQueue(TopicPartition partition, SourceNode source) { return new RecordQueue(partition, source); } } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamThread.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamThread.java index 93059746ff999..5807fbf9d52a8 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamThread.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamThread.java @@ -35,7 +35,6 @@ import org.apache.kafka.common.metrics.stats.Rate; import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.ByteArraySerializer; -import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.common.utils.SystemTime; import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; @@ -53,24 +52,24 @@ public class StreamThread extends Thread { private static final Logger log = LoggerFactory.getLogger(StreamThread.class); - private final Consumer consumer; + private volatile boolean running; + private final TopologyBuilder builder; - private final RecordCollector collector; - private final Map tasks = new HashMap<>(); - private final Metrics metrics; + private final Producer producer; + private final Consumer consumer; + private final Map tasks; private final Time time; - private final StreamingConfig config; private final File stateDir; private final long pollTimeMs; + private final long cleanTimeMs; private final long commitTimeMs; - private final long stateCleanupDelayMs; private final long totalRecordsToProcess; - private final KafkaStreamingMetrics streamingMetrics; + private final KafkaStreamingMetrics metrics; + private final StreamingConfig config; - private volatile boolean running; + private long lastClean; private long lastCommit; - private long nextStateCleaning; private long recordsProcessed; protected final ConsumerRebalanceCallback rebalanceCallback = new ConsumerRebalanceCallback() { @@ -93,34 +92,34 @@ public StreamThread(TopologyBuilder builder, StreamingConfig config) throws Exce this.config = config; this.builder = builder; - this.streamingMetrics = new KafkaStreamingMetrics(); - // create the producer and consumer clients - Producer producer = new KafkaProducer<>(config.getProducerProperties(), + this.producer = new KafkaProducer<>(config.getProducerProperties(), new ByteArraySerializer(), new ByteArraySerializer()); - this.collector = new RecordCollector(producer, - (Serializer) config.getConfiguredInstance(StreamingConfig.KEY_DESERIALIZER_CLASS_CONFIG, Serializer.class), - (Serializer) config.getConfiguredInstance(StreamingConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Serializer.class)); - consumer = new KafkaConsumer<>(config.getConsumerProperties(), + this.consumer = new KafkaConsumer<>(config.getConsumerProperties(), rebalanceCallback, new ByteArrayDeserializer(), new ByteArrayDeserializer()); + // initialize the task list + this.tasks = new HashMap<>(); + + // read in task specific config values this.stateDir = new File(this.config.getString(StreamingConfig.STATE_DIR_CONFIG)); this.pollTimeMs = config.getLong(StreamingConfig.POLL_MS_CONFIG); this.commitTimeMs = config.getLong(StreamingConfig.COMMIT_INTERVAL_MS_CONFIG); - this.stateCleanupDelayMs = config.getLong(StreamingConfig.STATE_CLEANUP_DELAY_MS_CONFIG); + this.cleanTimeMs = config.getLong(StreamingConfig.STATE_CLEANUP_DELAY_MS_CONFIG); this.totalRecordsToProcess = config.getLong(StreamingConfig.TOTAL_RECORDS_TO_PROCESS); - this.running = true; + this.lastClean = 0; this.lastCommit = 0; - this.nextStateCleaning = Long.MAX_VALUE; this.recordsProcessed = 0; this.time = new SystemTime(); - this.metrics = new Metrics(); + this.metrics = new KafkaStreamingMetrics(); + + this.running = true; } /** @@ -128,7 +127,7 @@ public StreamThread(TopologyBuilder builder, StreamingConfig config) throws Exce */ @Override public synchronized void run() { - log.info("Starting a kstream thread"); + log.info("Starting a stream thread"); try { runLoop(); } catch (RuntimeException e) { @@ -139,16 +138,6 @@ public synchronized void run() { } } - private void shutdown() { - log.info("Shutting down a kstream thread"); - commitAll(time.milliseconds()); - - collector.close(); - consumer.close(); - removePartitions(); - log.info("kstream thread shutdown complete"); - } - /** * Shutdown this streaming thread. */ @@ -156,14 +145,23 @@ public synchronized void close() { running = false; } + private void shutdown() { + log.info("Shutting down a stream thread"); + commitAll(time.milliseconds()); + + producer.close(); + consumer.close(); + removePartitions(); + log.info("Stream thread shutdown complete"); + } + private void runLoop() { try { - boolean readyForNextExecution = false; + int totalNumBuffered = 0; while (stillRunning()) { - // try to fetch some records and put them to tasks' queues - // TODO: we may not need to poll every iteration - ConsumerRecords records = consumer.poll(readyForNextExecution ? 0 : this.pollTimeMs); + // try to fetch some records if necessary + ConsumerRecords records = consumer.poll(totalNumBuffered == 0 ? this.pollTimeMs : 0); for (StreamTask task : tasks.values()) { for (TopicPartition partition : task.partitions()) { @@ -172,13 +170,14 @@ private void runLoop() { } // try to process one record from each task - // TODO: we may want to process more than one record in each iteration + totalNumBuffered = 0; + for (StreamTask task : tasks.values()) { - readyForNextExecution = task.process(); + totalNumBuffered += task.process(); } + maybeClean(); maybeCommit(); - maybeCleanState(); } } catch (Exception e) { throw new KafkaException(e); @@ -190,52 +189,52 @@ private boolean stillRunning() { log.debug("Shutting down at user request."); return false; } + if (totalRecordsToProcess >= 0 && recordsProcessed >= totalRecordsToProcess) { - log.debug("Shutting down as we've reached the user-configured limit of {} records to process.", totalRecordsToProcess); + log.debug("Shutting down as we've reached the user configured limit of {} records to process.", totalRecordsToProcess); return false; } + return true; } private void maybeCommit() { long now = time.milliseconds(); + if (commitTimeMs >= 0 && lastCommit + commitTimeMs < now) { log.trace("Committing processor instances because the commit interval has elapsed."); commitAll(now); } } + /** + * Commit the states of all its tasks + * @param now + */ private void commitAll(long now) { - - /* - Map commit = new HashMap<>(); - for (ProcessorContextImpl context : tasks.values()) { - context.flush(); - commit.putAll(context.consumedOffsets()); + for (StreamTask task : tasks.values()) { + task.commit(); } - // check if commit is really needed, i.e. if all the offsets are already committed - if (consumer.commitNeeded(commit)) { - // TODO: for exactly-once we need to make sure the flush and commit - // are executed atomically whenever it is triggered by user - collector.flush(); - consumer.commit(commit); // TODO: can this be async? - streamingMetrics.commitTime.record(now - lastCommit); - } - */ + metrics.commitTime.record(now - time.milliseconds()); + + lastCommit = now; } - /* delete any state dirs that aren't for active contexts */ - private void maybeCleanState() { + /** + * Cleanup any states of the tasks that have been removed from this thread + */ + private void maybeClean() { long now = time.milliseconds(); - if (now > nextStateCleaning) { + + if (now > lastClean) { File[] stateDirs = stateDir.listFiles(); if (stateDirs != null) { for (File dir : stateDirs) { try { Integer id = Integer.parseInt(dir.getName()); if (!tasks.keySet().contains(id)) { - log.info("Deleting obsolete state directory {} after {} delay ms.", dir.getAbsolutePath(), stateCleanupDelayMs); + log.info("Deleting obsolete state directory {} after delayed {} ms.", dir.getAbsolutePath(), cleanTimeMs); Utils.rm(dir); } } catch (NumberFormatException e) { @@ -244,7 +243,8 @@ private void maybeCleanState() { } } } - nextStateCleaning = Long.MAX_VALUE; + + lastClean = now; } } @@ -263,13 +263,13 @@ private void addPartitions(Collection assignment) { partitionsForTask.add(part); // create the task - task = new StreamTask(id, consumer, builder.build(), partitionsForTask, collector, config); + task = new StreamTask(id, consumer, producer, partitionsForTask, builder.build(), config); tasks.put(id, task); } } - nextStateCleaning = time.milliseconds() + stateCleanupDelayMs; + lastClean = time.milliseconds() + cleanTimeMs; } private void removePartitions() { @@ -282,12 +282,14 @@ private void removePartitions() { } catch (Exception e) { throw new KafkaException(e); } - streamingMetrics.processorDestruction.record(); + metrics.processorDestruction.record(); } tasks.clear(); } private class KafkaStreamingMetrics { + final Metrics metrics; + final Sensor commitTime; final Sensor processTime; final Sensor windowTime; @@ -297,15 +299,17 @@ private class KafkaStreamingMetrics { public KafkaStreamingMetrics() { String group = "kafka-streaming"; + this.metrics = new Metrics(); + this.commitTime = metrics.sensor("commit-time"); this.commitTime.add(new MetricName(group, "commit-time-avg-ms"), new Avg()); - this.commitTime.add(new MetricName(group, "commits-time-max-ms"), new Max()); - this.commitTime.add(new MetricName(group, "commits-per-second"), new Rate(new Count())); + this.commitTime.add(new MetricName(group, "commit-time-max-ms"), new Max()); + this.commitTime.add(new MetricName(group, "commit-per-second"), new Rate(new Count())); this.processTime = metrics.sensor("process-time"); - this.commitTime.add(new MetricName(group, "process-time-avg-ms"), new Avg()); - this.commitTime.add(new MetricName(group, "process-time-max-ms"), new Max()); - this.commitTime.add(new MetricName(group, "process-calls-per-second"), new Rate(new Count())); + this.processTime.add(new MetricName(group, "process-time-avg-ms"), new Avg()); + this.processTime.add(new MetricName(group, "process-time-max-ms"), new Max()); + this.processTime.add(new MetricName(group, "process-calls-per-second"), new Rate(new Count())); this.windowTime = metrics.sensor("window-time"); this.windowTime.add(new MetricName(group, "window-time-avg-ms"), new Avg()); @@ -317,9 +321,6 @@ public KafkaStreamingMetrics() { this.processorDestruction = metrics.sensor("processor-destruction"); this.processorDestruction.add(new MetricName(group, "processor-destruction"), new Rate(new Count())); - } - } - } diff --git a/stream/src/test/java/org/apache/kafka/streaming/FilteredIteratorTest.java b/stream/src/test/java/org/apache/kafka/streaming/FilteredIteratorTest.java deleted file mode 100644 index f2c1f9f5023eb..0000000000000 --- a/stream/src/test/java/org/apache/kafka/streaming/FilteredIteratorTest.java +++ /dev/null @@ -1,95 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.kafka.streaming; - -import static org.junit.Assert.assertEquals; - -import org.apache.kafka.streaming.kstream.internals.FilteredIterator; -import org.junit.Test; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Iterator; -import java.util.List; - -public class FilteredIteratorTest { - - @Test - public void testFiltering() { - List list = Arrays.asList(3, 1, 4, 1, 5, 9, 2, 6, 5, 3, 5); - - Iterator filtered = new FilteredIterator(list.iterator()) { - protected String filter(Integer i) { - if (i % 3 == 0) return i.toString(); - return null; - } - }; - - List expected = Arrays.asList("3", "9", "6", "3"); - List result = new ArrayList(); - - while (filtered.hasNext()) { - result.add(filtered.next()); - } - - assertEquals(expected, result); - } - - @Test - public void testEmptySource() { - List list = new ArrayList(); - - Iterator filtered = new FilteredIterator(list.iterator()) { - protected String filter(Integer i) { - if (i % 3 == 0) return i.toString(); - return null; - } - }; - - List expected = new ArrayList(); - List result = new ArrayList(); - - while (filtered.hasNext()) { - result.add(filtered.next()); - } - - assertEquals(expected, result); - } - - @Test - public void testNoMatch() { - List list = Arrays.asList(3, 1, 4, 1, 5, 9, 2, 6, 5, 3, 5); - - Iterator filtered = new FilteredIterator(list.iterator()) { - protected String filter(Integer i) { - if (i % 7 == 0) return i.toString(); - return null; - } - }; - - List expected = new ArrayList(); - List result = new ArrayList(); - - while (filtered.hasNext()) { - result.add(filtered.next()); - } - - assertEquals(expected, result); - } - -} diff --git a/stream/src/test/java/org/apache/kafka/streaming/KStreamBranchTest.java b/stream/src/test/java/org/apache/kafka/streaming/KStreamBranchTest.java deleted file mode 100644 index 013998df14f7d..0000000000000 --- a/stream/src/test/java/org/apache/kafka/streaming/KStreamBranchTest.java +++ /dev/null @@ -1,90 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.kafka.streaming; - -import org.apache.kafka.common.serialization.IntegerDeserializer; -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.streaming.kstream.KStream; -import org.apache.kafka.streaming.kstream.KStreamBuilder; -import org.apache.kafka.streaming.kstream.Predicate; -import org.apache.kafka.streaming.kstream.internals.KStreamSource; -import org.apache.kafka.test.MockKStreamBuilder; -import org.apache.kafka.test.MockProcessor; -import org.junit.Test; - -import java.lang.reflect.Array; - -import static org.junit.Assert.assertEquals; - -public class KStreamBranchTest { - - private String topic1 = "topic"; - - private KStreamBuilder topology = new MockKStreamBuilder(); - private IntegerDeserializer keyDeserializer = new IntegerDeserializer(); - private StringDeserializer valDeserializer = new StringDeserializer(); - - @SuppressWarnings("unchecked") - @Test - public void testKStreamBranch() { - - Predicate isEven = new Predicate() { - @Override - public boolean apply(Integer key, String value) { - return (key % 2) == 0; - } - }; - Predicate isMultipleOfThree = new Predicate() { - @Override - public boolean apply(Integer key, String value) { - return (key % 3) == 0; - } - }; - Predicate isOdd = new Predicate() { - @Override - public boolean apply(Integer key, String value) { - return (key % 2) != 0; - } - }; - - final int[] expectedKeys = new int[]{1, 2, 3, 4, 5, 6, 7}; - - KStream stream; - KStream[] branches; - MockProcessor[] processors; - - stream = topology.from(keyDeserializer, valDeserializer, topic1); - branches = stream.branch(isEven, isMultipleOfThree, isOdd); - - assertEquals(3, branches.length); - - processors = (MockProcessor[]) Array.newInstance(MockProcessor.class, branches.length); - for (int i = 0; i < branches.length; i++) { - processors[i] = new MockProcessor<>(); - branches[i].process(processors[i]); - } - - for (int i = 0; i < expectedKeys.length; i++) { - ((KStreamSource) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]); - } - - assertEquals(3, processors[0].processed.size()); - assertEquals(2, processors[1].processed.size()); - assertEquals(4, processors[2].processed.size()); - } -} diff --git a/stream/src/test/java/org/apache/kafka/streaming/KStreamFilterTest.java b/stream/src/test/java/org/apache/kafka/streaming/KStreamFilterTest.java deleted file mode 100644 index b344e8c4c4522..0000000000000 --- a/stream/src/test/java/org/apache/kafka/streaming/KStreamFilterTest.java +++ /dev/null @@ -1,84 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.kafka.streaming; - -import org.apache.kafka.common.serialization.IntegerDeserializer; -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.streaming.kstream.KStream; -import org.apache.kafka.streaming.kstream.KStreamBuilder; -import org.apache.kafka.streaming.kstream.Predicate; -import org.apache.kafka.streaming.kstream.internals.KStreamSource; -import org.apache.kafka.test.MockKStreamBuilder; -import org.apache.kafka.test.MockProcessor; - -import org.junit.Test; - -import static org.junit.Assert.assertEquals; - -public class KStreamFilterTest { - - private String topicName = "topic"; - - private KStreamBuilder topology = new MockKStreamBuilder(); - private IntegerDeserializer keyDeserializer = new IntegerDeserializer(); - private StringDeserializer valDeserializer = new StringDeserializer(); - - private Predicate isMultipleOfThree = new Predicate() { - @Override - public boolean apply(Integer key, String value) { - return (key % 3) == 0; - } - }; - - @Test - public void testFilter() { - final int[] expectedKeys = new int[]{1, 2, 3, 4, 5, 6, 7}; - - KStream stream; - MockProcessor processor; - - processor = new MockProcessor<>(); - stream = topology.from(keyDeserializer, valDeserializer, topicName); - stream.filter(isMultipleOfThree).process(processor); - - for (int i = 0; i < expectedKeys.length; i++) { - ((KStreamSource) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]); - } - - assertEquals(2, processor.processed.size()); - } - - @Test - public void testFilterOut() { - final int[] expectedKeys = new int[]{1, 2, 3, 4, 5, 6, 7}; - - KStream stream; - MockProcessor processor; - - processor = new MockProcessor<>(); - stream = topology.from(keyDeserializer, valDeserializer, topicName); - stream.filterOut(isMultipleOfThree).process(processor); - - for (int i = 0; i < expectedKeys.length; i++) { - ((KStreamSource) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]); - } - - assertEquals(5, processor.processed.size()); - } - -} diff --git a/stream/src/test/java/org/apache/kafka/streaming/KStreamMapValuesTest.java b/stream/src/test/java/org/apache/kafka/streaming/KStreamMapValuesTest.java deleted file mode 100644 index 8142fe1237181..0000000000000 --- a/stream/src/test/java/org/apache/kafka/streaming/KStreamMapValuesTest.java +++ /dev/null @@ -1,71 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.kafka.streaming; - -import org.apache.kafka.common.serialization.IntegerDeserializer; -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.streaming.kstream.KStream; -import org.apache.kafka.streaming.kstream.KStreamBuilder; -import org.apache.kafka.streaming.kstream.ValueMapper; -import org.apache.kafka.streaming.kstream.internals.KStreamSource; -import org.apache.kafka.test.MockKStreamBuilder; -import org.apache.kafka.test.MockProcessor; -import org.junit.Test; - -import static org.junit.Assert.assertEquals; - -public class KStreamMapValuesTest { - - private String topicName = "topic"; - - private KStreamBuilder topology = new MockKStreamBuilder(); - private IntegerDeserializer keyDeserializer = new IntegerDeserializer(); - private StringDeserializer valDeserializer = new StringDeserializer(); - - @Test - public void testFlatMapValues() { - - ValueMapper mapper = - new ValueMapper() { - @Override - public Integer apply(String value) { - return value.length(); - } - }; - - final int[] expectedKeys = new int[]{1, 10, 100, 1000}; - - KStream stream; - MockProcessor processor = new MockProcessor<>(); - stream = topology.from(keyDeserializer, valDeserializer, topicName); - stream.mapValues(mapper).process(processor); - - for (int i = 0; i < expectedKeys.length; i++) { - ((KStreamSource) stream).source().process(expectedKeys[i], Integer.toString(expectedKeys[i])); - } - - assertEquals(4, processor.processed.size()); - - String[] expected = new String[]{"1:1", "10:2", "100:3", "1000:4"}; - - for (int i = 0; i < expected.length; i++) { - assertEquals(expected[i], processor.processed.get(i)); - } - } - -} diff --git a/stream/src/test/java/org/apache/kafka/streaming/KStreamSourceTest.java b/stream/src/test/java/org/apache/kafka/streaming/KStreamSourceTest.java deleted file mode 100644 index 5790e34cb1723..0000000000000 --- a/stream/src/test/java/org/apache/kafka/streaming/KStreamSourceTest.java +++ /dev/null @@ -1,59 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.kafka.streaming; - -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.streaming.kstream.KStream; -import org.apache.kafka.streaming.kstream.KStreamBuilder; -import org.apache.kafka.streaming.kstream.internals.KStreamSource; -import org.apache.kafka.test.MockKStreamBuilder; -import org.apache.kafka.test.MockProcessor; -import org.junit.Test; - -import static org.junit.Assert.assertEquals; - -public class KStreamSourceTest { - - private String topicName = "topic"; - - private KStreamBuilder topology = new MockKStreamBuilder(); - private StringDeserializer keyDeserializer = new StringDeserializer(); - private StringDeserializer valDeserializer = new StringDeserializer(); - - @Test - public void testKStreamSource() { - - MockProcessor processor = new MockProcessor<>(); - - KStream stream = topology.from(keyDeserializer, valDeserializer, topicName); - stream.process(processor); - - final String[] expectedKeys = new String[]{"k1", "k2", "k3"}; - final String[] expectedValues = new String[]{"v1", "v2", "v3"}; - - for (int i = 0; i < expectedKeys.length; i++) { - ((KStreamSource) stream).source().process(expectedKeys[i], expectedValues[i]); - } - - assertEquals(3, processor.processed.size()); - - for (int i = 0; i < expectedKeys.length; i++) { - assertEquals(expectedKeys[i] + ":" + expectedValues[i], processor.processed.get(i)); - } - } -} diff --git a/stream/src/test/java/org/apache/kafka/streaming/KStreamWindowedTest.java b/stream/src/test/java/org/apache/kafka/streaming/KStreamWindowedTest.java deleted file mode 100644 index 85fcbec8d1825..0000000000000 --- a/stream/src/test/java/org/apache/kafka/streaming/KStreamWindowedTest.java +++ /dev/null @@ -1,91 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.kafka.streaming; - -import org.apache.kafka.common.serialization.IntegerDeserializer; -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.streaming.kstream.KStream; -import org.apache.kafka.streaming.kstream.KStreamBuilder; -import org.apache.kafka.streaming.kstream.WindowDef; -import org.apache.kafka.streaming.kstream.internals.KStreamSource; -import org.apache.kafka.test.MockKStreamBuilder; -import org.apache.kafka.test.MockProcessorContext; -import org.apache.kafka.test.UnlimitedWindow; -import org.junit.Test; - -import java.util.Iterator; - -import static org.junit.Assert.assertEquals; - -public class KStreamWindowedTest { - - private String topicName = "topic"; - - private KStreamBuilder topology = new MockKStreamBuilder(); - private IntegerDeserializer keyDeserializer = new IntegerDeserializer(); - private StringDeserializer valDeserializer = new StringDeserializer(); - - @Test - public void testWindowedStream() { - - final int[] expectedKeys = new int[]{0, 1, 2, 3}; - - KStream stream; - WindowDef window; - - window = new UnlimitedWindow<>(); - stream = topology.from(keyDeserializer, valDeserializer, topicName); - stream.with(window); - - MockProcessorContext context = new MockProcessorContext(null, null); - topology.init(context); - context.setTime(0L); - - // two items in the window - - for (int i = 0; i < 2; i++) { - ((KStreamSource) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]); - } - - assertEquals(1, countItem(window.find(0, 0L))); - assertEquals(1, countItem(window.find(1, 0L))); - assertEquals(0, countItem(window.find(2, 0L))); - assertEquals(0, countItem(window.find(3, 0L))); - - // previous two items + all items, thus two are duplicates, in the window - - for (int i = 0; i < expectedKeys.length; i++) { - ((KStreamSource) stream).source().process(expectedKeys[i], "Y" + expectedKeys[i]); - } - - assertEquals(2, countItem(window.find(0, 0L))); - assertEquals(2, countItem(window.find(1, 0L))); - assertEquals(1, countItem(window.find(2, 0L))); - assertEquals(1, countItem(window.find(3, 0L))); - } - - - private int countItem(Iterator iter) { - int i = 0; - while (iter.hasNext()) { - i++; - iter.next(); - } - return i; - } -} diff --git a/stream/src/test/java/org/apache/kafka/streaming/StreamGroupTest.java b/stream/src/test/java/org/apache/kafka/streaming/StreamGroupTest.java deleted file mode 100644 index 8f43c4cb1c2c8..0000000000000 --- a/stream/src/test/java/org/apache/kafka/streaming/StreamGroupTest.java +++ /dev/null @@ -1,162 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.kafka.streaming; - -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.streaming.processor.TimestampExtractor; -import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.serialization.Deserializer; -import org.apache.kafka.common.serialization.IntegerDeserializer; -import org.apache.kafka.common.serialization.IntegerSerializer; -import org.apache.kafka.common.serialization.Serializer; -import org.apache.kafka.streaming.processor.internals.StreamGroup; -import org.apache.kafka.test.MockProcessorContext; -import org.apache.kafka.test.MockSourceNode; -import org.junit.Test; - -import java.util.Arrays; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; - -public class StreamGroupTest { - - private static Serializer serializer = new IntegerSerializer(); - private static Deserializer deserializer = new IntegerDeserializer(); - - @SuppressWarnings("unchecked") - @Test - public void testAddPartition() { - - MockIngestor mockIngestor = new MockIngestor(); - - StreamGroup streamGroup = new StreamGroup( - new MockProcessorContext(serializer, deserializer), - mockIngestor, - new TimestampExtractor() { - public long extract(String topic, Object key, Object value) { - if (topic.equals("topic1")) - return ((Integer) key).longValue(); - else - return ((Integer) key).longValue() / 10L + 5L; - } - }, - 3 - ); - - TopicPartition partition1 = new TopicPartition("topic1", 1); - TopicPartition partition2 = new TopicPartition("topic2", 1); - MockSourceNode source1 = new MockSourceNode(deserializer, deserializer); - MockSourceNode source2 = new MockSourceNode(deserializer, deserializer); - MockSourceNode source3 = new MockSourceNode(deserializer, deserializer); - - streamGroup.addPartition(partition1, source1); - mockIngestor.addPartitionStreamToGroup(streamGroup, partition1); - - streamGroup.addPartition(partition2, source2); - mockIngestor.addPartitionStreamToGroup(streamGroup, partition2); - - Exception exception = null; - try { - streamGroup.addPartition(partition1, source3); - } catch (Exception ex) { - exception = ex; - } - assertTrue(exception != null); - - byte[] recordValue = serializer.serialize(null, new Integer(10)); - - mockIngestor.addRecords(partition1, records( - new ConsumerRecord<>(partition1.topic(), partition1.partition(), 1, serializer.serialize(partition1.topic(), new Integer(10)), recordValue), - new ConsumerRecord<>(partition1.topic(), partition1.partition(), 2, serializer.serialize(partition1.topic(), new Integer(20)), recordValue) - )); - - mockIngestor.addRecords(partition2, records( - new ConsumerRecord<>(partition2.topic(), partition2.partition(), 1, serializer.serialize(partition1.topic(), new Integer(300)), recordValue), - new ConsumerRecord<>(partition2.topic(), partition2.partition(), 2, serializer.serialize(partition1.topic(), new Integer(400)), recordValue), - new ConsumerRecord<>(partition2.topic(), partition2.partition(), 3, serializer.serialize(partition1.topic(), new Integer(500)), recordValue), - new ConsumerRecord<>(partition2.topic(), partition2.partition(), 4, serializer.serialize(partition1.topic(), new Integer(600)), recordValue) - )); - - streamGroup.process(); - assertEquals(source1.numReceived, 1); - assertEquals(source2.numReceived, 0); - - assertEquals(mockIngestor.paused.size(), 1); - assertTrue(mockIngestor.paused.contains(partition2)); - - mockIngestor.addRecords(partition1, records( - new ConsumerRecord<>(partition1.topic(), partition1.partition(), 3, serializer.serialize(partition1.topic(), new Integer(30)), recordValue), - new ConsumerRecord<>(partition1.topic(), partition1.partition(), 4, serializer.serialize(partition1.topic(), new Integer(40)), recordValue), - new ConsumerRecord<>(partition1.topic(), partition1.partition(), 5, serializer.serialize(partition1.topic(), new Integer(50)), recordValue) - )); - - streamGroup.process(); - assertEquals(source1.numReceived, 2); - assertEquals(source2.numReceived, 0); - - assertEquals(mockIngestor.paused.size(), 2); - assertTrue(mockIngestor.paused.contains(partition1)); - assertTrue(mockIngestor.paused.contains(partition2)); - - streamGroup.process(); - assertEquals(source1.numReceived, 3); - assertEquals(source2.numReceived, 0); - - streamGroup.process(); - assertEquals(source1.numReceived, 3); - assertEquals(source2.numReceived, 1); - - assertEquals(mockIngestor.paused.size(), 1); - assertTrue(mockIngestor.paused.contains(partition2)); - - streamGroup.process(); - assertEquals(source1.numReceived, 4); - assertEquals(source2.numReceived, 1); - - assertEquals(mockIngestor.paused.size(), 1); - - streamGroup.process(); - assertEquals(source1.numReceived, 4); - assertEquals(source2.numReceived, 2); - - assertEquals(mockIngestor.paused.size(), 0); - - streamGroup.process(); - assertEquals(source1.numReceived, 5); - assertEquals(source2.numReceived, 2); - - streamGroup.process(); - assertEquals(source1.numReceived, 5); - assertEquals(source2.numReceived, 3); - - streamGroup.process(); - assertEquals(source1.numReceived, 5); - assertEquals(source2.numReceived, 4); - - assertEquals(mockIngestor.paused.size(), 0); - - streamGroup.process(); - assertEquals(source1.numReceived, 5); - assertEquals(source2.numReceived, 4); - } - - private Iterable> records(ConsumerRecord... recs) { - return Arrays.asList(recs); - } -} diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamFlatMapTest.java b/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamFlatMapTest.java deleted file mode 100644 index 82b36308a0c8b..0000000000000 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamFlatMapTest.java +++ /dev/null @@ -1,80 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.kafka.streaming.internals; - -import org.apache.kafka.common.serialization.IntegerDeserializer; -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.streaming.kstream.KStream; -import org.apache.kafka.streaming.kstream.KStreamBuilder; -import org.apache.kafka.streaming.kstream.KeyValue; -import org.apache.kafka.streaming.kstream.KeyValueMapper; -import org.apache.kafka.streaming.kstream.internals.KStreamSource; -import org.apache.kafka.test.MockKStreamBuilder; -import org.apache.kafka.test.MockProcessor; -import org.junit.Test; - -import static org.junit.Assert.assertEquals; - -import java.util.ArrayList; - -public class KStreamFlatMapTest { - - private String topicName = "topic"; - - private KStreamBuilder topology = new MockKStreamBuilder(); - private IntegerDeserializer keyDeserializer = new IntegerDeserializer(); - private StringDeserializer valDeserializer = new StringDeserializer(); - - @Test - public void testFlatMap() { - - KeyValueMapper> mapper = - new KeyValueMapper>() { - @Override - public KeyValue> apply(Integer key, String value) { - ArrayList result = new ArrayList(); - for (int i = 0; i < key; i++) { - result.add(value); - } - return KeyValue.pair(Integer.toString(key * 10), (Iterable) result); - } - }; - - final int[] expectedKeys = new int[]{0, 1, 2, 3}; - - KStream stream; - MockProcessor processor; - - processor = new MockProcessor<>(); - stream = topology.from(keyDeserializer, valDeserializer, topicName); - stream.flatMap(mapper).process(processor); - - for (int i = 0; i < expectedKeys.length; i++) { - ((KStreamSource) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]); - } - - assertEquals(6, processor.processed.size()); - - String[] expected = new String[]{"10:V1", "20:V2", "20:V2", "30:V3", "30:V3", "30:V3"}; - - for (int i = 0; i < expected.length; i++) { - assertEquals(expected[i], processor.processed.get(i)); - } - } - -} diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamFlatMapValuesTest.java b/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamFlatMapValuesTest.java deleted file mode 100644 index 9c5863bf3ed19..0000000000000 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamFlatMapValuesTest.java +++ /dev/null @@ -1,77 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.kafka.streaming.internals; - -import org.apache.kafka.common.serialization.IntegerDeserializer; -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.streaming.kstream.KStream; -import org.apache.kafka.streaming.kstream.KStreamBuilder; -import org.apache.kafka.streaming.kstream.ValueMapper; -import org.apache.kafka.streaming.kstream.internals.KStreamSource; -import org.apache.kafka.test.MockKStreamBuilder; -import org.apache.kafka.test.MockProcessor; -import org.junit.Test; - -import java.util.ArrayList; - -import static org.junit.Assert.assertEquals; - -public class KStreamFlatMapValuesTest { - - private String topicName = "topic"; - - private KStreamBuilder topology = new MockKStreamBuilder(); - private IntegerDeserializer keyDeserializer = new IntegerDeserializer(); - private StringDeserializer valDeserializer = new StringDeserializer(); - - @Test - public void testFlatMapValues() { - - ValueMapper> mapper = - new ValueMapper>() { - @Override - public Iterable apply(String value) { - ArrayList result = new ArrayList(); - result.add(value.toLowerCase()); - result.add(value); - return result; - } - }; - - final int[] expectedKeys = new int[]{0, 1, 2, 3}; - - KStream stream; - MockProcessor processor; - - processor = new MockProcessor<>(); - stream = topology.from(keyDeserializer, valDeserializer, topicName); - stream.flatMapValues(mapper).process(processor); - - for (int i = 0; i < expectedKeys.length; i++) { - ((KStreamSource) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]); - } - - assertEquals(8, processor.processed.size()); - - String[] expected = new String[]{"0:v0", "0:V0", "1:v1", "1:V1", "2:v2", "2:V2", "3:v3", "3:V3"}; - - for (int i = 0; i < expected.length; i++) { - assertEquals(expected[i], processor.processed.get(i)); - } - } -} diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamJoinTest.java b/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamJoinTest.java deleted file mode 100644 index bdc2b23a82e5b..0000000000000 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamJoinTest.java +++ /dev/null @@ -1,249 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.kafka.streaming.internals; - -import org.apache.kafka.common.serialization.IntegerDeserializer; -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.common.utils.Utils; -import org.apache.kafka.streaming.kstream.KStream; -import org.apache.kafka.streaming.kstream.KStreamBuilder; -import org.apache.kafka.streaming.kstream.KStreamWindowed; -import org.apache.kafka.streaming.kstream.KeyValue; -import org.apache.kafka.streaming.kstream.KeyValueMapper; -import org.apache.kafka.streaming.kstream.ValueJoiner; -import org.apache.kafka.streaming.kstream.ValueMapper; -import org.apache.kafka.streaming.kstream.internals.KStreamSource; -import org.apache.kafka.test.MockKStreamBuilder; -import org.apache.kafka.test.MockProcessor; -import org.apache.kafka.test.MockProcessorContext; -import org.apache.kafka.test.UnlimitedWindow; -import org.junit.Test; - -import static org.junit.Assert.assertEquals; - -public class KStreamJoinTest { - - private String topic1 = "topic1"; - private String topic2 = "topic2"; - - private KStreamBuilder topology = new MockKStreamBuilder(); - private IntegerDeserializer keyDeserializer = new IntegerDeserializer(); - private StringDeserializer valDeserializer = new StringDeserializer(); - - private ValueJoiner joiner = new ValueJoiner() { - @Override - public String apply(String value1, String value2) { - return value1 + "+" + value2; - } - }; - - private ValueMapper valueMapper = new ValueMapper() { - @Override - public String apply(String value) { - return "#" + value; - } - }; - - private ValueMapper> valueMapper2 = new ValueMapper>() { - @Override - public Iterable apply(String value) { - return (Iterable) Utils.mkSet(value); - } - }; - - private KeyValueMapper keyValueMapper = - new KeyValueMapper() { - @Override - public KeyValue apply(Integer key, String value) { - return KeyValue.pair(key, value); - } - }; - - KeyValueMapper> keyValueMapper2 = - new KeyValueMapper>() { - @Override - public KeyValue> apply(Integer key, String value) { - return KeyValue.pair(key, (Iterable) Utils.mkSet(value)); - } - }; - - - @Test - public void testJoin() { - - final int[] expectedKeys = new int[]{0, 1, 2, 3}; - - KStream stream1; - KStream stream2; - KStreamWindowed windowed1; - KStreamWindowed windowed2; - MockProcessor processor; - String[] expected; - - processor = new MockProcessor<>(); - stream1 = topology.from(keyDeserializer, valDeserializer, topic1); - stream2 = topology.from(keyDeserializer, valDeserializer, topic2); - windowed1 = stream1.with(new UnlimitedWindow()); - windowed2 = stream2.with(new UnlimitedWindow()); - - windowed1.join(windowed2, joiner).process(processor); - - MockProcessorContext context = new MockProcessorContext(null, null); - topology.init(context); - context.setTime(0L); - - // push two items to the main stream. the other stream's window is empty - - for (int i = 0; i < 2; i++) { - ((KStreamSource) stream1).source().process(expectedKeys[i], "X" + expectedKeys[i]); - } - - assertEquals(0, processor.processed.size()); - - // push two items to the other stream. the main stream's window has two items - - for (int i = 0; i < 2; i++) { - ((KStreamSource) stream2).source().process(expectedKeys[i], "Y" + expectedKeys[i]); - } - - assertEquals(2, processor.processed.size()); - - expected = new String[]{"0:X0+Y0", "1:X1+Y1"}; - - for (int i = 0; i < expected.length; i++) { - assertEquals(expected[i], processor.processed.get(i)); - } - - processor.processed.clear(); - - // push all items to the main stream. this should produce two items. - - for (int i = 0; i < expectedKeys.length; i++) { - ((KStreamSource) stream1).source().process(expectedKeys[i], "X" + expectedKeys[i]); - } - - assertEquals(2, processor.processed.size()); - - expected = new String[]{"0:X0+Y0", "1:X1+Y1"}; - - for (int i = 0; i < expected.length; i++) { - assertEquals(expected[i], processor.processed.get(i)); - } - - processor.processed.clear(); - - // there will be previous two items + all items in the main stream's window, thus two are duplicates. - - // push all items to the other stream. this should produce 6 items - for (int i = 0; i < expectedKeys.length; i++) { - ((KStreamSource) stream2).source().process(expectedKeys[i], "Y" + expectedKeys[i]); - } - - assertEquals(6, processor.processed.size()); - - expected = new String[]{"0:X0+Y0", "0:X0+Y0", "1:X1+Y1", "1:X1+Y1", "2:X2+Y2", "3:X3+Y3"}; - - for (int i = 0; i < expected.length; i++) { - assertEquals(expected[i], processor.processed.get(i)); - } - } - - @Test - public void testJoinPrior() { - - final int[] expectedKeys = new int[]{0, 1, 2, 3}; - - KStream stream1; - KStream stream2; - KStreamWindowed windowed1; - KStreamWindowed windowed2; - MockProcessor processor; - String[] expected; - - processor = new MockProcessor<>(); - stream1 = topology.from(keyDeserializer, valDeserializer, topic1); - stream2 = topology.from(keyDeserializer, valDeserializer, topic2); - windowed1 = stream1.with(new UnlimitedWindow()); - windowed2 = stream2.with(new UnlimitedWindow()); - - windowed1.joinPrior(windowed2, joiner).process(processor); - - MockProcessorContext context = new MockProcessorContext(null, null); - topology.init(context); - - // push two items to the main stream. the other stream's window is empty - - for (int i = 0; i < 2; i++) { - context.setTime(i); - - ((KStreamSource) stream1).source().process(expectedKeys[i], "X" + expectedKeys[i]); - } - - assertEquals(0, processor.processed.size()); - - // push two items to the other stream. the main stream's window has two items - // no corresponding item in the main window has a newer timestamp - - for (int i = 0; i < 2; i++) { - context.setTime(i + 1); - - ((KStreamSource) stream2).source().process(expectedKeys[i], "Y" + expectedKeys[i]); - } - - assertEquals(0, processor.processed.size()); - - processor.processed.clear(); - - // push all items with newer timestamps to the main stream. this should produce two items. - - for (int i = 0; i < expectedKeys.length; i++) { - context.setTime(i + 2); - - ((KStreamSource) stream1).source().process(expectedKeys[i], "X" + expectedKeys[i]); - } - - assertEquals(2, processor.processed.size()); - - expected = new String[]{"0:X0+Y0", "1:X1+Y1"}; - - for (int i = 0; i < expected.length; i++) { - assertEquals(expected[i], processor.processed.get(i)); - } - - processor.processed.clear(); - - // there will be previous two items + all items in the main stream's window, thus two are duplicates. - - // push all items with older timestamps to the other stream. this should produce six items - for (int i = 0; i < expectedKeys.length; i++) { - context.setTime(i); - - ((KStreamSource) stream2).source().process(expectedKeys[i], "Y" + expectedKeys[i]); - } - - assertEquals(6, processor.processed.size()); - - expected = new String[]{"0:X0+Y0", "0:X0+Y0", "1:X1+Y1", "1:X1+Y1", "2:X2+Y2", "3:X3+Y3"}; - - for (int i = 0; i < expected.length; i++) { - assertEquals(expected[i], processor.processed.get(i)); - } - } - - // TODO: test for joinability -} diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamMapTest.java b/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamMapTest.java deleted file mode 100644 index 7104e5efefec4..0000000000000 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamMapTest.java +++ /dev/null @@ -1,73 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.kafka.streaming.internals; - -import org.apache.kafka.common.serialization.IntegerDeserializer; -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.streaming.kstream.KStream; -import org.apache.kafka.streaming.kstream.KStreamBuilder; -import org.apache.kafka.streaming.kstream.KeyValue; -import org.apache.kafka.streaming.kstream.KeyValueMapper; -import org.apache.kafka.streaming.kstream.internals.KStreamSource; -import org.apache.kafka.test.MockKStreamBuilder; -import org.apache.kafka.test.MockProcessor; -import org.junit.Test; - -import static org.junit.Assert.assertEquals; - -public class KStreamMapTest { - - private String topicName = "topic"; - - private KStreamBuilder topology = new MockKStreamBuilder(); - private IntegerDeserializer keyDeserializer = new IntegerDeserializer(); - private StringDeserializer valDeserializer = new StringDeserializer(); - - @Test - public void testMap() { - - KeyValueMapper mapper = - new KeyValueMapper() { - @Override - public KeyValue apply(Integer key, String value) { - return KeyValue.pair(value, key); - } - }; - - final int[] expectedKeys = new int[]{0, 1, 2, 3}; - - KStream stream; - MockProcessor processor; - - processor = new MockProcessor<>(); - stream = topology.from(keyDeserializer, valDeserializer, topicName); - stream.map(mapper).process(processor); - - for (int i = 0; i < expectedKeys.length; i++) { - ((KStreamSource) stream).source().process(expectedKeys[i], "V" + expectedKeys[i]); - } - - assertEquals(4, processor.processed.size()); - - String[] expected = new String[]{"V0:0", "V1:1", "V2:2", "V3:3"}; - - for (int i = 0; i < expected.length; i++) { - assertEquals(expected[i], processor.processed.get(i)); - } - } -} diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/FilteredIteratorTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/FilteredIteratorTest.java similarity index 98% rename from stream/src/test/java/org/apache/kafka/streaming/internals/FilteredIteratorTest.java rename to stream/src/test/java/org/apache/kafka/streaming/kstream/internals/FilteredIteratorTest.java index 2bd658c826a0c..ff8ed098f0977 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/FilteredIteratorTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/FilteredIteratorTest.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.streaming.internals; +package org.apache.kafka.streaming.kstream.internals; import static org.junit.Assert.assertEquals; diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamBranchTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamBranchTest.java similarity index 98% rename from stream/src/test/java/org/apache/kafka/streaming/internals/KStreamBranchTest.java rename to stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamBranchTest.java index 518d00f43d7e7..217a65f79afde 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamBranchTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamBranchTest.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.streaming.internals; +package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.common.serialization.IntegerDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamFilterTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFilterTest.java similarity index 98% rename from stream/src/test/java/org/apache/kafka/streaming/internals/KStreamFilterTest.java rename to stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFilterTest.java index 276302b738bbf..653550e22085d 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamFilterTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFilterTest.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.streaming.internals; +package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.common.serialization.IntegerDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; diff --git a/stream/src/test/java/org/apache/kafka/streaming/KStreamFlatMapTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapTest.java similarity index 98% rename from stream/src/test/java/org/apache/kafka/streaming/KStreamFlatMapTest.java rename to stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapTest.java index 0f9302491d0b6..25abac1c13f8c 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/KStreamFlatMapTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapTest.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.streaming; +package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.common.serialization.IntegerDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; diff --git a/stream/src/test/java/org/apache/kafka/streaming/KStreamFlatMapValuesTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValuesTest.java similarity index 98% rename from stream/src/test/java/org/apache/kafka/streaming/KStreamFlatMapValuesTest.java rename to stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValuesTest.java index 96e2098aa4db5..8f37c6fc4674f 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/KStreamFlatMapValuesTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValuesTest.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.streaming; +package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.common.serialization.IntegerDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; diff --git a/stream/src/test/java/org/apache/kafka/streaming/KStreamJoinTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamJoinTest.java similarity index 99% rename from stream/src/test/java/org/apache/kafka/streaming/KStreamJoinTest.java rename to stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamJoinTest.java index 815ff2c83f71a..472dfd44eb7f0 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/KStreamJoinTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamJoinTest.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.streaming; +package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.common.serialization.IntegerDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; diff --git a/stream/src/test/java/org/apache/kafka/streaming/KStreamMapTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapTest.java similarity index 98% rename from stream/src/test/java/org/apache/kafka/streaming/KStreamMapTest.java rename to stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapTest.java index 8b5fc638cd366..08711cdd8b178 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/KStreamMapTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapTest.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.streaming; +package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.common.serialization.IntegerDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamMapValuesTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValuesTest.java similarity index 98% rename from stream/src/test/java/org/apache/kafka/streaming/internals/KStreamMapValuesTest.java rename to stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValuesTest.java index 9cca557bc85d9..45a793fc7ab9f 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamMapValuesTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValuesTest.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.streaming.internals; +package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.common.serialization.IntegerDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamSourceTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamSourceTest.java similarity index 97% rename from stream/src/test/java/org/apache/kafka/streaming/internals/KStreamSourceTest.java rename to stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamSourceTest.java index 2f792f415605a..ac295fd7419da 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamSourceTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamSourceTest.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.streaming.internals; +package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.streaming.kstream.KStream; diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamWindowedTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedTest.java similarity index 98% rename from stream/src/test/java/org/apache/kafka/streaming/internals/KStreamWindowedTest.java rename to stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedTest.java index aaa4e3a04a1f0..20cc445e814b4 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/KStreamWindowedTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedTest.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.streaming.internals; +package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.common.serialization.IntegerDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/StreamGroupTest.java b/stream/src/test/java/org/apache/kafka/streaming/processor/internals/StreamGroupTest.java similarity index 99% rename from stream/src/test/java/org/apache/kafka/streaming/internals/StreamGroupTest.java rename to stream/src/test/java/org/apache/kafka/streaming/processor/internals/StreamGroupTest.java index 496174ad52872..ac82bab0cd538 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/StreamGroupTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/processor/internals/StreamGroupTest.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.streaming.internals; +package org.apache.kafka.streaming.processor.internals; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.streaming.processor.TimestampExtractor;