From 113f61e33a16c25ca84197f81a4a7f6867e28b16 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Wed, 26 Aug 2015 15:39:12 -0700 Subject: [PATCH 1/7] wip: commit task states --- .../processor/internals/PartitionGroup.java | 13 +++--- .../internals/ProcessorContextImpl.java | 6 ++- .../internals/ProcessorStateManager.java | 4 -- .../processor/internals/RecordQueue.java | 18 ++++---- .../processor/internals/StampedRecord.java | 1 - .../processor/internals/StreamTask.java | 42 +++++++++++++++---- .../processor/internals/StreamThread.java | 17 ++++---- 7 files changed, 65 insertions(+), 36 deletions(-) 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..bcde36dbdc3b5 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,6 +81,8 @@ 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 @@ -92,7 +95,7 @@ public RecordQueue nextQueue() { } /** - * 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 +146,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..57d695242dd63 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; } @@ -220,7 +224,7 @@ public void forward(K key, V value, int childIndex) { @Override public void commit() { - task.commitOffset(); + task.commit(); } @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..4c1b22342aa0e 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.Callback; +import org.apache.kafka.clients.producer.RecordMetadata; 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.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,6 +57,8 @@ public class StreamTask { private final TimestampExtractor timestampExtractor; private final Map consumedOffsets; + private final Map producedoffsets; + private final Callback producerCallback; private boolean commitRequested = false; private StampedRecord currRecord = null; @@ -98,8 +107,22 @@ public StreamTask(int id, topology.init(this.processorContext); - // initialize the consumed offset cache + // initialize the consumed and produced offset cache this.consumedOffsets = new HashMap<>(); + this.producedoffsets = new HashMap<>(); + + this.producerCallback = new Callback() { + + @Override + public void onCompletion(RecordMetadata metadata, Exception exception) { + if (exception == null) { + TopicPartition partition = new TopicPartition(metadata.topic(), metadata.partition()); + producedoffsets.put(partition, metadata.offset()); + } else { + log.error("Error sending record: ", exception); + } + } + }; } public int id() { @@ -165,14 +188,19 @@ public boolean process() { this.currNode = queue.source(); this.currNode.process(currRecord.key(), currRecord.value()); - // update the consumed offset map. + // update the consumed offset map after processing is done consumedOffsets.put(queue.partition(), currRecord.offset()); + // commit the current task state if requested during the processing if (commitRequested) { - // TODO: flush the following states atomically // 1) flush local state + ((ProcessorContextImpl) processorContext).stateManager().flush(); + // 2) commit consumed offsets + consumer.commit(consumedOffsets, CommitType.SYNC); + // 3) flush produced records in the downstream + ((ProcessorContextImpl) processorContext).recordCollector().flush(); } // we can continue processing this task as long as its @@ -209,9 +237,9 @@ public void node(ProcessorNode node) { } /** - * Request committing the current record's offset + * Request committing the current task's state */ - public void commitOffset() { + public void commit() { this.commitRequested = true; } 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..be973e1b30c17 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 @@ -53,9 +53,9 @@ public class StreamThread extends Thread { private static final Logger log = LoggerFactory.getLogger(StreamThread.class); - private final Consumer consumer; private final TopologyBuilder builder; private final RecordCollector collector; + private final Consumer consumer; private final Map tasks = new HashMap<>(); private final Metrics metrics; private final Time time; @@ -128,7 +128,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) { @@ -140,13 +140,13 @@ public synchronized void run() { } private void shutdown() { - log.info("Shutting down a kstream thread"); + log.info("Shutting down a stream thread"); commitAll(time.milliseconds()); collector.close(); consumer.close(); removePartitions(); - log.info("kstream thread shutdown complete"); + log.info("Stream thread shutdown complete"); } /** @@ -206,8 +206,6 @@ private void maybeCommit() { } private void commitAll(long now) { - - /* Map commit = new HashMap<>(); for (ProcessorContextImpl context : tasks.values()) { context.flush(); @@ -222,7 +220,6 @@ private void commitAll(long now) { consumer.commit(commit); // TODO: can this be async? streamingMetrics.commitTime.record(now - lastCommit); } - */ } /* delete any state dirs that aren't for active contexts */ @@ -303,9 +300,9 @@ public KafkaStreamingMetrics() { this.commitTime.add(new MetricName(group, "commits-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()); From 1e60ee44d12e61b454f98b56a1ac1cab6f27a876 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Wed, 26 Aug 2015 17:06:18 -0700 Subject: [PATCH 2/7] Fix fetch / commit / clean logic --- .../processor/internals/PartitionGroup.java | 7 +- .../internals/ProcessorContextImpl.java | 2 +- .../processor/internals/StreamTask.java | 120 +++++++++-------- .../processor/internals/StreamThread.java | 126 +++++++++--------- 4 files changed, 132 insertions(+), 123 deletions(-) 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 bcde36dbdc3b5..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 @@ -86,12 +86,7 @@ public StampedRecord getRecord(RecordQueue queue) { */ 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(); } /** 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 57d695242dd63..a4954b408f806 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 @@ -224,7 +224,7 @@ public void forward(K key, V value, int childIndex) { @Override public void commit() { - task.commit(); + task.needCommit(); } @Override 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 4c1b22342aa0e..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 @@ -20,12 +20,12 @@ 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.Callback; -import org.apache.kafka.clients.producer.RecordMetadata; +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.common.serialization.Serializer; import org.apache.kafka.streaming.StreamingConfig; import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.ProcessorContext; @@ -57,10 +57,10 @@ public class StreamTask { private final TimestampExtractor timestampExtractor; private final Map consumedOffsets; - private final Map producedoffsets; - private final Callback producerCallback; + private final RecordCollector recordCollector; private boolean commitRequested = false; + private boolean commitOffsetNeeded = false; private StampedRecord currRecord = null; private ProcessorNode currNode = null; @@ -69,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; @@ -98,31 +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 and produced offset cache - this.consumedOffsets = new HashMap<>(); - this.producedoffsets = new HashMap<>(); - - this.producerCallback = new Callback() { - - @Override - public void onCompletion(RecordMetadata metadata, Exception exception) { - if (exception == null) { - TopicPartition partition = new TopicPartition(metadata.topic(), metadata.partition()); - producedoffsets.put(partition, metadata.offset()); - } else { - log.error("Error sending record: ", exception); - } - } - }; } public int id() { @@ -159,29 +152,30 @@ public void addRecords(TopicPartition partition, Iterator 0) { - readyForNextExecution = true; + commit(); } // if after processing this record, its partition queue's buffered size has been @@ -220,7 +202,7 @@ public boolean process() { long timestamp = partitionGroup.timestamp(); punctuationQueue.mayPunctuate(timestamp); - return readyForNextExecution; + return partitionGroup.numbuffered(); } } @@ -237,18 +219,46 @@ public void node(ProcessorNode node) { } /** - * Request committing the current task's state + * Commit the current task state */ 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 be973e1b30c17..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 volatile boolean running; + private final TopologyBuilder builder; - private final RecordCollector collector; + private final Producer producer; private final Consumer consumer; - private final Map tasks = new HashMap<>(); - private final Metrics metrics; + 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; } /** @@ -139,31 +138,30 @@ public synchronized void run() { } } + /** + * Shutdown this streaming thread. + */ + public synchronized void close() { + running = false; + } + private void shutdown() { log.info("Shutting down a stream thread"); commitAll(time.milliseconds()); - collector.close(); + producer.close(); consumer.close(); removePartitions(); log.info("Stream thread shutdown complete"); } - /** - * Shutdown this streaming thread. - */ - public synchronized void close() { - running = false; - } - 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,49 +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) { @@ -241,7 +243,8 @@ private void maybeCleanState() { } } } - nextStateCleaning = Long.MAX_VALUE; + + lastClean = now; } } @@ -260,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() { @@ -279,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; @@ -294,10 +299,12 @@ 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.processTime.add(new MetricName(group, "process-time-avg-ms"), new Avg()); @@ -314,9 +321,6 @@ public KafkaStreamingMetrics() { this.processorDestruction = metrics.sensor("processor-destruction"); this.processorDestruction.add(new MetricName(group, "processor-destruction"), new Rate(new Count())); - } - } - } From deed5d2f4ffac3be3fc4122d1552b681c0fc65e4 Mon Sep 17 00:00:00 2001 From: Yasuhiro Matsuda Date: Thu, 27 Aug 2015 09:50:35 -0700 Subject: [PATCH 3/7] kstream refactored --- .../streaming/kstream/SlidingWindow.java | 264 ++++++++++++++++++ .../kafka/streaming/kstream/Window.java | 19 +- .../kstream/internals/KStreamImpl.java | 8 +- .../kstream/internals/KStreamJoin.java | 9 + .../internals/ProcessorContextImpl.java | 4 + 5 files changed, 294 insertions(+), 10 deletions(-) create mode 100644 stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindow.java diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindow.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindow.java new file mode 100644 index 0000000000000..ffa366239c39f --- /dev/null +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindow.java @@ -0,0 +1,264 @@ +/** + * 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.kstream; + +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.ByteArraySerializer; +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.kstream.internals.FilteredIterator; +import org.apache.kafka.streaming.kstream.internals.WindowSupport; +import org.apache.kafka.streaming.processor.internals.ProcessorContextImpl; +import org.apache.kafka.streaming.processor.internals.RecordCollector; +import org.apache.kafka.streaming.processor.ProcessorContext; +import org.apache.kafka.streaming.processor.RestoreFunc; +import org.apache.kafka.streaming.processor.internals.Stamped; + +import java.util.HashMap; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.Map; + +public class SlidingWindow implements Window { + private String name; + private final long duration; + private final int maxCount; + private final Serializer keySerializer; + private final Serializer valueSerializer; + private final Deserializer keyDeserializer; + private final Deserializer valueDeserializer; + + public SlidingWindow( + String name, + long duration, + int maxCount, + Serializer keySerializer, + Serializer valueSerializer, + Deserializer keyDeseriaizer, + Deserializer valueDeserializer) { + this.name = name; + this.duration = duration; + this.maxCount = maxCount; + this.keySerializer = keySerializer; + this.valueSerializer = valueSerializer; + this.keyDeserializer = keyDeseriaizer; + this.valueDeserializer = valueDeserializer; + } + + @Override + public String name() { + return name; + } + + @Override + public WindowInstance build() { + return new SlidingWindowInstance(); + } + + public class SlidingWindowInstance extends WindowSupport implements WindowInstance { + private final Object lock = new Object(); + private ProcessorContext context; + private int slotNum; // used as a key for Kafka log compaction + private LinkedList list = new LinkedList(); + private HashMap> map = new HashMap<>(); + + @Override + public void init(ProcessorContext context) { + this.context = context; + RestoreFuncImpl restoreFunc = new RestoreFuncImpl(); + context.register(this, restoreFunc); + + for (ValueList valueList : map.values()) { + valueList.clearDirtyValues(); + } + this.slotNum = restoreFunc.slotNum; + } + + @Override + public Iterator findAfter(K key, final long timestamp) { + return find(key, timestamp, timestamp + duration); + } + + @Override + public Iterator findBefore(K key, final long timestamp) { + return find(key, timestamp - duration, timestamp); + } + + @Override + public Iterator find(K key, final long timestamp) { + return find(key, timestamp - duration, timestamp + duration); + } + + /* + * finds items in the window between startTime and endTime (both inclusive) + */ + private Iterator find(K key, final long startTime, final long endTime) { + final ValueList values = map.get(key); + + if (values == null) { + return null; + } else { + return new FilteredIterator>(values.iterator()) { + @Override + protected V filter(Value item) { + if (startTime <= item.timestamp && item.timestamp <= endTime) + return item.value; + else + return null; + } + }; + } + } + + @Override + public void put(K key, V value, long timestamp) { + synchronized (lock) { + slotNum++; + + list.offerLast(key); + + ValueList values = map.get(key); + if (values == null) { + values = new ValueList<>(); + map.put(key, values); + } + + values.add(slotNum, value, timestamp); + } + evictExcess(); + evictExpired(timestamp - duration); + } + + private void evictExcess() { + while (list.size() > maxCount) { + K oldestKey = list.pollFirst(); + + ValueList values = map.get(oldestKey); + values.removeFirst(); + + if (values.isEmpty()) map.remove(oldestKey); + } + } + + private void evictExpired(long cutoffTime) { + while (true) { + K oldestKey = list.peekFirst(); + + ValueList values = map.get(oldestKey); + Stamped oldestValue = values.first(); + + if (oldestValue.timestamp < cutoffTime) { + list.pollFirst(); + values.removeFirst(); + + if (values.isEmpty()) map.remove(oldestKey); + } else { + break; + } + } + } + + @Override + public String name() { + return name; + } + + @Override + public void flush() { + IntegerSerializer intSerializer = new IntegerSerializer(); + ByteArraySerializer byteArraySerializer = new ByteArraySerializer(); + + RecordCollector collector = ((ProcessorContextImpl) context).recordCollector(); + + for (Map.Entry> entry : map.entrySet()) { + ValueList values = entry.getValue(); + if (values.hasDirtyValues()) { + K key = entry.getKey(); + + byte[] keyBytes = keySerializer.serialize(name, key); + + Iterator> iterator = values.dirtyValueIterator(); + while (iterator.hasNext()) { + Value dirtyValue = iterator.next(); + byte[] slot = intSerializer.serialize("", dirtyValue.slotNum); + byte[] valBytes = valueSerializer.serialize(name, dirtyValue.value); + + byte[] combined = new byte[8 + 4 + keyBytes.length + 4 + valBytes.length]; + + int offset = 0; + offset += putLong(combined, offset, dirtyValue.timestamp); + offset += puts(combined, offset, keyBytes); + offset += puts(combined, offset, valBytes); + + if (offset != combined.length) + throw new IllegalStateException("serialized length does not match"); + + collector.send(new ProducerRecord<>(name, context.id(), slot, combined), byteArraySerializer, byteArraySerializer); + } + values.clearDirtyValues(); + } + } + } + + @Override + public void close() { + // TODO + } + + @Override + public boolean persistent() { + // TODO: should not be persistent, right? + return false; + } + + private class RestoreFuncImpl implements RestoreFunc { + + final IntegerDeserializer intDeserializer; + int slotNum = 0; + + RestoreFuncImpl() { + intDeserializer = new IntegerDeserializer(); + } + + @Override + public void apply(byte[] slot, byte[] bytes) { + + slotNum = intDeserializer.deserialize("", slot); + + int offset = 0; + // timestamp + long timestamp = getLong(bytes, offset); + offset += 8; + // key + int length = getInt(bytes, offset); + offset += 4; + K key = deserialize(bytes, offset, length, name, keyDeserializer); + offset += length; + // value + length = getInt(bytes, offset); + offset += 4; + V value = deserialize(bytes, offset, length, name, valueDeserializer); + + put(key, value, timestamp); + } + } + } + +} 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..b3ec0e2a108e0 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 @@ -22,15 +22,22 @@ import java.util.Iterator; -public interface Window extends StateStore { +public interface Window { - void init(ProcessorContext context); + interface WindowInstance extends StateStore { - Iterator find(K key, long timestamp); + void init(ProcessorContext context); - Iterator findAfter(K key, long timestamp); + Iterator find(K key, long timestamp); - Iterator findBefore(K key, long timestamp); + Iterator findAfter(K key, long timestamp); - void put(K key, V value, long timestamp); + Iterator findBefore(K key, long timestamp); + + void put(K key, V value, long timestamp); + } + + String name(); + + WindowInstance build(); } 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..ba386c7c85cbe 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.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 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..445ece58c435e 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 @@ -20,7 +20,11 @@ import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.ProcessorContext; import org.apache.kafka.streaming.kstream.ValueJoiner; +<<<<<<< HEAD import org.apache.kafka.streaming.kstream.Window; +======= +import org.apache.kafka.streaming.kstream.Window.WindowInstance; +>>>>>>> kstream refactored import org.apache.kafka.streaming.processor.ProcessorFactory; import java.util.Iterator; @@ -70,8 +74,13 @@ public void init(ProcessorContext context) { if (!context.joinable()) throw new IllegalStateException("Streams are not joinable."); +<<<<<<< HEAD final Window window1 = (Window) context.getStateStore(windowName1); final Window window2 = (Window) context.getStateStore(windowName2); +======= + final WindowInstance window1 = (WindowInstance) context.getStateStore(windowName1); + final WindowInstance window2 = (WindowInstance) context.getStateStore(windowName2); +>>>>>>> kstream refactored if (prior) { this.finder1 = new Finder() { 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 a4954b408f806..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 @@ -217,7 +217,11 @@ 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); } From 7afbcbfbe49460f6dd87ac2842bd0e63bf89b4e2 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Thu, 27 Aug 2015 13:11:18 -0700 Subject: [PATCH 4/7] rename factory to def, move some unit test files --- .../kafka/streaming/kstream/KStream.java | 6 +- .../streaming/kstream/SlidingWindowDef.java | 2 +- .../kafka/streaming/kstream/WindowDef.java | 2 +- .../kstream/internals/KStreamBranch.java | 6 +- .../kstream/internals/KStreamFilter.java | 6 +- .../kstream/internals/KStreamFlatMap.java | 6 +- .../internals/KStreamFlatMapValues.java | 6 +- .../kstream/internals/KStreamImpl.java | 7 +- .../kstream/internals/KStreamJoin.java | 18 +- .../kstream/internals/KStreamMap.java | 6 +- .../kstream/internals/KStreamMapValues.java | 6 +- .../kstream/internals/KStreamPassThrough.java | 6 +- .../kstream/internals/KStreamWindow.java | 8 +- .../internals/KStreamWindowedImpl.java | 2 +- ...rocessorFactory.java => ProcessorDef.java} | 4 +- .../streaming/processor/TopologyBuilder.java | 14 +- .../kafka/streaming/FilteredIteratorTest.java | 95 ------- .../kafka/streaming/KStreamBranchTest.java | 90 ------- .../kafka/streaming/KStreamFilterTest.java | 84 ------ .../kafka/streaming/KStreamMapValuesTest.java | 71 ----- .../kafka/streaming/KStreamSourceTest.java | 59 ----- .../kafka/streaming/KStreamWindowedTest.java | 91 ------- .../kafka/streaming/StreamGroupTest.java | 162 ------------ .../internals/KStreamFlatMapTest.java | 80 ------ .../internals/KStreamFlatMapValuesTest.java | 77 ------ .../streaming/internals/KStreamJoinTest.java | 249 ------------------ .../streaming/internals/KStreamMapTest.java | 73 ----- .../internals/FilteredIteratorTest.java | 2 +- .../internals/KStreamBranchTest.java | 2 +- .../internals/KStreamFilterTest.java | 2 +- .../internals}/KStreamFlatMapTest.java | 2 +- .../internals}/KStreamFlatMapValuesTest.java | 2 +- .../internals}/KStreamJoinTest.java | 2 +- .../internals}/KStreamMapTest.java | 2 +- .../internals/KStreamMapValuesTest.java | 2 +- .../internals/KStreamSourceTest.java | 2 +- .../internals/KStreamWindowedTest.java | 2 +- .../internals/StreamGroupTest.java | 2 +- 38 files changed, 60 insertions(+), 1198 deletions(-) rename stream/src/main/java/org/apache/kafka/streaming/processor/{ProcessorFactory.java => ProcessorDef.java} (93%) delete mode 100644 stream/src/test/java/org/apache/kafka/streaming/FilteredIteratorTest.java delete mode 100644 stream/src/test/java/org/apache/kafka/streaming/KStreamBranchTest.java delete mode 100644 stream/src/test/java/org/apache/kafka/streaming/KStreamFilterTest.java delete mode 100644 stream/src/test/java/org/apache/kafka/streaming/KStreamMapValuesTest.java delete mode 100644 stream/src/test/java/org/apache/kafka/streaming/KStreamSourceTest.java delete mode 100644 stream/src/test/java/org/apache/kafka/streaming/KStreamWindowedTest.java delete mode 100644 stream/src/test/java/org/apache/kafka/streaming/StreamGroupTest.java delete mode 100644 stream/src/test/java/org/apache/kafka/streaming/internals/KStreamFlatMapTest.java delete mode 100644 stream/src/test/java/org/apache/kafka/streaming/internals/KStreamFlatMapValuesTest.java delete mode 100644 stream/src/test/java/org/apache/kafka/streaming/internals/KStreamJoinTest.java delete mode 100644 stream/src/test/java/org/apache/kafka/streaming/internals/KStreamMapTest.java rename stream/src/test/java/org/apache/kafka/streaming/{ => kstream}/internals/FilteredIteratorTest.java (98%) rename stream/src/test/java/org/apache/kafka/streaming/{ => kstream}/internals/KStreamBranchTest.java (98%) rename stream/src/test/java/org/apache/kafka/streaming/{ => kstream}/internals/KStreamFilterTest.java (98%) rename stream/src/test/java/org/apache/kafka/streaming/{ => kstream/internals}/KStreamFlatMapTest.java (98%) rename stream/src/test/java/org/apache/kafka/streaming/{ => kstream/internals}/KStreamFlatMapValuesTest.java (98%) rename stream/src/test/java/org/apache/kafka/streaming/{ => kstream/internals}/KStreamJoinTest.java (99%) rename stream/src/test/java/org/apache/kafka/streaming/{ => kstream/internals}/KStreamMapTest.java (98%) rename stream/src/test/java/org/apache/kafka/streaming/{ => kstream}/internals/KStreamMapValuesTest.java (98%) rename stream/src/test/java/org/apache/kafka/streaming/{ => kstream}/internals/KStreamSourceTest.java (97%) rename stream/src/test/java/org/apache/kafka/streaming/{ => kstream}/internals/KStreamWindowedTest.java (98%) rename stream/src/test/java/org/apache/kafka/streaming/{ => processor}/internals/StreamGroupTest.java (99%) 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..6225fd990c9d2 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 define() { return new SlidingWindow(); } 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..454987677a79e 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 define(); } 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..4ecd10cff0a29 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 define() { 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..f7b593e2fc95d 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 define() { 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..ba872b167ef19 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 define() { 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..291a18fa1b529 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 define() { 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 ba386c7c85cbe..bb293f9b87efa 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,7 +19,8 @@ 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.kstream.KeyValueFlatMap; +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; @@ -185,10 +186,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 445ece58c435e..e34d19a4d71fe 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 @@ -20,16 +20,13 @@ import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.ProcessorContext; import org.apache.kafka.streaming.kstream.ValueJoiner; -<<<<<<< HEAD import org.apache.kafka.streaming.kstream.Window; -======= import org.apache.kafka.streaming.kstream.Window.WindowInstance; ->>>>>>> kstream refactored -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); @@ -41,9 +38,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 define() { return processorForOtherStream; } }; @@ -56,7 +53,7 @@ public Processor build() { } @Override - public Processor build() { + public Processor define() { return new KStreamJoinProcessor(); } @@ -74,13 +71,8 @@ public void init(ProcessorContext context) { if (!context.joinable()) throw new IllegalStateException("Streams are not joinable."); -<<<<<<< HEAD final Window window1 = (Window) context.getStateStore(windowName1); final Window window2 = (Window) context.getStateStore(windowName2); -======= - final WindowInstance window1 = (WindowInstance) context.getStateStore(windowName1); - final WindowInstance window2 = (WindowInstance) context.getStateStore(windowName2); ->>>>>>> kstream refactored if (prior) { this.finder1 = new Finder() { 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..f4ad16cebaddd 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 define() { 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..f7dc8027ad1a1 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 define() { 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..7c0863d7c733d 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 define() { 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..b9ea3bce45c7c 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 define() { 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.define(); 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..8f1ef4df60777 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 define(); } 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..8afacfc05f78f 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.define(); 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/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; From 6adb27a82c35d137e73c2581217d417ee1371e09 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Thu, 27 Aug 2015 13:41:16 -0700 Subject: [PATCH 5/7] rename define() to instance() --- .../org/apache/kafka/streaming/kstream/SlidingWindowDef.java | 2 +- .../java/org/apache/kafka/streaming/kstream/WindowDef.java | 2 +- .../kafka/streaming/kstream/internals/KStreamBranch.java | 2 +- .../kafka/streaming/kstream/internals/KStreamFilter.java | 2 +- .../kafka/streaming/kstream/internals/KStreamFlatMap.java | 2 +- .../streaming/kstream/internals/KStreamFlatMapValues.java | 2 +- .../apache/kafka/streaming/kstream/internals/KStreamJoin.java | 4 ++-- .../apache/kafka/streaming/kstream/internals/KStreamMap.java | 2 +- .../kafka/streaming/kstream/internals/KStreamMapValues.java | 2 +- .../kafka/streaming/kstream/internals/KStreamPassThrough.java | 2 +- .../kafka/streaming/kstream/internals/KStreamWindow.java | 4 ++-- .../org/apache/kafka/streaming/processor/ProcessorDef.java | 2 +- .../org/apache/kafka/streaming/processor/TopologyBuilder.java | 2 +- 13 files changed, 15 insertions(+), 15 deletions(-) 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 6225fd990c9d2..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 define() { + public Window instance() { return new SlidingWindow(); } 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 454987677a79e..f1ab71e9622a3 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 define(); + Window instance(); } 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 4ecd10cff0a29..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 @@ -31,7 +31,7 @@ public KStreamBranch(Predicate... predicates) { } @Override - public Processor define() { + 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 f7b593e2fc95d..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 @@ -32,7 +32,7 @@ public KStreamFilter(Predicate predicate, boolean filterOut) { } @Override - public Processor define() { + 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 ba872b167ef19..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 @@ -31,7 +31,7 @@ class KStreamFlatMap implements ProcessorDef { } @Override - public Processor define() { + 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 291a18fa1b529..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 @@ -31,7 +31,7 @@ class KStreamFlatMapValues implements ProcessorDef { } @Override - public Processor define() { + public Processor instance() { return new KStreamFlatMapValuesProcessor(); } 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 e34d19a4d71fe..ce1f2ce022d66 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 @@ -40,7 +40,7 @@ private static abstract class Finder { private Processor processorForOtherStream = null; public final ProcessorDef processorDefForOtherStream = new ProcessorDef() { @Override - public Processor define() { + public Processor instance() { return processorForOtherStream; } }; @@ -53,7 +53,7 @@ public Processor define() { } @Override - public Processor define() { + 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 f4ad16cebaddd..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 @@ -31,7 +31,7 @@ public KStreamMap(KeyValueMapper mapper) { } @Override - public Processor define() { + 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 f7dc8027ad1a1..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 @@ -30,7 +30,7 @@ public KStreamMapValues(ValueMapper mapper) { } @Override - public Processor define() { + 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 7c0863d7c733d..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 @@ -23,7 +23,7 @@ class KStreamPassThrough implements ProcessorDef { @Override - public Processor define() { + 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 b9ea3bce45c7c..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 @@ -36,7 +36,7 @@ public WindowDef window() { } @Override - public Processor define() { + 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.define(); + this.window = windowDef.instance(); this.window.init(context); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorDef.java b/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorDef.java index 8f1ef4df60777..382c95733ebcd 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorDef.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorDef.java @@ -19,5 +19,5 @@ public interface ProcessorDef { - Processor define(); + 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 8afacfc05f78f..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 @@ -56,7 +56,7 @@ public ProcessorNodeFactory(String name, String[] parents, ProcessorDef definiti } public ProcessorNode build() { - Processor processor = definition.define(); + Processor processor = definition.instance(); return new ProcessorNode(name, processor); } } From e44580708e220792b46a91a145330fb97faf9269 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Thu, 27 Aug 2015 14:00:15 -0700 Subject: [PATCH 6/7] rebased --- .../kafka/streaming/kstream/Window.java | 21 +++++++------------ .../kafka/streaming/kstream/WindowDef.java | 4 ++-- .../kstream/internals/KStreamJoin.java | 1 - 3 files changed, 9 insertions(+), 17 deletions(-) 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 b3ec0e2a108e0..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 @@ -22,22 +22,15 @@ import java.util.Iterator; -public interface Window { +public interface Window extends StateStore { - interface WindowInstance extends StateStore { + void init(ProcessorContext context); - void init(ProcessorContext context); + Iterator find(K key, long timestamp); - Iterator find(K key, long timestamp); + Iterator findAfter(K key, long timestamp); - Iterator findAfter(K key, long timestamp); + Iterator findBefore(K key, long timestamp); - Iterator findBefore(K key, long timestamp); - - void put(K key, V value, long timestamp); - } - - String name(); - - WindowInstance build(); -} + 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 f1ab71e9622a3..776171242f7fb 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 instance(); -} + Window build(); +} \ No newline at end of file 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 ce1f2ce022d66..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,7 +21,6 @@ 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.kstream.Window.WindowInstance; import org.apache.kafka.streaming.processor.ProcessorDef; import java.util.Iterator; From dad94e3c40f7603598b5187e75862ed7666fad78 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Thu, 27 Aug 2015 14:05:48 -0700 Subject: [PATCH 7/7] rebased continue --- .../streaming/kstream/SlidingWindow.java | 264 ------------------ .../kafka/streaming/kstream/WindowDef.java | 2 +- .../kstream/internals/KStreamImpl.java | 1 - 3 files changed, 1 insertion(+), 266 deletions(-) delete mode 100644 stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindow.java diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindow.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindow.java deleted file mode 100644 index ffa366239c39f..0000000000000 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindow.java +++ /dev/null @@ -1,264 +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.kstream; - -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.ByteArraySerializer; -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.kstream.internals.FilteredIterator; -import org.apache.kafka.streaming.kstream.internals.WindowSupport; -import org.apache.kafka.streaming.processor.internals.ProcessorContextImpl; -import org.apache.kafka.streaming.processor.internals.RecordCollector; -import org.apache.kafka.streaming.processor.ProcessorContext; -import org.apache.kafka.streaming.processor.RestoreFunc; -import org.apache.kafka.streaming.processor.internals.Stamped; - -import java.util.HashMap; -import java.util.Iterator; -import java.util.LinkedList; -import java.util.Map; - -public class SlidingWindow implements Window { - private String name; - private final long duration; - private final int maxCount; - private final Serializer keySerializer; - private final Serializer valueSerializer; - private final Deserializer keyDeserializer; - private final Deserializer valueDeserializer; - - public SlidingWindow( - String name, - long duration, - int maxCount, - Serializer keySerializer, - Serializer valueSerializer, - Deserializer keyDeseriaizer, - Deserializer valueDeserializer) { - this.name = name; - this.duration = duration; - this.maxCount = maxCount; - this.keySerializer = keySerializer; - this.valueSerializer = valueSerializer; - this.keyDeserializer = keyDeseriaizer; - this.valueDeserializer = valueDeserializer; - } - - @Override - public String name() { - return name; - } - - @Override - public WindowInstance build() { - return new SlidingWindowInstance(); - } - - public class SlidingWindowInstance extends WindowSupport implements WindowInstance { - private final Object lock = new Object(); - private ProcessorContext context; - private int slotNum; // used as a key for Kafka log compaction - private LinkedList list = new LinkedList(); - private HashMap> map = new HashMap<>(); - - @Override - public void init(ProcessorContext context) { - this.context = context; - RestoreFuncImpl restoreFunc = new RestoreFuncImpl(); - context.register(this, restoreFunc); - - for (ValueList valueList : map.values()) { - valueList.clearDirtyValues(); - } - this.slotNum = restoreFunc.slotNum; - } - - @Override - public Iterator findAfter(K key, final long timestamp) { - return find(key, timestamp, timestamp + duration); - } - - @Override - public Iterator findBefore(K key, final long timestamp) { - return find(key, timestamp - duration, timestamp); - } - - @Override - public Iterator find(K key, final long timestamp) { - return find(key, timestamp - duration, timestamp + duration); - } - - /* - * finds items in the window between startTime and endTime (both inclusive) - */ - private Iterator find(K key, final long startTime, final long endTime) { - final ValueList values = map.get(key); - - if (values == null) { - return null; - } else { - return new FilteredIterator>(values.iterator()) { - @Override - protected V filter(Value item) { - if (startTime <= item.timestamp && item.timestamp <= endTime) - return item.value; - else - return null; - } - }; - } - } - - @Override - public void put(K key, V value, long timestamp) { - synchronized (lock) { - slotNum++; - - list.offerLast(key); - - ValueList values = map.get(key); - if (values == null) { - values = new ValueList<>(); - map.put(key, values); - } - - values.add(slotNum, value, timestamp); - } - evictExcess(); - evictExpired(timestamp - duration); - } - - private void evictExcess() { - while (list.size() > maxCount) { - K oldestKey = list.pollFirst(); - - ValueList values = map.get(oldestKey); - values.removeFirst(); - - if (values.isEmpty()) map.remove(oldestKey); - } - } - - private void evictExpired(long cutoffTime) { - while (true) { - K oldestKey = list.peekFirst(); - - ValueList values = map.get(oldestKey); - Stamped oldestValue = values.first(); - - if (oldestValue.timestamp < cutoffTime) { - list.pollFirst(); - values.removeFirst(); - - if (values.isEmpty()) map.remove(oldestKey); - } else { - break; - } - } - } - - @Override - public String name() { - return name; - } - - @Override - public void flush() { - IntegerSerializer intSerializer = new IntegerSerializer(); - ByteArraySerializer byteArraySerializer = new ByteArraySerializer(); - - RecordCollector collector = ((ProcessorContextImpl) context).recordCollector(); - - for (Map.Entry> entry : map.entrySet()) { - ValueList values = entry.getValue(); - if (values.hasDirtyValues()) { - K key = entry.getKey(); - - byte[] keyBytes = keySerializer.serialize(name, key); - - Iterator> iterator = values.dirtyValueIterator(); - while (iterator.hasNext()) { - Value dirtyValue = iterator.next(); - byte[] slot = intSerializer.serialize("", dirtyValue.slotNum); - byte[] valBytes = valueSerializer.serialize(name, dirtyValue.value); - - byte[] combined = new byte[8 + 4 + keyBytes.length + 4 + valBytes.length]; - - int offset = 0; - offset += putLong(combined, offset, dirtyValue.timestamp); - offset += puts(combined, offset, keyBytes); - offset += puts(combined, offset, valBytes); - - if (offset != combined.length) - throw new IllegalStateException("serialized length does not match"); - - collector.send(new ProducerRecord<>(name, context.id(), slot, combined), byteArraySerializer, byteArraySerializer); - } - values.clearDirtyValues(); - } - } - } - - @Override - public void close() { - // TODO - } - - @Override - public boolean persistent() { - // TODO: should not be persistent, right? - return false; - } - - private class RestoreFuncImpl implements RestoreFunc { - - final IntegerDeserializer intDeserializer; - int slotNum = 0; - - RestoreFuncImpl() { - intDeserializer = new IntegerDeserializer(); - } - - @Override - public void apply(byte[] slot, byte[] bytes) { - - slotNum = intDeserializer.deserialize("", slot); - - int offset = 0; - // timestamp - long timestamp = getLong(bytes, offset); - offset += 8; - // key - int length = getInt(bytes, offset); - offset += 4; - K key = deserialize(bytes, offset, length, name, keyDeserializer); - offset += length; - // value - length = getInt(bytes, offset); - offset += 4; - V value = deserialize(bytes, offset, length, name, valueDeserializer); - - put(key, value, timestamp); - } - } - } - -} 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 776171242f7fb..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/KStreamImpl.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamImpl.java index bb293f9b87efa..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,7 +19,6 @@ 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.ProcessorDef; import org.apache.kafka.streaming.processor.TopologyBuilder; import org.apache.kafka.streaming.kstream.KeyValueFlatMap;