diff --git a/stream/src/main/java/org/apache/kafka/streaming/examples/SimpleProcessJob.java b/stream/src/main/java/org/apache/kafka/streaming/examples/SimpleProcessJob.java index e65ab50fb8807..22d6425e7aadf 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/examples/SimpleProcessJob.java +++ b/stream/src/main/java/org/apache/kafka/streaming/examples/SimpleProcessJob.java @@ -18,7 +18,7 @@ package org.apache.kafka.streaming.examples; import org.apache.kafka.streaming.KafkaStreaming; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.TopologyBuilder; import org.apache.kafka.streaming.StreamingConfig; import org.apache.kafka.streaming.processor.ProcessorContext; @@ -29,7 +29,7 @@ public class SimpleProcessJob { - private static class MyProcessor extends KafkaProcessor { + private static class MyProcessor extends Processor { private ProcessorContext context; public MyProcessor(String name) { diff --git a/stream/src/main/java/org/apache/kafka/streaming/examples/StatefulProcessJob.java b/stream/src/main/java/org/apache/kafka/streaming/examples/StatefulProcessJob.java index d1fc9a2270692..6a01551b4972a 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/examples/StatefulProcessJob.java +++ b/stream/src/main/java/org/apache/kafka/streaming/examples/StatefulProcessJob.java @@ -20,7 +20,7 @@ import org.apache.kafka.common.serialization.IntegerDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.streaming.KafkaStreaming; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.TopologyBuilder; import org.apache.kafka.streaming.StreamingConfig; import org.apache.kafka.streaming.processor.ProcessorContext; @@ -33,7 +33,7 @@ public class StatefulProcessJob { - private static class MyProcessor extends KafkaProcessor { + private static class MyProcessor extends Processor { private ProcessorContext context; private KeyValueStore kvStore; 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 271a89bc868c5..fc23e2d6d530d 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,6 @@ import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.Serializer; -import org.apache.kafka.streaming.processor.KafkaProcessor; import org.apache.kafka.streaming.processor.Processor; /** 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 index f5cbdb8b876e7..ad50f436300e3 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindow.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindow.java @@ -25,8 +25,9 @@ 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.RecordCollector; import org.apache.kafka.streaming.processor.RestoreFunc; import org.apache.kafka.streaming.processor.internals.Stamped; @@ -173,7 +174,7 @@ public void flush() { IntegerSerializer intSerializer = new IntegerSerializer(); ByteArraySerializer byteArraySerializer = new ByteArraySerializer(); - RecordCollector collector = context.recordCollector(); + RecordCollector collector = ((ProcessorContextImpl)context).recordCollector(); for (Map.Entry> entry : map.entrySet()) { ValueList values = entry.getValue(); 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 229e07bb6d509..25189f677c994 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,11 +18,11 @@ package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.common.KafkaException; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.ProcessorMetadata; import org.apache.kafka.streaming.kstream.Predicate; -class KStreamBranch extends KafkaProcessor { +class KStreamBranch extends Processor { private final Predicate[] predicates; 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 d4d73c60e0bb9..b467a9d3d0fc3 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 @@ -17,11 +17,11 @@ package org.apache.kafka.streaming.kstream.internals; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.kstream.Predicate; import org.apache.kafka.streaming.processor.ProcessorMetadata; -class KStreamFilter extends KafkaProcessor { +class KStreamFilter extends Processor { private final PredicateOut predicateOut; 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 408c7cff3d8b2..89543b704e986 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 @@ -17,12 +17,12 @@ package org.apache.kafka.streaming.kstream.internals; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.ProcessorMetadata; import org.apache.kafka.streaming.kstream.KeyValue; import org.apache.kafka.streaming.kstream.KeyValueFlatMap; -class KStreamFlatMap extends KafkaProcessor { +class KStreamFlatMap extends Processor { private final KeyValueFlatMap mapper; 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 99a1ba2f69866..0df01e62358ed 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 @@ -17,11 +17,11 @@ package org.apache.kafka.streaming.kstream.internals; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.kstream.ValueMapper; import org.apache.kafka.streaming.processor.ProcessorMetadata; -class KStreamFlatMapValues extends KafkaProcessor { +class KStreamFlatMapValues extends Processor { private final ValueMapper> mapper; 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 619a0af77fe30..14b09c52bceb3 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 @@ -20,7 +20,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.Processor; import org.apache.kafka.streaming.processor.ProcessorMetadata; import org.apache.kafka.streaming.processor.TopologyBuilder; import org.apache.kafka.streaming.kstream.KStreamWindowed; 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 236996710b3a6..7751a7f3603f3 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 @@ -17,14 +17,14 @@ package org.apache.kafka.streaming.kstream.internals; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.ProcessorContext; import org.apache.kafka.streaming.kstream.ValueJoiner; import org.apache.kafka.streaming.kstream.Window; import java.util.Iterator; -class KStreamJoin extends KafkaProcessor { +class KStreamJoin extends Processor { private static final String JOIN_NAME = "KAFKA-JOIN"; private static final String JOIN_OTHER_NAME = "KAFKA-JOIN-OTHER"; @@ -38,7 +38,7 @@ private static abstract class Finder { private final Finder finder1; private final Finder finder2; private final ValueJoiner joiner; - final KafkaProcessor processorForOtherStream; + final Processor processorForOtherStream; private ProcessorContext context; @@ -100,8 +100,8 @@ public void process(K key, V1 value) { } } - private KafkaProcessor processorForOther() { - return new KafkaProcessor(JOIN_OTHER_NAME) { + private Processor processorForOther() { + return new Processor(JOIN_OTHER_NAME) { @SuppressWarnings("unchecked") @Override 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 f6ca1fe35f77a..903fb785a7561 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 @@ -17,13 +17,12 @@ package org.apache.kafka.streaming.kstream.internals; -import org.apache.kafka.streaming.processor.KafkaProcessor; +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.ProcessorContext; import org.apache.kafka.streaming.processor.ProcessorMetadata; -class KStreamMap extends KafkaProcessor { +class KStreamMap extends Processor { private final KeyValueMapper mapper; 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 77d88ebf11004..a8aac1285039e 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 @@ -17,11 +17,11 @@ package org.apache.kafka.streaming.kstream.internals; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.kstream.ValueMapper; import org.apache.kafka.streaming.processor.ProcessorMetadata; -class KStreamMapValues extends KafkaProcessor { +class KStreamMapValues extends Processor { private final ValueMapper mapper; diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamProcessor.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamProcessor.java index 2e1826ec3e795..486b67e3c7625 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamProcessor.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamProcessor.java @@ -17,11 +17,10 @@ package org.apache.kafka.streaming.kstream.internals; -import org.apache.kafka.streaming.processor.KafkaProcessor; import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.ProcessorMetadata; -public class KStreamProcessor extends KafkaProcessor { +public class KStreamProcessor extends Processor { private final Processor processor; diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamSend.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamSend.java index 0da4f899bf2e0..12b5a8ffa2837 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamSend.java +++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamSend.java @@ -18,11 +18,11 @@ package org.apache.kafka.streaming.kstream.internals; import org.apache.kafka.common.serialization.Serializer; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.ProcessorMetadata; import org.apache.kafka.streaming.processor.ProcessorContext; -class KStreamSend extends KafkaProcessor { +class KStreamSend extends Processor { private ProcessorContext context; 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 2288a07485cef..2a6628ab5859a 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 @@ -17,7 +17,7 @@ package org.apache.kafka.streaming.kstream.internals; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.TopologyBuilder; import org.apache.kafka.streaming.processor.ProcessorContext; import org.apache.kafka.streaming.kstream.KStream; @@ -25,7 +25,7 @@ import org.apache.kafka.streaming.kstream.ValueJoiner; import org.apache.kafka.streaming.kstream.Window; -public class KStreamWindow extends KafkaProcessor { +public class KStreamWindow extends Processor { public static final class KStreamWindowedImpl extends KStreamImpl implements KStreamWindowed { diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/KafkaProcessor.java b/stream/src/main/java/org/apache/kafka/streaming/processor/KafkaProcessor.java deleted file mode 100644 index cc0cf943a9848..0000000000000 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/KafkaProcessor.java +++ /dev/null @@ -1,50 +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.processor; - -public abstract class KafkaProcessor implements Processor, Punctuator { - - private final ProcessorMetadata metadata; - - protected ProcessorContext context; - - public KafkaProcessor(ProcessorMetadata metadata) { - this.metadata = metadata; - } - - public ProcessorMetadata metadata() { - return metadata; - } - - /* Following functions can be overridden by users */ - - @Override - public void punctuate(long streamTime) { - // do nothing - } - - @Override - public void init(ProcessorContext context) { - this.context = context; - } - - @Override - public void close() { - // do nothing - } -} diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/Processor.java b/stream/src/main/java/org/apache/kafka/streaming/processor/Processor.java index c3fc267e13076..b94b03f451049 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/Processor.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/Processor.java @@ -23,5 +23,7 @@ public interface Processor { void process(K key, V value); + void punctuate(long streamTime); + void close(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorContext.java b/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorContext.java index 787d91c8f2e5f..15d2e2981eee7 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorContext.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorContext.java @@ -63,13 +63,6 @@ public interface ProcessorContext { */ Deserializer valueDeserializer(); - /** - * Returns a RecordCollector - * - * @return RecordCollector - */ - RecordCollector recordCollector(); - /** * Returns the state directory for the partition. * @@ -91,19 +84,10 @@ public interface ProcessorContext { */ void register(StateStore store, RestoreFunc restoreFunc); - /** - * Flush the local state of this context - */ - void flush(); + void schedule(Processor processor, long interval); void forward(K key, V value); - void send(String topic, Object key, Object value); - - void send(String topic, Object key, Object value, Serializer keySerializer, Serializer valSerializer); - - void schedule(KafkaProcessor processor, long interval); - void commit(); String topic(); diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/Punctuator.java b/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorFactory.java similarity index 92% rename from stream/src/main/java/org/apache/kafka/streaming/processor/Punctuator.java rename to stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorFactory.java index 4619263db3b4c..1d1c4816d8cce 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/Punctuator.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorFactory.java @@ -17,7 +17,7 @@ package org.apache.kafka.streaming.processor; -public interface Punctuator { +public interface ProcessorFactory { - void punctuate(long streamTime); + Processor build(); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/RecordCollector.java b/stream/src/main/java/org/apache/kafka/streaming/processor/RecordCollector.java deleted file mode 100644 index e65b2b93f84ac..0000000000000 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/RecordCollector.java +++ /dev/null @@ -1,30 +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.processor; - -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.Serializer; - -public interface RecordCollector { - - void send(ProducerRecord record); - - void send(ProducerRecord record, Serializer keySerializer, Serializer valueSerializer); - - void flush(); -} 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 d7d55620f2ff4..591b212963b28 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 @@ -42,12 +42,10 @@ public class TopologyBuilder { private Map> children = new HashMap<>(); private class ProcessorClazz { - public Class clazz; - public ProcessorMetadata metadata; + public ProcessorFactory factory; - public ProcessorClazz(Class clazz, ProcessorMetadata metadata) { - this.clazz = clazz; - this.metadata = metadata; + public ProcessorClazz(ProcessorFactory factory) { + this.factory = factory; } } @@ -96,11 +94,11 @@ public final void addSink(String name, Serializer keySerializer, Serializer valS sinkClasses.put(name, new SinkClazz(keySerializer, valSerializer)); } - public final void addProcessor(String name, Class processorClass, ProcessorMetadata config, String... parentNames) { + public final void addProcessor(String name, ProcessorFactory factory, String... parentNames) { if (processorClasses.containsKey(name)) throw new IllegalArgumentException("Processor " + name + " is already added."); - processorClasses.put(name, new ProcessorClazz(processorClass, config)); + processorClasses.put(name, new ProcessorClazz(factory)); if (parentNames != null) { for (String parent : parentNames) { @@ -145,17 +143,16 @@ public ProcessorTopology build() { processorMap.put(name, node); } - // create normal processors + // create processors try { for (String name : processorClasses.keySet()) { - ProcessorMetadata metadata = processorClasses.get(name).metadata; - Class processorClass = processorClasses.get(name).clazz; - KafkaProcessor processor = processorClass.getConstructor(ProcessorMetadata.class).newInstance(metadata); + ProcessorFactory processorFactory = processorClasses.get(name).factory; + Processor processor = processorFactory.build(); ProcessorNode node = new ProcessorNode(name, processor); processorMap.put(name, node); } } catch (Exception e) { - throw new KafkaException("KafkaProcessor(String) constructor failed: this should not happen."); + throw new KafkaException("Processor(String) constructor failed: this should not happen."); } // construct topics to sources map @@ -167,8 +164,8 @@ public ProcessorTopology build() { // construct topics to sinks map for (String topic : topicsToSinkNames.keySet()) { SinkNode node = (SinkNode) processorMap.get(topicsToSourceNames.get(topic)); - node.addTopic(topic); topicSinkMap.put(topic, node); + node.addTopic(topic); } // chain children to parents to build the DAG diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/Ingestor.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/Ingestor.java deleted file mode 100644 index b66808abda579..0000000000000 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/Ingestor.java +++ /dev/null @@ -1,41 +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.processor.internals; - -import org.apache.kafka.common.TopicPartition; - -import java.util.Map; -import java.util.Set; - -public interface Ingestor { - - Set topics(); - - void poll(long timeoutMs); - - void pause(TopicPartition partition); - - void unpause(TopicPartition partition, long offset); - - void commit(Map offsets); - - int numPartitions(String topic); - - void addPartitionStreamToGroup(StreamGroup streamGroup, TopicPartition partition); - -} diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/IngestorImpl.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/IngestorImpl.java deleted file mode 100644 index 886cb5c799bd7..0000000000000 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/IngestorImpl.java +++ /dev/null @@ -1,133 +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.processor.internals; - -import org.apache.kafka.clients.consumer.CommitType; -import org.apache.kafka.clients.consumer.Consumer; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.common.TopicPartition; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Collections; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - -public class IngestorImpl implements Ingestor { - - private static final Logger log = LoggerFactory.getLogger(IngestorImpl.class); - - private final Set topics; - private final Consumer consumer; - private final Set unpaused = new HashSet<>(); - private final Map partitionGroups = new HashMap<>(); - - public IngestorImpl(Consumer consumer, Set topics) { - this.consumer = consumer; - this.topics = Collections.unmodifiableSet(topics); - for (String topic : this.topics) consumer.subscribe(topic); - } - - public void open() { - for (String topic : this.topics) consumer.subscribe(topic); - } - - public void init() { - unpaused.clear(); - unpaused.addAll(consumer.subscriptions()); - } - - @Override - public Set topics() { - return topics; - } - - @Override - public void poll(long timeoutMs) { - synchronized (this) { - ConsumerRecords records = consumer.poll(timeoutMs); - - for (TopicPartition partition : unpaused) { - StreamGroup streamGroup = partitionGroups.get(partition); - - if (streamGroup != null) - streamGroup.addRecords(partition, records.records(partition).iterator()); - else - log.warn("unused topic: " + partition.topic()); - } - } - } - - @Override - public void pause(TopicPartition partition) { - synchronized (this) { - consumer.seek(partition, Long.MAX_VALUE); // hack: stop consuming from this partition by setting a big offset - unpaused.remove(partition); - } - } - - @Override - public void unpause(TopicPartition partition, long lastOffset) { - synchronized (this) { - consumer.seek(partition, lastOffset); - unpaused.add(partition); - } - } - - @Override - public void commit(Map offsets) { - synchronized (this) { - consumer.commit(offsets, CommitType.SYNC); - } - } - - @Override - public int numPartitions(String topic) { - return consumer.partitionsFor(topic).size(); - } - - @SuppressWarnings("unchecked") - @Override - public void addPartitionStreamToGroup(StreamGroup streamGroup, TopicPartition partition) { - synchronized (this) { - partitionGroups.put(partition, streamGroup); - unpaused.add(partition); - } - } - - public void clear() { - unpaused.clear(); - partitionGroups.clear(); - } - - public boolean commitNeeded(Map offsets) { - for (TopicPartition tp : offsets.keySet()) { - if (consumer.committed(tp) != offsets.get(tp)) { - return true; - } - } - return false; - } - - public void close() { - consumer.close(); - clear(); - } -} diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/MinTimestampTracker.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/MinTimestampTracker.java index 3e94622f6612b..c9bc580ece19b 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/MinTimestampTracker.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/MinTimestampTracker.java @@ -20,15 +20,14 @@ import java.util.LinkedList; /** - * MinTimestampTracker is a helper class for a sliding window implementation. - * It is assumed that elements are added or removed in a FIFO manner. - * It maintains the minimum timestamp of stamped elements that were added but not yet removed. + * MinTimestampTracker implements {@link TimestampTracker} that maintains the min + * timestamp of the maintained stamped elements. */ public class MinTimestampTracker implements TimestampTracker { - private final LinkedList> descendingSubsequence = new LinkedList>(); + private final LinkedList> descendingSubsequence = new LinkedList<>(); - public void addStampedElement(Stamped elem) { + public void addElement(Stamped elem) { if (elem == null) throw new NullPointerException(); Stamped minElem = descendingSubsequence.peekLast(); @@ -39,7 +38,7 @@ public void addStampedElement(Stamped elem) { descendingSubsequence.offerLast(elem); } - public void removeStampedElement(Stamped elem) { + public void removeElement(Stamped elem) { if (elem != null && descendingSubsequence.peekFirst() == elem) descendingSubsequence.removeFirst(); } @@ -50,9 +49,11 @@ public int size() { public long get() { Stamped stamped = descendingSubsequence.peekFirst(); - if (stamped == null) return -1L; - return stamped.timestamp; + if (stamped == null) + return TimestampTracker.NOT_KNOWN; + else + return stamped.timestamp; } -} +} \ No newline at end of file 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 c5a530b6666ea..a473c4eff0891 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,8 +22,10 @@ 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. @@ -34,12 +36,13 @@ public class PartitionGroup { private final PriorityQueue queuesByTime; - private volatile int totalBuffered; + // since task is thread-safe, we do not need to synchronize on local variables + private int totalBuffered; public PartitionGroup(Map partitionQueues) { - - this.partitionQueues = partitionQueues; this.queuesByTime = new PriorityQueue<>(new Comparator() { + + @Override public int compare(RecordQueue queue1, RecordQueue queue2) { long time1 = queue1.timestamp(); long time2 = queue2.timestamp(); @@ -50,30 +53,44 @@ public int compare(RecordQueue queue1, RecordQueue queue2) { } }); - totalBuffered = 0; + this.partitionQueues = partitionQueues; + + this.totalBuffered = 0; } /** - * Get the next record from the partition with the lowest timestamp to be processed + * Get one record from the specified partition queue */ - public StampedRecord nextRecord() { - - // Get the partition with the lowest timestamp. - RecordQueue recordQueue = queuesByTime.poll(); + public StampedRecord getRecord(RecordQueue queue) { + // get the first record from this queue. + StampedRecord record = queue.poll(); - // Get the first record from this partition's queue. - StampedRecord record = recordQueue.next(); + // update the partition's timestamp and re-order it against other partitions. - totalBuffered--; + queuesByTime.remove(queue); - // Update the partition's timestamp and re-order it with other partitions. - if (recordQueue.size() > 0) { - queuesByTime.offer(recordQueue); + if (queue.size() > 0) { + queuesByTime.offer(queue); } + totalBuffered--; + return record; } + /** + * Get the next partition queue that has the lowest timestamp to process + */ + 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; + } + /** * Put a timestamped record associated into its corresponding partition's queues. */ @@ -93,8 +110,9 @@ public void putRecord(StampedRecord record, TopicPartition partition) { totalBuffered++; // add this record queue to be considered for processing in the future if it was empty before - if (wasEmpty) + if (wasEmpty) { queuesByTime.offer(recordQueue); + } } public Deserializer keyDeserializer(TopicPartition partition) { @@ -115,13 +133,20 @@ public Deserializer valDeserializer(TopicPartition partition) { return recordQueue.source().valDeserializer; } - public long timestamp() { + public Set partitions() { + return partitionQueues.keySet(); + } - // return the timestamp of this partition-group as the smallest partition timestamp - if (queuesByTime.isEmpty()) + /** + * Return the timestamp of this partition group as the smallest + * partition timestamp among all its partitions + */ + public long timestamp() { + if (queuesByTime.isEmpty()) { return -1L; - else + } else { return queuesByTime.peek().timestamp(); + } } public int numbuffered(TopicPartition partition) { @@ -136,4 +161,9 @@ public int numbuffered(TopicPartition partition) { public int numbuffered() { return totalBuffered; } + + public void close() { + queuesByTime.clear(); + partitionQueues.clear(); + } } 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 dd6d3c48602d3..1691632d879a0 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 @@ -19,7 +19,6 @@ import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.KafkaConsumer; -import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.metrics.Metrics; @@ -28,11 +27,9 @@ import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.streaming.StreamingConfig; import org.apache.kafka.streaming.processor.ProcessorContext; -import org.apache.kafka.streaming.processor.RecordCollector; import org.apache.kafka.streaming.processor.StateStore; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.RestoreFunc; -import org.apache.kafka.streaming.processor.TimestampExtractor; import org.slf4j.Logger; @@ -50,13 +47,10 @@ public class ProcessorContextImpl implements ProcessorContext { private static final Logger log = LoggerFactory.getLogger(ProcessorContextImpl.class); - public final int id; - public final Ingestor ingestor; - public final StreamGroup streamGroup; - + private final int id; + private final StreamTask task; private final Metrics metrics; - private final ProcessorTopology topology; - private final RecordCollectorImpl collector; + private final RecordCollector collector; private final ProcessorStateManager stateMgr; private final Serializer keySerializer; @@ -68,47 +62,35 @@ public class ProcessorContextImpl implements ProcessorContext { @SuppressWarnings("unchecked") public ProcessorContextImpl(int id, - Ingestor ingestor, - ProcessorTopology topology, - RecordCollectorImpl collector, + StreamTask task, StreamingConfig config, + RecordCollector collector, Metrics metrics) throws IOException { this.id = id; + this.task = task; this.metrics = metrics; - this.ingestor = ingestor; - this.topology = topology; this.collector = collector; - for (String topic : this.topology.sourceTopics()) { - if (!ingestor.topics().contains(topic)) - throw new IllegalArgumentException("topic not subscribed: " + topic); - } - this.keySerializer = config.getConfiguredInstance(StreamingConfig.KEY_SERIALIZER_CLASS_CONFIG, Serializer.class); this.valSerializer = config.getConfiguredInstance(StreamingConfig.VALUE_SERIALIZER_CLASS_CONFIG, Serializer.class); this.keyDeserializer = config.getConfiguredInstance(StreamingConfig.KEY_DESERIALIZER_CLASS_CONFIG, Deserializer.class); this.valDeserializer = config.getConfiguredInstance(StreamingConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Deserializer.class); - TimestampExtractor extractor = config.getConfiguredInstance(StreamingConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class); - int bufferedRecordsPerPartition = config.getInt(StreamingConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG); - File stateFile = new File(config.getString(StreamingConfig.STATE_DIR_CONFIG), Integer.toString(id)); - Consumer restoreConsumer = new KafkaConsumer<>(config.getConsumerProperties(), null, new ByteArrayDeserializer(), new ByteArrayDeserializer()); - this.stateMgr = new ProcessorStateManager(id, stateFile, restoreConsumer); - this.streamGroup = new StreamGroup(this, this.ingestor, extractor, bufferedRecordsPerPartition); + Consumer restoreConsumer = new KafkaConsumer<>( + config.getConsumerProperties(), + null /* no callback for restore consumer */, + new ByteArrayDeserializer(), + new ByteArrayDeserializer()); - stateMgr.init(); + this.stateMgr = new ProcessorStateManager(id, stateFile, restoreConsumer); - initialized = false; + this.initialized = false; } - public void addPartition(TopicPartition partition) { - // update the partition -> source stream map - SourceNode source = topology.source(partition.topic()); - - this.streamGroup.addPartition(partition, source); - this.ingestor.addPartitionStreamToGroup(this.streamGroup, partition); + public RecordCollector recordCollector() { + return this.collector; } @Override @@ -116,10 +98,10 @@ public boolean joinable(ProcessorContext o) { ProcessorContextImpl other = (ProcessorContextImpl) o; - if (this.streamGroup != other.streamGroup) + if (this.task != other.task) return false; - Set partitions = this.streamGroup.partitions(); + Set partitions = this.task.partitions(); Map> partitionsById = new HashMap<>(); int firstId = -1; for (TopicPartition partition : partitions) { @@ -171,11 +153,6 @@ public Deserializer valueDeserializer() { return this.valDeserializer; } - @Override - public RecordCollector recordCollector() { - return collector; - } - @Override public File stateDir() { return stateMgr.baseDir(); @@ -195,85 +172,53 @@ public void register(StateStore store, RestoreFunc restoreFunc) { } @Override - public void flush() { - stateMgr.flush(); - } - public String topic() { - if (streamGroup.record() == null) + if (task.record() == null) throw new IllegalStateException("this should not happen as topic() should only be called while a record is processed"); - return streamGroup.record().topic(); + return task.record().topic(); } @Override public int partition() { - if (streamGroup.record() == null) + if (task.record() == null) throw new IllegalStateException("this should not happen as partition() should only be called while a record is processed"); - return streamGroup.record().partition(); + return task.record().partition(); } @Override public long offset() { - if (this.streamGroup.record() == null) + if (this.task.record() == null) throw new IllegalStateException("this should not happen as offset() should only be called while a record is processed"); - return this.streamGroup.record().offset(); + return this.task.record().offset(); } @Override public long timestamp() { - if (streamGroup.record() == null) + if (task.record() == null) throw new IllegalStateException("this should not happen as timestamp() should only be called while a record is processed"); - return streamGroup.record().timestamp; + return task.record().timestamp; } @Override @SuppressWarnings("unchecked") public void forward(K key, V value) { - for (ProcessorNode childNode : (List>) streamGroup.node().children()) { - streamGroup.setNode(childNode); + for (ProcessorNode childNode : (List>) task.node().children()) { + task.node(childNode); childNode.process(key, value); } } - @Override - public void send(String topic, Object key, Object value) { - collector.send(new ProducerRecord<>(topic, key, value)); - } - - @Override - public void send(String topic, Object key, Object value, Serializer keySerializer, Serializer valSerializer) { - if (keySerializer == null || valSerializer == null) - throw new IllegalStateException("key and value serializers must be specified"); - - collector.send(new ProducerRecord<>(topic, key, value), keySerializer, valSerializer); - } - @Override public void commit() { - streamGroup.commitOffset(); + task.commitOffset(); } @Override - public void schedule(KafkaProcessor processor, long interval) { - streamGroup.schedule(processor, interval); + public void schedule(Processor processor, long interval) { + task.schedule(processor, interval); } - - public void initialized() { - initialized = true; - } - - public Map consumedOffsets() { - return streamGroup.consumedOffsets(); - } - - public void close() throws Exception { - topology.close(); - stateMgr.close(collector.offsets()); - streamGroup.close(); - } - } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorNode.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorNode.java index efdf11add7637..167863172149e 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorNode.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorNode.java @@ -17,7 +17,7 @@ package org.apache.kafka.streaming.processor.internals; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.ProcessorContext; import java.util.ArrayList; @@ -29,7 +29,7 @@ public class ProcessorNode { private final List> parents; private final String name; - private final KafkaProcessor processor; + private final Processor processor; public boolean initialized; @@ -37,7 +37,7 @@ public ProcessorNode(String name) { this(name, null); } - public ProcessorNode(String name, KafkaProcessor processor) { + public ProcessorNode(String name, Processor processor) { this.name = name; this.processor = processor; this.parents = new ArrayList<>(); @@ -50,7 +50,7 @@ public String name() { return name; } - public KafkaProcessor processor() { + public Processor processor() { return processor; } 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 58d18ede0f0f5..5052421e0fe9f 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 @@ -44,13 +44,19 @@ public class ProcessorStateManager { private final Map restoredOffsets; private final Map checkpointedOffsets; - public ProcessorStateManager(int id, File baseDir, Consumer restoreConsumer) { + public ProcessorStateManager(int id, File baseDir, Consumer restoreConsumer) throws IOException { this.id = id; this.baseDir = baseDir; this.stores = new HashMap<>(); this.restoreConsumer = restoreConsumer; this.restoredOffsets = new HashMap<>(); this.checkpointedOffsets = new HashMap<>(); + + OffsetCheckpoint checkpoint = new OffsetCheckpoint(new File(this.baseDir, CHECKPOINT_FILE_NAME)); + this.checkpointedOffsets.putAll(checkpoint.read()); + + // delete the checkpoint file after finish loading its stored offsets + checkpoint.delete(); } public File baseDir() { @@ -61,14 +67,6 @@ public Consumer restoreConsumer() { return this.restoreConsumer; } - public void init() throws IOException { - OffsetCheckpoint checkpoint = new OffsetCheckpoint(new File(this.baseDir, CHECKPOINT_FILE_NAME)); - this.checkpointedOffsets.putAll(checkpoint.read()); - - // delete the checkpoint file after finish loading its stored offsets - checkpoint.delete(); - } - 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/ProcessorTopology.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorTopology.java index 364d3d640a5b0..6ae7c3175096d 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorTopology.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorTopology.java @@ -19,6 +19,7 @@ import org.apache.kafka.streaming.processor.ProcessorContext; +import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -54,6 +55,14 @@ public SinkNode sink(String topic) { return sinkTopics.get(topic); } + public Collection sources() { + return sourceTopics.values(); + } + + public Collection sinks() { + return sinkTopics.values(); + } + /** * Initialize the processors following the DAG reverse ordering * such that parents are always initialized before children @@ -93,12 +102,13 @@ private void init(ProcessorNode node, ProcessorContext context) { public final void close() { // close the processors - // TODO: do we need to follow the DAG ordering + // TODO: do we need to follow the DAG ordering? for (ProcessorNode processorNode : processors.values()) { processorNode.close(); } processors.clear(); sourceTopics.clear(); + sinkTopics.clear(); } } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/PunctuationQueue.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/PunctuationQueue.java index 2e7cd77420b83..68eb1bdfba7a8 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/PunctuationQueue.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/PunctuationQueue.java @@ -41,7 +41,7 @@ public void mayPunctuate(long streamTime) { while (top != null && top.timestamp <= streamTime) { PunctuationSchedule sched = top; pq.poll(); - sched.punctuator().punctuate(streamTime); + sched.processor().punctuate(streamTime); pq.add(sched.next()); top = pq.peek(); diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/PunctuationSchedule.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/PunctuationSchedule.java index 6f67e3e09dfe0..0d55876d10580 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/PunctuationSchedule.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/PunctuationSchedule.java @@ -17,18 +17,18 @@ package org.apache.kafka.streaming.processor.internals; -import org.apache.kafka.streaming.processor.Punctuator; +import org.apache.kafka.streaming.processor.Processor; -public class PunctuationSchedule extends Stamped { +public class PunctuationSchedule extends Stamped { final long interval; - public PunctuationSchedule(Punctuator punctuator, long interval) { - super(punctuator, System.currentTimeMillis() + interval); + public PunctuationSchedule(Processor processor, long interval) { + super(processor, System.currentTimeMillis() + interval); this.interval = interval; } - public Punctuator punctuator() { + public Processor processor() { return value; } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/RecordCollectorImpl.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/RecordCollector.java similarity index 90% rename from stream/src/main/java/org/apache/kafka/streaming/processor/internals/RecordCollectorImpl.java rename to stream/src/main/java/org/apache/kafka/streaming/processor/internals/RecordCollector.java index 878f5ce4026bb..1be19aebcd729 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/RecordCollectorImpl.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/RecordCollector.java @@ -17,7 +17,6 @@ package org.apache.kafka.streaming.processor.internals; -import org.apache.kafka.streaming.processor.RecordCollector; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; @@ -30,9 +29,9 @@ import java.util.HashMap; import java.util.Map; -public class RecordCollectorImpl implements RecordCollector { +public class RecordCollector { - private static final Logger log = LoggerFactory.getLogger(RecordCollectorImpl.class); + private static final Logger log = LoggerFactory.getLogger(RecordCollector.class); private final Producer producer; private final Map offsets; @@ -50,26 +49,23 @@ public void onCompletion(RecordMetadata metadata, Exception exception) { private final Serializer valueSerializer; - public RecordCollectorImpl(Producer producer, Serializer keySerializer, Serializer valueSerializer) { + public RecordCollector(Producer producer, Serializer keySerializer, Serializer valueSerializer) { this.producer = producer; this.offsets = new HashMap<>(); this.keySerializer = keySerializer; this.valueSerializer = valueSerializer; } - @Override public void send(ProducerRecord record) { send(record, this.keySerializer, this.valueSerializer); } - @Override public void send(ProducerRecord record, Serializer keySerializer, Serializer valueSerializer) { byte[] keyBytes = keySerializer.serialize(record.topic(), record.key()); byte[] valBytes = valueSerializer.serialize(record.topic(), record.value()); this.producer.send(new ProducerRecord<>(record.topic(), keyBytes, valBytes), callback); } - @Override public void flush() { this.producer.flush(); } 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 aaa656d008f73..93e9420e5bd57 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 @@ -23,16 +23,18 @@ import java.util.ArrayDeque; /** - * RecordQueue is a queue of {@link StampedRecord} (ConsumerRecord + timestamp). + * RecordQueue is a FIFO queue of {@link StampedRecord} (ConsumerRecord + timestamp). It also keeps track of the + * partition timestamp defined as the minimum timestamp of records in its queue; in addition, its partition + * timestamp is monotonically increasing such that once it is advanced, it will not be decremented. */ public class RecordQueue { private final SourceNode source; private final TopicPartition partition; - private final ArrayDeque queue = new ArrayDeque<>(); - private final TimestampTracker> timestampTracker = new MinTimestampTracker<>(); + private final ArrayDeque fifoQueue = new ArrayDeque<>(); + private final TimestampTracker> timeTracker = new MinTimestampTracker<>(); - private long offset; + private long partitionTime = TimestampTracker.NOT_KNOWN; /** * Creates a new instance of RecordQueue @@ -59,39 +61,35 @@ public TopicPartition partition() { } /** - * Adds a StampedRecord to the queue + * Add a {@link StampedRecord} into the queue * * @param record StampedRecord */ public void add(StampedRecord record) { - queue.addLast(record); - - offset = record.offset(); - timestampTracker.addStampedElement(record); + fifoQueue.addLast(record); + timeTracker.addElement(record); } /** - * Returns the next record fro the queue + * Get the next {@link StampedRecord} from the queue * * @return StampedRecord */ - public StampedRecord next() { - StampedRecord elem = queue.pollFirst(); + public StampedRecord poll() { + StampedRecord elem = fifoQueue.pollFirst(); if (elem == null) return null; - timestampTracker.removeStampedElement(elem); + timeTracker.removeElement(elem); - return elem; - } + // only advance the partition timestamp if its currently + // tracked min timestamp has exceeded its value + long timestamp = timeTracker.get(); - /** - * Returns the highest offset in the queue - * - * @return offset - */ - public long offset() { - return offset; + if (timestamp > partitionTime) + partitionTime = timestamp; + + return elem; } /** @@ -100,7 +98,7 @@ public long offset() { * @return the number of records */ public int size() { - return queue.size(); + return fifoQueue.size(); } /** @@ -109,15 +107,15 @@ public int size() { * @return true if the queue is empty, otherwise false */ public boolean isEmpty() { - return queue.isEmpty(); + return fifoQueue.isEmpty(); } /** - * Returns a timestamp tracked by the TimestampTracker + * Returns the tracked partition timestamp * * @return timestamp */ public long timestamp() { - return timestampTracker.get(); + return partitionTime; } } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/SinkNode.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/SinkNode.java index 36ebd3d53fe37..f3cbadf2deda4 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/SinkNode.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/SinkNode.java @@ -20,7 +20,6 @@ import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.streaming.processor.ProcessorContext; -import org.apache.kafka.streaming.processor.RecordCollector; import java.util.ArrayList; import java.util.List; @@ -53,7 +52,7 @@ public void init(ProcessorContext context) { @Override public void process(K key, V value) { // send to all the registered topics - RecordCollector collector = context.recordCollector(); + RecordCollector collector = ((ProcessorContextImpl)context).recordCollector(); for (String topic : topics) { collector.send(new ProducerRecord<>(topic, key, value), keySerializer, valSerializer); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/Stamped.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/Stamped.java index 7a078b681dd4a..dec23759afc12 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/Stamped.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/Stamped.java @@ -27,6 +27,7 @@ public Stamped(V value, long timestamp) { this.timestamp = timestamp; } + @Override public int compareTo(Object other) { long otherTimestamp = ((Stamped) other).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 f5496ec62428e..751f504c8f778 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,87 +17,97 @@ package org.apache.kafka.streaming.processor.internals; -import org.apache.kafka.clients.consumer.Consumer; -import org.apache.kafka.clients.consumer.ConsumerRecord; +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.Punctuator; import org.apache.kafka.streaming.processor.TimestampExtractor; -import java.util.ArrayDeque; +import java.io.IOException; import java.util.Collection; -import java.util.Collections; import java.util.HashMap; import java.util.Iterator; import java.util.Map; import java.util.Set; /** - * A StreamTask is associated with a {@link PartitionGroup} which is assigned to a StreamThread for processing. + * A StreamTask is associated with a {@link PartitionGroup}, and is assigned to a StreamThread for processing. */ public class StreamTask { private final int id; - private final int desiredUnprocessed; + private final int maxBufferedSize; private final Consumer consumer; private final PartitionGroup partitionGroup; + private final PunctuationQueue punctuationQueue; + private final ProcessorContext processorContext; private final TimestampExtractor timestampExtractor; private final Map consumedOffsets; - private final PunctuationQueue punctuationQueue = new PunctuationQueue(); - private final ArrayDeque newRecordBuffer = new ArrayDeque<>(); - private long streamTime = -1; private boolean commitRequested = false; private StampedRecord currRecord = null; private ProcessorNode currNode = null; /** - * Creates StreamGroup + * Create {@link StreamTask} with its assigned partitions * - * @param consumer the instance of {@link Consumer} - * @param partitions the instance of {@link TimestampExtractor} - * @param desiredUnprocessedPerPartition the target number of records kept in a queue for each topic + * @param id the ID of this task + * @param consumer the instance of {@link Consumer} + * @param topology the instance of {@link ProcessorTopology} + * @param partitions the collection of assigned {@link TopicPartition} + * @param config the {@link StreamingConfig} specified by the user */ public StreamTask(int id, Consumer consumer, ProcessorTopology topology, Collection partitions, - int desiredUnprocessedPerPartition) { + RecordCollector collector, + StreamingConfig config) { + this.id = id; this.consumer = consumer; - this.desiredUnprocessed = desiredUnprocessedPerPartition; - this.consumedOffsets = new HashMap<>(); + this.punctuationQueue = new PunctuationQueue(); + this.maxBufferedSize = config.getInt(StreamingConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG); + this.timestampExtractor = config.getConfiguredInstance(StreamingConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class); - // create partition queues and pipe them to corresponding source nodes in the topology + // create queues for each assigned partition and associate them + // to corresponding source nodes in the processor topology Map partitionQueues = new HashMap<>(); + for (TopicPartition partition : partitions) { - RecordQueue queue = createRecordQueue(partition, topology.source(partition.topic())); + SourceNode source = topology.source(partition.topic()); + RecordQueue queue = createRecordQueue(partition, source); partitionQueues.put(partition, queue); } + this.partitionGroup = new PartitionGroup(partitionQueues); - } - public int id() { - return id; - } + // initialize the topology with its own context + try { + this.processorContext = new ProcessorContextImpl(id, this, config, collector, new Metrics()); + } catch (IOException e) { + throw new KafkaException("Error while creating the state manager in processor context."); + } - public StampedRecord record() { - return currRecord; - } + topology.init(this.processorContext); - public ProcessorNode node() { - return currNode; + // initialize the consumed offset cache + this.consumedOffsets = new HashMap<>(); } - public void setNode(ProcessorNode node) { - currNode = node; + public int id() { + return id; } public Set partitions() { - return queuesPerPartition.keySet(); + return this.partitionGroup.partitions(); } /** @@ -131,11 +141,11 @@ public void addRecords(TopicPartition partition, Iterator 0) { + // we can continue processing this task as long as its + // partition group still have buffered records + if (partitionGroup.numbuffered() > 0) { readyForNextExecution = true; } - if (recordQueue.size() == this.desiredUnprocessed) { - ingestor.unpause(recordQueue.partition(), recordQueue.offset()); + // if after processing this record, its partition queue's buffered size has been + // decreased to the threshold, we can then resume the consumption on this partition + if (partitionGroup.numbuffered(queue.partition()) == this.maxBufferedSize) { + consumer.resume(queue.partition()); } - buffered--; - currRecord = null; - - punctuationQueue.mayPunctuate(streamTime); + // possibly trigger registered punctuation functions if + // partition group's time has reached the defined stamp + long timestamp = partitionGroup.timestamp(); + punctuationQueue.mayPunctuate(timestamp); return readyForNextExecution; } } - /** - * Returns consumed offsets - * - * @return the map of partition to consumed offset - */ - public Map consumedOffsets() { - return this.consumedOffsets; + public StampedRecord record() { + return this.currRecord; + } + + public ProcessorNode node() { + return this.currNode; + } + + public void node(ProcessorNode node) { + this.currNode = node; } /** @@ -210,21 +216,11 @@ public void commitOffset() { } public void close() { - queuesByTime.clear(); - queuesPerPartition.clear(); + this.partitionGroup.close(); + this.consumedOffsets.clear(); } protected RecordQueue createRecordQueue(TopicPartition partition, SourceNode source) { return new RecordQueue(partition, source); } - - private static class NewRecords { - final TopicPartition partition; - final Iterator> iterator; - - NewRecords(TopicPartition partition, Iterator> iterator) { - this.partition = partition; - this.iterator = iterator; - } - } } 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 05123a54acd0a..93059746ff999 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 @@ -39,6 +39,7 @@ import org.apache.kafka.common.utils.SystemTime; import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.streaming.processor.TopologyBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -53,8 +54,8 @@ public class StreamThread extends Thread { private static final Logger log = LoggerFactory.getLogger(StreamThread.class); private final Consumer consumer; - private final ProcessorTopology topology; - private final RecordCollectorImpl collector; + private final TopologyBuilder builder; + private final RecordCollector collector; private final Map tasks = new HashMap<>(); private final Metrics metrics; private final Time time; @@ -86,11 +87,11 @@ public void onPartitionsRevoked(Consumer consumer, Collection producer = new KafkaProducer<>(config.getProducerProperties(), new ByteArraySerializer(), new ByteArraySerializer()); - this.collector = new RecordCollectorImpl(producer, + 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)); @@ -214,11 +215,11 @@ private void commitAll(long now) { } // check if commit is really needed, i.e. if all the offsets are already committed - if (ingestor.commitNeeded(commit)) { + 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(); - ingestor.commit(commit); // TODO: can this be async? + consumer.commit(commit); // TODO: can this be async? streamingMetrics.commitTime.record(now - lastCommit); } */ @@ -261,8 +262,8 @@ private void addPartitions(Collection assignment) { if (part.partition() == id) partitionsForTask.add(part); - // creat the task - task = new StreamTask(id, consumer, topology, partitionsForTask, config.getInt(StreamingConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG)); + // create the task + task = new StreamTask(id, consumer, builder.build(), partitionsForTask, collector, config); tasks.put(id, task); } diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/TimestampTracker.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/TimestampTracker.java index 80fc20898ad09..fdd79dc8dfdbc 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/TimestampTracker.java +++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/TimestampTracker.java @@ -19,30 +19,32 @@ /** * TimestampTracker is a helper class for a sliding window implementation. - * It is assumed that elements are added or removed in a FIFO manner. - * It maintains the timestamp, like the min timestamp, the max timestamp, etc. of stamped elements - * that were added but not yet removed. + * It is assumed that stamped elements are added or removed in a FIFO manner. + * It maintains the timestamp, such as the min timestamp, the max timestamp, etc. + * of stamped elements that were added but not yet removed. */ public interface TimestampTracker { + static final long NOT_KNOWN = -1L; + /** * Adds a stamped elements to this tracker. * * @param elem the added element */ - void addStampedElement(Stamped elem); + void addElement(Stamped elem); /** * Removed a stamped elements to this tracker. * * @param elem the removed element */ - void removeStampedElement(Stamped elem); + void removeElement(Stamped elem); /** - * Returns the timestamp + * Returns the current tracked timestamp * - * @return timestamp, or -1L when empty + * @return timestamp, or {@link #NOT_KNOWN} when empty */ long get(); @@ -53,4 +55,4 @@ public interface TimestampTracker { */ int size(); -} +} \ No newline at end of file diff --git a/stream/src/main/java/org/apache/kafka/streaming/state/MeteredKeyValueStore.java b/stream/src/main/java/org/apache/kafka/streaming/state/MeteredKeyValueStore.java index d7664ed5e9b50..93f7c093a7107 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/state/MeteredKeyValueStore.java +++ b/stream/src/main/java/org/apache/kafka/streaming/state/MeteredKeyValueStore.java @@ -18,7 +18,6 @@ package org.apache.kafka.streaming.state; import org.apache.kafka.streaming.processor.ProcessorContext; -import org.apache.kafka.streaming.processor.RecordCollector; import org.apache.kafka.streaming.processor.RestoreFunc; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.MetricName; @@ -32,6 +31,8 @@ import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.common.utils.Time; +import org.apache.kafka.streaming.processor.internals.ProcessorContextImpl; +import org.apache.kafka.streaming.processor.internals.RecordCollector; import java.util.HashSet; import java.util.List; @@ -212,7 +213,7 @@ public void flush() { } private void logChange() { - RecordCollector collector = context.recordCollector(); + RecordCollector collector = ((ProcessorContextImpl)context).recordCollector(); Serializer keySerializer = (Serializer) context.keySerializer(); Serializer valueSerializer = (Serializer) context.valueSerializer(); diff --git a/stream/src/main/java/org/apache/kafka/streaming/state/OffsetCheckpoint.java b/stream/src/main/java/org/apache/kafka/streaming/state/OffsetCheckpoint.java index 65c1ecc97fb8e..68bc35bc42f6f 100644 --- a/stream/src/main/java/org/apache/kafka/streaming/state/OffsetCheckpoint.java +++ b/stream/src/main/java/org/apache/kafka/streaming/state/OffsetCheckpoint.java @@ -43,7 +43,7 @@ * . * <topic_name_n> <partition_n> <offset_n> * - * The first line contains a number designating the format version (currently 0), the next line contains + * The first line contains a number designating the format version (currently 0), the get line contains * a number giving the total number of offsets. Each successive line gives a topic/partition/offset triple * separated by spaces. */ diff --git a/stream/src/test/java/org/apache/kafka/streaming/StreamGroupTest.java b/stream/src/test/java/org/apache/kafka/streaming/StreamGroupTest.java index 58046ff670928..8f43c4cb1c2c8 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/StreamGroupTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/StreamGroupTest.java @@ -25,7 +25,6 @@ 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.MockIngestor; import org.apache.kafka.test.MockProcessorContext; import org.apache.kafka.test.MockSourceNode; import org.junit.Test; diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/MinTimestampTrackerTest.java b/stream/src/test/java/org/apache/kafka/streaming/internals/MinTimestampTrackerTest.java deleted file mode 100644 index 77836967a50b3..0000000000000 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/MinTimestampTrackerTest.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.internals; - -import static org.junit.Assert.assertEquals; - -import org.apache.kafka.streaming.processor.internals.MinTimestampTracker; -import org.apache.kafka.streaming.processor.internals.Stamped; -import org.junit.Test; - -public class MinTimestampTrackerTest { - - private Stamped elem(long timestamp) { - return new Stamped<>("", timestamp); - } - - @SuppressWarnings("unchecked") - @Test - public void testTracking() { - MinTimestampTracker tracker = new MinTimestampTracker(); - - Object[] elems = new Object[]{ - elem(100), elem(101), elem(102), elem(98), elem(99), elem(100) - }; - - int insertionIndex = 0; - int removalIndex = 0; - - // add 100 - tracker.addStampedElement((Stamped) elems[insertionIndex++]); - assertEquals(100L, tracker.get()); - - // add 101 - tracker.addStampedElement((Stamped) elems[insertionIndex++]); - assertEquals(100L, tracker.get()); - - // remove 100 - tracker.removeStampedElement((Stamped) elems[removalIndex++]); - assertEquals(101L, tracker.get()); - - // add 102 - tracker.addStampedElement((Stamped) elems[insertionIndex++]); - assertEquals(101L, tracker.get()); - - // add 98 - tracker.addStampedElement((Stamped) elems[insertionIndex++]); - assertEquals(98L, tracker.get()); - - // add 99 - tracker.addStampedElement((Stamped) elems[insertionIndex++]); - assertEquals(98L, tracker.get()); - - // add 100 - tracker.addStampedElement((Stamped) elems[insertionIndex++]); - assertEquals(98L, tracker.get()); - - // remove 101 - tracker.removeStampedElement((Stamped) elems[removalIndex++]); - assertEquals(98L, tracker.get()); - - // remove 102 - tracker.removeStampedElement((Stamped) elems[removalIndex++]); - assertEquals(98L, tracker.get()); - - // remove 98 - tracker.removeStampedElement((Stamped) elems[removalIndex++]); - assertEquals(99L, tracker.get()); - - // remove 99 - tracker.removeStampedElement((Stamped) elems[removalIndex++]); - assertEquals(100L, tracker.get()); - - // remove 100 - tracker.removeStampedElement((Stamped) elems[removalIndex++]); - assertEquals(-1L, tracker.get()); - - assertEquals(insertionIndex, removalIndex); - } - -} diff --git a/stream/src/test/java/org/apache/kafka/streaming/internals/StreamGroupTest.java b/stream/src/test/java/org/apache/kafka/streaming/internals/StreamGroupTest.java index daf28eb3aee80..496174ad52872 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/internals/StreamGroupTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/internals/StreamGroupTest.java @@ -25,7 +25,6 @@ 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.MockIngestor; import org.apache.kafka.test.MockProcessorContext; import org.apache.kafka.test.MockSourceNode; import org.junit.Test; diff --git a/stream/src/test/java/org/apache/kafka/streaming/MinTimestampTrackerTest.java b/stream/src/test/java/org/apache/kafka/streaming/processor/internals/MinTimestampTrackerTest.java similarity index 63% rename from stream/src/test/java/org/apache/kafka/streaming/MinTimestampTrackerTest.java rename to stream/src/test/java/org/apache/kafka/streaming/processor/internals/MinTimestampTrackerTest.java index 8346b78447fdd..74be3f43122df 100644 --- a/stream/src/test/java/org/apache/kafka/streaming/MinTimestampTrackerTest.java +++ b/stream/src/test/java/org/apache/kafka/streaming/processor/internals/MinTimestampTrackerTest.java @@ -15,12 +15,10 @@ * limitations under the License. */ -package org.apache.kafka.streaming; +package org.apache.kafka.streaming.processor.internals; import static org.junit.Assert.assertEquals; -import org.apache.kafka.streaming.processor.internals.MinTimestampTracker; -import org.apache.kafka.streaming.processor.internals.Stamped; import org.junit.Test; public class MinTimestampTrackerTest { @@ -32,7 +30,7 @@ private Stamped elem(long timestamp) { @SuppressWarnings("unchecked") @Test public void testTracking() { - MinTimestampTracker tracker = new MinTimestampTracker(); + TimestampTracker tracker = new MinTimestampTracker<>(); Object[] elems = new Object[]{ elem(100), elem(101), elem(102), elem(98), elem(99), elem(100) @@ -42,54 +40,54 @@ public void testTracking() { int removalIndex = 0; // add 100 - tracker.addStampedElement((Stamped) elems[insertionIndex++]); + tracker.addElement((Stamped) elems[insertionIndex++]); assertEquals(100L, tracker.get()); // add 101 - tracker.addStampedElement((Stamped) elems[insertionIndex++]); + tracker.addElement((Stamped) elems[insertionIndex++]); assertEquals(100L, tracker.get()); // remove 100 - tracker.removeStampedElement((Stamped) elems[removalIndex++]); + tracker.addElement((Stamped) elems[removalIndex++]); assertEquals(101L, tracker.get()); // add 102 - tracker.addStampedElement((Stamped) elems[insertionIndex++]); + tracker.addElement((Stamped) elems[insertionIndex++]); assertEquals(101L, tracker.get()); // add 98 - tracker.addStampedElement((Stamped) elems[insertionIndex++]); + tracker.addElement((Stamped) elems[insertionIndex++]); assertEquals(98L, tracker.get()); // add 99 - tracker.addStampedElement((Stamped) elems[insertionIndex++]); + tracker.addElement((Stamped) elems[insertionIndex++]); assertEquals(98L, tracker.get()); // add 100 - tracker.addStampedElement((Stamped) elems[insertionIndex++]); + tracker.addElement((Stamped) elems[insertionIndex++]); assertEquals(98L, tracker.get()); // remove 101 - tracker.removeStampedElement((Stamped) elems[removalIndex++]); + tracker.addElement((Stamped) elems[removalIndex++]); assertEquals(98L, tracker.get()); // remove 102 - tracker.removeStampedElement((Stamped) elems[removalIndex++]); + tracker.addElement((Stamped) elems[removalIndex++]); assertEquals(98L, tracker.get()); // remove 98 - tracker.removeStampedElement((Stamped) elems[removalIndex++]); + tracker.addElement((Stamped) elems[removalIndex++]); assertEquals(99L, tracker.get()); // remove 99 - tracker.removeStampedElement((Stamped) elems[removalIndex++]); + tracker.addElement((Stamped) elems[removalIndex++]); assertEquals(100L, tracker.get()); // remove 100 - tracker.removeStampedElement((Stamped) elems[removalIndex++]); + tracker.addElement((Stamped) elems[removalIndex++]); assertEquals(-1L, tracker.get()); assertEquals(insertionIndex, removalIndex); } -} +} \ No newline at end of file diff --git a/stream/src/test/java/org/apache/kafka/test/MockIngestor.java b/stream/src/test/java/org/apache/kafka/test/MockIngestor.java deleted file mode 100644 index 1624feb47ba2a..0000000000000 --- a/stream/src/test/java/org/apache/kafka/test/MockIngestor.java +++ /dev/null @@ -1,72 +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.test; - -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.streaming.processor.internals.Ingestor; -import org.apache.kafka.streaming.processor.internals.StreamGroup; - -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - -public class MockIngestor implements Ingestor { - - private HashMap streamSynchronizers = new HashMap<>(); - - public HashSet paused = new HashSet<>(); - - @Override - public Set topics() { - return null; - } - - @Override - public void poll(long timeoutMs) { - } - - @Override - public void pause(TopicPartition partition) { - paused.add(partition); - } - - @Override - public void unpause(TopicPartition partition, long offset) { - paused.remove(partition); - } - - @Override - public void commit(Map offsets) { /* do nothing */} - - @Override - public int numPartitions(String topic) { - return 1; - } - - @Override - public void addPartitionStreamToGroup(StreamGroup streamGroup, TopicPartition partition) { - streamSynchronizers.put(partition, streamGroup); - } - - public void addRecords(TopicPartition partition, Iterable> records) { - streamSynchronizers.get(partition).addRecords(partition, records.iterator()); - } - -} diff --git a/stream/src/test/java/org/apache/kafka/test/MockProcessor.java b/stream/src/test/java/org/apache/kafka/test/MockProcessor.java index 4b1fba293f9e6..81dbf49a5fe87 100644 --- a/stream/src/test/java/org/apache/kafka/test/MockProcessor.java +++ b/stream/src/test/java/org/apache/kafka/test/MockProcessor.java @@ -17,11 +17,11 @@ package org.apache.kafka.test; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import java.util.ArrayList; -public class MockProcessor extends KafkaProcessor { +public class MockProcessor extends Processor { public final ArrayList processed = new ArrayList<>(); public final ArrayList punctuated = new ArrayList<>(); diff --git a/stream/src/test/java/org/apache/kafka/test/MockProcessorContext.java b/stream/src/test/java/org/apache/kafka/test/MockProcessorContext.java index 206ed391c37a6..7d930b10ac07f 100644 --- a/stream/src/test/java/org/apache/kafka/test/MockProcessorContext.java +++ b/stream/src/test/java/org/apache/kafka/test/MockProcessorContext.java @@ -17,9 +17,8 @@ package org.apache.kafka.test; -import org.apache.kafka.streaming.processor.KafkaProcessor; +import org.apache.kafka.streaming.processor.Processor; import org.apache.kafka.streaming.processor.ProcessorContext; -import org.apache.kafka.streaming.processor.RecordCollector; import org.apache.kafka.streaming.processor.RestoreFunc; import org.apache.kafka.streaming.processor.StateStore; import org.apache.kafka.common.metrics.Metrics; @@ -75,11 +74,6 @@ public Deserializer valueDeserializer() { return deserializer; } - @Override - public RecordCollector recordCollector() { - throw new UnsupportedOperationException("recordCollector() not supported."); - } - @Override public File stateDir() { throw new UnsupportedOperationException("stateDir() not supported."); @@ -96,23 +90,13 @@ public void register(StateStore store, RestoreFunc func) { } @Override - public void flush() { - throw new UnsupportedOperationException("flush() not supported."); - } - - @Override - public void send(String topic, Object key, Object value) { - throw new UnsupportedOperationException("send() not supported."); - } - - @Override - public void send(String topic, Object key, Object value, Serializer keySerializer, Serializer valSerializer) { - throw new UnsupportedOperationException("send() not supported."); + public void schedule(Processor processor, long interval) { + throw new UnsupportedOperationException("schedule() not supported"); } @Override - public void schedule(KafkaProcessor processor, long interval) { - throw new UnsupportedOperationException("schedule() not supported"); + public void forward(K key, V value) { + throw new UnsupportedOperationException("forward() not supported"); } @Override