You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gu...@apache.org on 2017/01/25 03:36:15 UTC
[2/2] kafka git commit: HOTIFX: streams system test do not start up
correctly
HOTIFX: streams system test do not start up correctly
Author: Matthias J. Sax <ma...@confluent.io>
Reviewers: Guozhang Wang, Damian Guy, Eno Thereska
Closes #2428 from mjsax/hotfixSystemTests
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/448c1a41
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/448c1a41
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/448c1a41
Branch: refs/heads/trunk
Commit: 448c1a4114798892bef60d44199a07a86fc4505a
Parents: abe19fe
Author: Matthias J. Sax <ma...@confluent.io>
Authored: Tue Jan 24 19:36:08 2017 -0800
Committer: Guozhang Wang <wa...@gmail.com>
Committed: Tue Jan 24 19:36:08 2017 -0800
----------------------------------------------------------------------
.../kafka/streams/perf/SimpleBenchmark.java | 9 +-
.../streams/smoketest/ShutdownDeadlockTest.java | 95 ---
.../streams/smoketest/SmokeTestClient.java | 217 -------
.../streams/smoketest/SmokeTestDriver.java | 596 -------------------
.../kafka/streams/smoketest/SmokeTestUtil.java | 157 -----
.../streams/smoketest/StreamsSmokeTest.java | 74 ---
.../streams/tests/ShutdownDeadlockTest.java | 95 +++
.../kafka/streams/tests/SmokeTestClient.java | 217 +++++++
.../kafka/streams/tests/SmokeTestDriver.java | 596 +++++++++++++++++++
.../kafka/streams/tests/SmokeTestUtil.java | 157 +++++
.../kafka/streams/tests/StreamsSmokeTest.java | 74 +++
.../streams/streams_simple_benchmark_test.py | 11 +-
.../services/performance/streams_performance.py | 9 +
tests/kafkatest/services/streams.py | 4 +-
14 files changed, 1165 insertions(+), 1146 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/448c1a41/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java b/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
index 90226c1..7a10b79 100644
--- a/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
+++ b/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
@@ -46,6 +46,7 @@ import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.test.TestUtils;
import java.io.File;
import java.util.ArrayList;
@@ -98,7 +99,7 @@ public class SimpleBenchmark {
public static void main(String[] args) throws Exception {
String kafka = args.length > 0 ? args[0] : "localhost:9092";
- String stateDirStr = args.length > 1 ? args[1] : "/tmp/kafka-streams-simple-benchmark";
+ String stateDirStr = args.length > 1 ? args[1] : TestUtils.tempDirectory().getAbsolutePath();
numRecords = args.length > 2 ? Integer.parseInt(args[2]) : 10000000;
endKey = numRecords - 1;
@@ -128,11 +129,11 @@ public class SimpleBenchmark {
// simple stream performance source->cache->store
benchmark.processStreamWithCachedStateStore(SOURCE_TOPIC);
// simple streams performance KSTREAM-KTABLE join
- benchmark.kStreamKTableJoin(JOIN_TOPIC_1_PREFIX + "kStreamKTable", JOIN_TOPIC_2_PREFIX + "kStreamKTable");
+ benchmark.kStreamKTableJoin(JOIN_TOPIC_1_PREFIX + "KStreamKTable", JOIN_TOPIC_2_PREFIX + "KStreamKTable");
// simple streams performance KSTREAM-KSTREAM join
- benchmark.kStreamKStreamJoin(JOIN_TOPIC_1_PREFIX + "kStreamKStream", JOIN_TOPIC_2_PREFIX + "kStreamKStream");
+ benchmark.kStreamKStreamJoin(JOIN_TOPIC_1_PREFIX + "KStreamKStream", JOIN_TOPIC_2_PREFIX + "KStreamKStream");
// simple streams performance KTABLE-KTABLE join
- benchmark.kTableKTableJoin(JOIN_TOPIC_1_PREFIX + "kTableKTable", JOIN_TOPIC_2_PREFIX + "kTableKTable");
+ benchmark.kTableKTableJoin(JOIN_TOPIC_1_PREFIX + "KTableKTable", JOIN_TOPIC_2_PREFIX + "KTableKTable");
}
private Properties setJoinProperties(final String applicationId) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/448c1a41/streams/src/test/java/org/apache/kafka/streams/smoketest/ShutdownDeadlockTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/smoketest/ShutdownDeadlockTest.java b/streams/src/test/java/org/apache/kafka/streams/smoketest/ShutdownDeadlockTest.java
deleted file mode 100644
index 942e0c3..0000000
--- a/streams/src/test/java/org/apache/kafka/streams/smoketest/ShutdownDeadlockTest.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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.streams.smoketest;
-
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.serialization.Serdes;
-import org.apache.kafka.common.serialization.StringSerializer;
-import org.apache.kafka.streams.KafkaStreams;
-import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.kstream.ForeachAction;
-import org.apache.kafka.streams.kstream.KStream;
-import org.apache.kafka.streams.kstream.KStreamBuilder;
-
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
-
-public class ShutdownDeadlockTest {
-
- private final String kafka;
-
- public ShutdownDeadlockTest(final String kafka) {
- this.kafka = kafka;
- }
-
- public void start() {
- final String topic = "source";
- final Properties props = new Properties();
- props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "shouldNotDeadlock");
- props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
- final KStreamBuilder builder = new KStreamBuilder();
- final KStream<String, String> source = builder.stream(Serdes.String(), Serdes.String(), topic);
-
- source.foreach(new ForeachAction<String, String>() {
- @Override
- public void apply(final String key, final String value) {
- throw new RuntimeException("KABOOM!");
- }
- });
- final KafkaStreams streams = new KafkaStreams(builder, props);
- streams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
- @Override
- public void uncaughtException(final Thread t, final Throwable e) {
- System.exit(-1);
- }
- });
-
- Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
- @Override
- public void run() {
- streams.close(5, TimeUnit.SECONDS);
- }
- }));
-
- final Properties producerProps = new Properties();
- producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, "SmokeTest");
- producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
- producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
- producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-
- final KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
- producer.send(new ProducerRecord<>(topic, "a", "a"));
- producer.flush();
-
- streams.start();
-
- synchronized (this) {
- try {
- wait();
- } catch (InterruptedException e) {
- // ignored
- }
- }
-
-
- }
-
-
-
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/448c1a41/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestClient.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestClient.java b/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestClient.java
deleted file mode 100644
index 374cc2f..0000000
--- a/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestClient.java
+++ /dev/null
@@ -1,217 +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.streams.smoketest;
-
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.streams.KafkaStreams;
-import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.kstream.Aggregator;
-import org.apache.kafka.streams.kstream.Initializer;
-import org.apache.kafka.streams.kstream.KGroupedStream;
-import org.apache.kafka.streams.kstream.KStream;
-import org.apache.kafka.streams.kstream.KStreamBuilder;
-import org.apache.kafka.streams.kstream.KTable;
-import org.apache.kafka.streams.kstream.Predicate;
-import org.apache.kafka.streams.kstream.TimeWindows;
-import org.apache.kafka.streams.kstream.ValueJoiner;
-
-import java.io.File;
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
-
-public class SmokeTestClient extends SmokeTestUtil {
-
- private final String kafka;
- private final File stateDir;
- private KafkaStreams streams;
- private Thread thread;
-
- public SmokeTestClient(File stateDir, String kafka) {
- super();
- this.stateDir = stateDir;
- this.kafka = kafka;
- }
-
- public void start() {
- streams = createKafkaStreams(stateDir, kafka);
- streams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
- @Override
- public void uncaughtException(Thread t, Throwable e) {
- e.printStackTrace();
- }
- });
-
- thread = new Thread() {
- public void run() {
- streams.start();
- }
- };
- thread.start();
- }
-
- public void close() {
- streams.close();
- try {
- thread.join();
- } catch (Exception ex) {
- // ignore
- }
- }
-
- private static KafkaStreams createKafkaStreams(File stateDir, String kafka) {
- Properties props = new Properties();
- props.put(StreamsConfig.APPLICATION_ID_CONFIG, "SmokeTest");
- props.put(StreamsConfig.STATE_DIR_CONFIG, stateDir.toString());
- props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
- props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 3);
- props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 2);
- props.put(StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG, 100);
- props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 2);
- props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
- props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-
- KStreamBuilder builder = new KStreamBuilder();
-
- KStream<String, Integer> source = builder.stream(stringSerde, intSerde, "data");
-
- source.to(stringSerde, intSerde, "echo");
-
- KStream<String, Integer> data = source.filter(new Predicate<String, Integer>() {
- @Override
- public boolean test(String key, Integer value) {
- return value == null || value != END;
- }
- });
-
- data.process(SmokeTestUtil.printProcessorSupplier("data"));
-
- // min
- KGroupedStream<String, Integer>
- groupedData =
- data.groupByKey(stringSerde, intSerde);
-
- groupedData.aggregate(
- new Initializer<Integer>() {
- public Integer apply() {
- return Integer.MAX_VALUE;
- }
- },
- new Aggregator<String, Integer, Integer>() {
- @Override
- public Integer apply(String aggKey, Integer value, Integer aggregate) {
- return (value < aggregate) ? value : aggregate;
- }
- },
- TimeWindows.of(TimeUnit.DAYS.toMillis(1)),
- intSerde, "uwin-min"
- ).toStream().map(
- new Unwindow<String, Integer>()
- ).to(stringSerde, intSerde, "min");
-
- KTable<String, Integer> minTable = builder.table(stringSerde, intSerde, "min", "minStoreName");
- minTable.toStream().process(SmokeTestUtil.printProcessorSupplier("min"));
-
- // max
- groupedData.aggregate(
- new Initializer<Integer>() {
- public Integer apply() {
- return Integer.MIN_VALUE;
- }
- },
- new Aggregator<String, Integer, Integer>() {
- @Override
- public Integer apply(String aggKey, Integer value, Integer aggregate) {
- return (value > aggregate) ? value : aggregate;
- }
- },
- TimeWindows.of(TimeUnit.DAYS.toMillis(2)),
- intSerde, "uwin-max"
- ).toStream().map(
- new Unwindow<String, Integer>()
- ).to(stringSerde, intSerde, "max");
-
- KTable<String, Integer> maxTable = builder.table(stringSerde, intSerde, "max", "maxStoreName");
- maxTable.toStream().process(SmokeTestUtil.printProcessorSupplier("max"));
-
- // sum
- groupedData.aggregate(
- new Initializer<Long>() {
- public Long apply() {
- return 0L;
- }
- },
- new Aggregator<String, Integer, Long>() {
- @Override
- public Long apply(String aggKey, Integer value, Long aggregate) {
- return (long) value + aggregate;
- }
- },
- TimeWindows.of(TimeUnit.DAYS.toMillis(2)),
- longSerde, "win-sum"
- ).toStream().map(
- new Unwindow<String, Long>()
- ).to(stringSerde, longSerde, "sum");
-
-
- KTable<String, Long> sumTable = builder.table(stringSerde, longSerde, "sum", "sumStoreName");
- sumTable.toStream().process(SmokeTestUtil.printProcessorSupplier("sum"));
-
- // cnt
- groupedData.count(TimeWindows.of(TimeUnit.DAYS.toMillis(2)), "uwin-cnt")
- .toStream().map(
- new Unwindow<String, Long>()
- ).to(stringSerde, longSerde, "cnt");
-
- KTable<String, Long> cntTable = builder.table(stringSerde, longSerde, "cnt", "cntStoreName");
- cntTable.toStream().process(SmokeTestUtil.printProcessorSupplier("cnt"));
-
- // dif
- maxTable.join(minTable,
- new ValueJoiner<Integer, Integer, Integer>() {
- public Integer apply(Integer value1, Integer value2) {
- return value1 - value2;
- }
- }
- ).to(stringSerde, intSerde, "dif");
-
- // avg
- sumTable.join(
- cntTable,
- new ValueJoiner<Long, Long, Double>() {
- public Double apply(Long value1, Long value2) {
- return (double) value1 / (double) value2;
- }
- }
- ).to(stringSerde, doubleSerde, "avg");
-
- // test repartition
- Agg agg = new Agg();
- cntTable.groupBy(agg.selector(),
- stringSerde,
- longSerde
- ).aggregate(agg.init(),
- agg.adder(),
- agg.remover(),
- longSerde,
- "cntByCnt"
- ).to(stringSerde, longSerde, "tagg");
-
- return new KafkaStreams(builder, props);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/448c1a41/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestDriver.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestDriver.java
deleted file mode 100644
index 33464f7..0000000
--- a/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestDriver.java
+++ /dev/null
@@ -1,596 +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.streams.smoketest;
-
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.clients.producer.Callback;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.clients.producer.RecordMetadata;
-import org.apache.kafka.common.PartitionInfo;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.apache.kafka.common.utils.Utils;
-import org.apache.kafka.test.TestUtils;
-
-import java.io.File;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Random;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-
-public class SmokeTestDriver extends SmokeTestUtil {
-
- public static final int MAX_RECORD_EMPTY_RETRIES = 60;
-
- private static class ValueList {
- public final String key;
- private final int[] values;
- private int index;
-
- ValueList(int min, int max) {
- this.key = min + "-" + max;
-
- this.values = new int[max - min + 1];
- for (int i = 0; i < this.values.length; i++) {
- this.values[i] = min + i;
- }
- // We want to randomize the order of data to test not completely predictable processing order
- // However, values are also use as a timestamp of the record. (TODO: separate data and timestamp)
- // We keep some correlation of time and order. Thus, the shuffling is done with a sliding window
- shuffle(this.values, 10);
-
- this.index = 0;
- }
-
- int next() {
- return (index < values.length) ? values[index++] : -1;
- }
- }
-
- // This main() is not used by the system test. It is intended to be used for local debugging.
- public static void main(String[] args) throws Exception {
- final String kafka = "localhost:9092";
- final String zookeeper = "localhost:2181";
- final File stateDir = TestUtils.tempDirectory();
-
- final int numKeys = 20;
- final int maxRecordsPerKey = 1000;
-
- Thread driver = new Thread() {
- public void run() {
- try {
- Map<String, Set<Integer>> allData = generate(kafka, numKeys, maxRecordsPerKey);
- verify(kafka, allData, maxRecordsPerKey);
- } catch (Exception ex) {
- ex.printStackTrace();
- }
- }
- };
-
- SmokeTestClient streams1 = new SmokeTestClient(createDir(stateDir, "1"), kafka);
- SmokeTestClient streams2 = new SmokeTestClient(createDir(stateDir, "2"), kafka);
- SmokeTestClient streams3 = new SmokeTestClient(createDir(stateDir, "3"), kafka);
- SmokeTestClient streams4 = new SmokeTestClient(createDir(stateDir, "4"), kafka);
-
- System.out.println("starting the driver");
- driver.start();
-
- System.out.println("starting the first and second client");
- streams1.start();
- streams2.start();
-
- sleep(10000);
-
- System.out.println("starting the third client");
- streams3.start();
-
- System.out.println("closing the first client");
- streams1.close();
- System.out.println("closed the first client");
-
- sleep(10000);
-
- System.out.println("starting the forth client");
- streams4.start();
-
- driver.join();
-
- System.out.println("driver stopped");
- streams2.close();
- streams3.close();
- streams4.close();
-
- System.out.println("shutdown");
- }
-
- public static Map<String, Set<Integer>> generate(String kafka, final int numKeys, final int maxRecordsPerKey) throws Exception {
- Properties props = new Properties();
- props.put(ProducerConfig.CLIENT_ID_CONFIG, "SmokeTest");
- props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
- props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
- props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
-
- KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(props);
-
- int numRecordsProduced = 0;
-
- Map<String, Set<Integer>> allData = new HashMap<>();
- ValueList[] data = new ValueList[numKeys];
- for (int i = 0; i < numKeys; i++) {
- data[i] = new ValueList(i, i + maxRecordsPerKey - 1);
- allData.put(data[i].key, new HashSet<Integer>());
- }
- Random rand = new Random();
-
- int remaining = data.length;
-
- while (remaining > 0) {
- int index = rand.nextInt(remaining);
- String key = data[index].key;
- int value = data[index].next();
-
- if (value < 0) {
- remaining--;
- data[index] = data[remaining];
- } else {
-
- ProducerRecord<byte[], byte[]> record =
- new ProducerRecord<>("data", stringSerde.serializer().serialize("", key), intSerde.serializer().serialize("", value));
-
- producer.send(record, new Callback() {
- @Override
- public void onCompletion(final RecordMetadata metadata, final Exception exception) {
- if (exception != null) {
- exception.printStackTrace();
- System.exit(-1);
- }
- }
- });
-
-
- numRecordsProduced++;
- allData.get(key).add(value);
- if (numRecordsProduced % 100 == 0)
- System.out.println(numRecordsProduced + " records produced");
- Utils.sleep(2);
-
- }
- }
- producer.close();
- return Collections.unmodifiableMap(allData);
- }
-
- private static void shuffle(int[] data, int windowSize) {
- Random rand = new Random();
- for (int i = 0; i < data.length; i++) {
- // we shuffle data within windowSize
- int j = rand.nextInt(Math.min(data.length - i, windowSize)) + i;
-
- // swap
- int tmp = data[i];
- data[i] = data[j];
- data[j] = tmp;
- }
- }
-
- public static void verify(String kafka, Map<String, Set<Integer>> allData, int maxRecordsPerKey) {
- Properties props = new Properties();
- props.put(ConsumerConfig.CLIENT_ID_CONFIG, "verifier");
- props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
-
- KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props);
- List<TopicPartition> partitions = getAllPartitions(consumer, "echo", "max", "min", "dif", "sum", "cnt", "avg", "wcnt", "tagg");
- consumer.assign(partitions);
- consumer.seekToBeginning(partitions);
-
- final int recordsGenerated = allData.size() * maxRecordsPerKey;
- int recordsProcessed = 0;
-
- HashMap<String, Integer> max = new HashMap<>();
- HashMap<String, Integer> min = new HashMap<>();
- HashMap<String, Integer> dif = new HashMap<>();
- HashMap<String, Long> sum = new HashMap<>();
- HashMap<String, Long> cnt = new HashMap<>();
- HashMap<String, Double> avg = new HashMap<>();
- HashMap<String, Long> wcnt = new HashMap<>();
- HashMap<String, Long> tagg = new HashMap<>();
-
- HashSet<String> keys = new HashSet<>();
- HashMap<String, Set<Integer>> received = new HashMap<>();
- for (String key : allData.keySet()) {
- keys.add(key);
- received.put(key, new HashSet<Integer>());
- }
- int retry = 0;
- final long start = System.currentTimeMillis();
- while (System.currentTimeMillis() - start < TimeUnit.MINUTES.toMillis(3)) {
- ConsumerRecords<byte[], byte[]> records = consumer.poll(500);
- if (records.isEmpty() && recordsProcessed >= recordsGenerated) {
- if (verifyMin(min, allData, false)
- && verifyMax(max, allData, false)
- && verifyDif(dif, allData, false)
- && verifySum(sum, allData, false)
- && verifyCnt(cnt, allData, false)
- && verifyAvg(avg, allData, false)
- && verifyTAgg(tagg, allData, false)) {
- break;
- }
- if (retry++ > MAX_RECORD_EMPTY_RETRIES) {
- break;
- }
- } else {
- for (ConsumerRecord<byte[], byte[]> record : records) {
- String key = stringSerde.deserializer().deserialize("", record.key());
- switch (record.topic()) {
- case "echo":
- Integer value = intSerde.deserializer().deserialize("", record.value());
- recordsProcessed++;
- if (recordsProcessed % 100 == 0) {
- System.out.println("Echo records processed = " + recordsProcessed);
- }
- received.get(key).add(value);
- break;
- case "min":
- min.put(key, intSerde.deserializer().deserialize("", record.value()));
- break;
- case "max":
- max.put(key, intSerde.deserializer().deserialize("", record.value()));
- break;
- case "dif":
- dif.put(key, intSerde.deserializer().deserialize("", record.value()));
- break;
- case "sum":
- sum.put(key, longSerde.deserializer().deserialize("", record.value()));
- break;
- case "cnt":
- cnt.put(key, longSerde.deserializer().deserialize("", record.value()));
- break;
- case "avg":
- avg.put(key, doubleSerde.deserializer().deserialize("", record.value()));
- break;
- case "wcnt":
- wcnt.put(key, longSerde.deserializer().deserialize("", record.value()));
- break;
- case "tagg":
- tagg.put(key, longSerde.deserializer().deserialize("", record.value()));
- break;
- default:
- System.out.println("unknown topic: " + record.topic());
- }
- }
- }
- }
- consumer.close();
- final long finished = System.currentTimeMillis() - start;
- System.out.println("Verification time=" + finished);
- System.out.println("-------------------");
- System.out.println("Result Verification");
- System.out.println("-------------------");
- System.out.println("recordGenerated=" + recordsGenerated);
- System.out.println("recordProcessed=" + recordsProcessed);
-
- if (recordsProcessed > recordsGenerated) {
- System.out.println("PROCESSED-MORE-THAN-GENERATED");
- } else if (recordsProcessed < recordsGenerated) {
- System.out.println("PROCESSED-LESS-THAN-GENERATED");
- }
-
- boolean success;
- success = allData.equals(received);
-
- if (success) {
- System.out.println("ALL-RECORDS-DELIVERED");
- } else {
- int missedCount = 0;
- for (Map.Entry<String, Set<Integer>> entry : allData.entrySet()) {
- missedCount += received.get(entry.getKey()).size();
- }
- System.out.println("missedRecords=" + missedCount);
- }
-
- success &= verifyMin(min, allData, true);
- success &= verifyMax(max, allData, true);
- success &= verifyDif(dif, allData, true);
- success &= verifySum(sum, allData, true);
- success &= verifyCnt(cnt, allData, true);
- success &= verifyAvg(avg, allData, true);
- success &= verifyTAgg(tagg, allData, true);
-
- System.out.println(success ? "SUCCESS" : "FAILURE");
- }
-
- private static boolean verifyMin(Map<String, Integer> map, Map<String, Set<Integer>> allData, final boolean print) {
- if (map.isEmpty()) {
- if (print) {
- System.out.println("min is empty");
- }
- return false;
- } else {
- if (print) {
- System.out.println("verifying min");
- }
-
- if (map.size() != allData.size()) {
- if (print) {
- System.out.println("fail: resultCount=" + map.size() + " expectedCount=" + allData.size());
- }
- return false;
- }
- for (Map.Entry<String, Integer> entry : map.entrySet()) {
- int expected = getMin(entry.getKey());
- if (expected != entry.getValue()) {
- if (print) {
- System.out.println("fail: key=" + entry.getKey() + " min=" + entry.getValue() + " expected=" + expected);
- }
- return false;
- }
- }
- }
- return true;
- }
-
- private static boolean verifyMax(Map<String, Integer> map, Map<String, Set<Integer>> allData, final boolean print) {
- if (map.isEmpty()) {
- if (print) {
- System.out.println("max is empty");
- }
- return false;
- } else {
- if (print) {
- System.out.println("verifying max");
- }
-
- if (map.size() != allData.size()) {
- if (print) {
- System.out.println("fail: resultCount=" + map.size() + " expectedCount=" + allData.size());
- }
- return false;
- }
- for (Map.Entry<String, Integer> entry : map.entrySet()) {
- int expected = getMax(entry.getKey());
- if (expected != entry.getValue()) {
- if (print) {
- System.out.println("fail: key=" + entry.getKey() + " max=" + entry.getValue() + " expected=" + expected);
- }
- return false;
- }
- }
- }
- return true;
- }
-
- private static boolean verifyDif(Map<String, Integer> map, Map<String, Set<Integer>> allData, final boolean print) {
- if (map.isEmpty()) {
- if (print) {
- System.out.println("dif is empty");
- }
- return false;
- } else {
- if (print) {
- System.out.println("verifying dif");
- }
-
- if (map.size() != allData.size()) {
- if (print) {
- System.out.println("fail: resultCount=" + map.size() + " expectedCount=" + allData.size());
- }
- return false;
- }
- for (Map.Entry<String, Integer> entry : map.entrySet()) {
- int min = getMin(entry.getKey());
- int max = getMax(entry.getKey());
- int expected = max - min;
- if (entry.getValue() == null || expected != entry.getValue()) {
- if (print) {
- System.out.println("fail: key=" + entry.getKey() + " dif=" + entry.getValue() + " expected=" + expected);
- }
- return false;
- }
- }
- }
- return true;
- }
-
- private static boolean verifyCnt(Map<String, Long> map, Map<String, Set<Integer>> allData, final boolean print) {
- if (map.isEmpty()) {
- if (print) {
- System.out.println("cnt is empty");
- }
- return false;
- } else {
- if (print) {
- System.out.println("verifying cnt");
- }
-
- if (map.size() != allData.size()) {
- if (print) {
- System.out.println("fail: resultCount=" + map.size() + " expectedCount=" + allData.size());
- }
- return false;
- }
- for (Map.Entry<String, Long> entry : map.entrySet()) {
- int min = getMin(entry.getKey());
- int max = getMax(entry.getKey());
- long expected = (max - min) + 1L;
- if (expected != entry.getValue()) {
- if (print) {
- System.out.println("fail: key=" + entry.getKey() + " cnt=" + entry.getValue() + " expected=" + expected);
- }
- return false;
- }
- }
- }
- return true;
- }
-
- private static boolean verifySum(Map<String, Long> map, Map<String, Set<Integer>> allData, final boolean print) {
- if (map.isEmpty()) {
- if (print) {
- System.out.println("sum is empty");
- }
- return false;
- } else {
- if (print) {
- System.out.println("verifying sum");
- }
-
- if (map.size() != allData.size()) {
- if (print) {
- System.out.println("fail: resultCount=" + map.size() + " expectedCount=" + allData.size());
- }
- return false;
- }
- for (Map.Entry<String, Long> entry : map.entrySet()) {
- int min = getMin(entry.getKey());
- int max = getMax(entry.getKey());
- long expected = ((long) min + (long) max) * (max - min + 1L) / 2L;
- if (expected != entry.getValue()) {
- if (print) {
- System.out.println("fail: key=" + entry.getKey() + " sum=" + entry.getValue() + " expected=" + expected);
- }
- return false;
- }
- }
- }
- return true;
- }
-
- private static boolean verifyAvg(Map<String, Double> map, Map<String, Set<Integer>> allData, final boolean print) {
- if (map.isEmpty()) {
- if (print) {
- System.out.println("avg is empty");
- }
- return false;
- } else {
- if (print) {
- System.out.println("verifying avg");
- }
-
- if (map.size() != allData.size()) {
- if (print) {
- System.out.println("fail: resultCount=" + map.size() + " expectedCount=" + allData.size());
- }
- return false;
- }
- for (Map.Entry<String, Double> entry : map.entrySet()) {
- int min = getMin(entry.getKey());
- int max = getMax(entry.getKey());
- double expected = ((long) min + (long) max) / 2.0;
-
- if (entry.getValue() == null || expected != entry.getValue()) {
- if (print) {
- System.out.println("fail: key=" + entry.getKey() + " avg=" + entry.getValue() + " expected=" + expected);
- }
- return false;
- }
- }
- }
- return true;
- }
-
-
- private static boolean verifyTAgg(Map<String, Long> map, Map<String, Set<Integer>> allData, final boolean print) {
- if (map.isEmpty()) {
- if (print) {
- System.out.println("tagg is empty");
- }
- return false;
- } else {
- if (print) {
- System.out.println("verifying tagg");
- }
-
- // generate expected answer
- Map<String, Long> expected = new HashMap<>();
- for (String key : allData.keySet()) {
- int min = getMin(key);
- int max = getMax(key);
- String cnt = Long.toString(max - min + 1L);
-
- if (expected.containsKey(cnt)) {
- expected.put(cnt, expected.get(cnt) + 1L);
- } else {
- expected.put(cnt, 1L);
- }
- }
-
- // check the result
- for (Map.Entry<String, Long> entry : map.entrySet()) {
- String key = entry.getKey();
- Long expectedCount = expected.remove(key);
- if (expectedCount == null)
- expectedCount = 0L;
-
- if (entry.getValue() != expectedCount) {
- if (print) {
- System.out.println("fail: key=" + key + " tagg=" + entry.getValue() + " expected=" + expected.get(key));
- }
- return false;
- }
- }
-
- }
- return true;
- }
-
- private static int getMin(String key) {
- return Integer.parseInt(key.split("-")[0]);
- }
-
- private static int getMax(String key) {
- return Integer.parseInt(key.split("-")[1]);
- }
-
- private static int getMinFromWKey(String key) {
- return getMin(key.split("@")[0]);
- }
-
- private static int getMaxFromWKey(String key) {
- return getMax(key.split("@")[0]);
- }
-
- private static long getStartFromWKey(String key) {
- return Long.parseLong(key.split("@")[1]);
- }
-
- private static List<TopicPartition> getAllPartitions(KafkaConsumer<?, ?> consumer, String... topics) {
- ArrayList<TopicPartition> partitions = new ArrayList<>();
-
- for (String topic : topics) {
- for (PartitionInfo info : consumer.partitionsFor(topic)) {
- partitions.add(new TopicPartition(info.topic(), info.partition()));
- }
- }
- return partitions;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/448c1a41/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestUtil.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestUtil.java b/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestUtil.java
deleted file mode 100644
index 36660fb..0000000
--- a/streams/src/test/java/org/apache/kafka/streams/smoketest/SmokeTestUtil.java
+++ /dev/null
@@ -1,157 +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.streams.smoketest;
-
-import org.apache.kafka.common.serialization.Serde;
-import org.apache.kafka.common.serialization.Serdes;
-import org.apache.kafka.streams.KeyValue;
-import org.apache.kafka.streams.kstream.Aggregator;
-import org.apache.kafka.streams.kstream.Initializer;
-import org.apache.kafka.streams.kstream.KeyValueMapper;
-import org.apache.kafka.streams.kstream.Windowed;
-import org.apache.kafka.streams.processor.AbstractProcessor;
-import org.apache.kafka.streams.processor.Processor;
-import org.apache.kafka.streams.processor.ProcessorContext;
-import org.apache.kafka.streams.processor.ProcessorSupplier;
-
-import java.io.File;
-
-public class SmokeTestUtil {
-
- public final static int WINDOW_SIZE = 100;
- public final static long START_TIME = 60000L * 60 * 24 * 365 * 30;
- public final static int END = Integer.MAX_VALUE;
-
- public static ProcessorSupplier<Object, Object> printProcessorSupplier(final String topic) {
- return printProcessorSupplier(topic, false);
- }
-
- public static ProcessorSupplier<Object, Object> printProcessorSupplier(final String topic, final boolean printOffset) {
- return new ProcessorSupplier<Object, Object>() {
- public Processor<Object, Object> get() {
- return new AbstractProcessor<Object, Object>() {
- private int numRecordsProcessed = 0;
- private ProcessorContext context;
-
- @Override
- public void init(ProcessorContext context) {
- System.out.println("initializing processor: topic=" + topic + " taskId=" + context.taskId());
- numRecordsProcessed = 0;
- this.context = context;
- }
-
- @Override
- public void process(Object key, Object value) {
- if (printOffset) System.out.println(">>> " + context.offset());
- numRecordsProcessed++;
- if (numRecordsProcessed % 100 == 0) {
- System.out.println("processed " + numRecordsProcessed + " records from topic=" + topic);
- }
- }
-
- @Override
- public void punctuate(long timestamp) {
- }
-
- @Override
- public void close() {
- }
- };
- }
- };
- }
-
- public static final class Unwindow<K, V> implements KeyValueMapper<Windowed<K>, V, KeyValue<K, V>> {
- public KeyValue<K, V> apply(Windowed<K> winKey, V value) {
- return new KeyValue<K, V>(winKey.key(), value);
- }
- }
-
- public static class Agg {
-
- public KeyValueMapper<String, Long, KeyValue<String, Long>> selector() {
- return new KeyValueMapper<String, Long, KeyValue<String, Long>>() {
- @Override
- public KeyValue<String, Long> apply(String key, Long value) {
- return new KeyValue<>(value == null ? null : Long.toString(value), 1L);
- }
- };
- }
-
- public Initializer<Long> init() {
- return new Initializer<Long>() {
- @Override
- public Long apply() {
- return 0L;
- }
- };
- }
-
- public Aggregator<String, Long, Long> adder() {
- return new Aggregator<String, Long, Long>() {
- @Override
- public Long apply(String aggKey, Long value, Long aggregate) {
- return aggregate + value;
- }
- };
- }
-
- public Aggregator<String, Long, Long> remover() {
- return new Aggregator<String, Long, Long>() {
- @Override
- public Long apply(String aggKey, Long value, Long aggregate) {
- return aggregate - value;
- }
- };
- }
- }
-
- public static Serde<String> stringSerde = Serdes.String();
-
- public static Serde<Integer> intSerde = Serdes.Integer();
-
- public static Serde<Long> longSerde = Serdes.Long();
-
- public static Serde<Double> doubleSerde = Serdes.Double();
-
- public static File createDir(String path) throws Exception {
- File dir = new File(path);
-
- dir.mkdir();
-
- return dir;
- }
-
- public static File createDir(File parent, String child) throws Exception {
- File dir = new File(parent, child);
-
- dir.mkdir();
-
- return dir;
- }
-
- public static void sleep(long duration) {
- try {
- Thread.sleep(duration);
- } catch (Exception ex) {
- //
- }
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/448c1a41/streams/src/test/java/org/apache/kafka/streams/smoketest/StreamsSmokeTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/smoketest/StreamsSmokeTest.java b/streams/src/test/java/org/apache/kafka/streams/smoketest/StreamsSmokeTest.java
deleted file mode 100644
index 31dd325..0000000
--- a/streams/src/test/java/org/apache/kafka/streams/smoketest/StreamsSmokeTest.java
+++ /dev/null
@@ -1,74 +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.streams.smoketest;
-
-import java.io.File;
-import java.util.Map;
-import java.util.Set;
-
-public class StreamsSmokeTest {
-
- /**
- * args ::= command kafka zookeeper stateDir
- * command := "run" | "process"
- *
- * @param args
- */
- public static void main(String[] args) throws Exception {
- String command = args[0];
- String kafka = args.length > 1 ? args[1] : null;
- String stateDir = args.length > 2 ? args[2] : null;
-
- System.out.println("StreamsTest instance started");
- System.out.println("command=" + command);
- System.out.println("kafka=" + kafka);
- System.out.println("stateDir=" + stateDir);
-
- switch (command) {
- case "standalone":
- SmokeTestDriver.main(args);
- break;
- case "run":
- // this starts the driver (data generation and result verification)
- final int numKeys = 10;
- final int maxRecordsPerKey = 500;
- Map<String, Set<Integer>> allData = SmokeTestDriver.generate(kafka, numKeys, maxRecordsPerKey);
- SmokeTestDriver.verify(kafka, allData, maxRecordsPerKey);
- break;
- case "process":
- // this starts a KafkaStreams client
- final SmokeTestClient client = new SmokeTestClient(new File(stateDir), kafka);
- client.start();
-
- Runtime.getRuntime().addShutdownHook(new Thread() {
- @Override
- public void run() {
- client.close();
- }
- });
- break;
- case "close-deadlock-test":
- final ShutdownDeadlockTest test = new ShutdownDeadlockTest(kafka);
- test.start();
- break;
- default:
- System.out.println("unknown command: " + command);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/448c1a41/streams/src/test/java/org/apache/kafka/streams/tests/ShutdownDeadlockTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/ShutdownDeadlockTest.java b/streams/src/test/java/org/apache/kafka/streams/tests/ShutdownDeadlockTest.java
new file mode 100644
index 0000000..e06e034
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/tests/ShutdownDeadlockTest.java
@@ -0,0 +1,95 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.streams.tests;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.kstream.ForeachAction;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KStreamBuilder;
+
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+public class ShutdownDeadlockTest {
+
+ private final String kafka;
+
+ public ShutdownDeadlockTest(final String kafka) {
+ this.kafka = kafka;
+ }
+
+ public void start() {
+ final String topic = "source";
+ final Properties props = new Properties();
+ props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "shouldNotDeadlock");
+ props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
+ final KStreamBuilder builder = new KStreamBuilder();
+ final KStream<String, String> source = builder.stream(Serdes.String(), Serdes.String(), topic);
+
+ source.foreach(new ForeachAction<String, String>() {
+ @Override
+ public void apply(final String key, final String value) {
+ throw new RuntimeException("KABOOM!");
+ }
+ });
+ final KafkaStreams streams = new KafkaStreams(builder, props);
+ streams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+ @Override
+ public void uncaughtException(final Thread t, final Throwable e) {
+ System.exit(-1);
+ }
+ });
+
+ Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+ @Override
+ public void run() {
+ streams.close(5, TimeUnit.SECONDS);
+ }
+ }));
+
+ final Properties producerProps = new Properties();
+ producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, "SmokeTest");
+ producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
+ producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+
+ final KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
+ producer.send(new ProducerRecord<>(topic, "a", "a"));
+ producer.flush();
+
+ streams.start();
+
+ synchronized (this) {
+ try {
+ wait();
+ } catch (InterruptedException e) {
+ // ignored
+ }
+ }
+
+
+ }
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/448c1a41/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java
new file mode 100644
index 0000000..35f1883
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java
@@ -0,0 +1,217 @@
+/**
+ * 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.streams.tests;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.kstream.Aggregator;
+import org.apache.kafka.streams.kstream.Initializer;
+import org.apache.kafka.streams.kstream.KGroupedStream;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KStreamBuilder;
+import org.apache.kafka.streams.kstream.KTable;
+import org.apache.kafka.streams.kstream.Predicate;
+import org.apache.kafka.streams.kstream.TimeWindows;
+import org.apache.kafka.streams.kstream.ValueJoiner;
+
+import java.io.File;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+public class SmokeTestClient extends SmokeTestUtil {
+
+ private final String kafka;
+ private final File stateDir;
+ private KafkaStreams streams;
+ private Thread thread;
+
+ public SmokeTestClient(File stateDir, String kafka) {
+ super();
+ this.stateDir = stateDir;
+ this.kafka = kafka;
+ }
+
+ public void start() {
+ streams = createKafkaStreams(stateDir, kafka);
+ streams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+ @Override
+ public void uncaughtException(Thread t, Throwable e) {
+ e.printStackTrace();
+ }
+ });
+
+ thread = new Thread() {
+ public void run() {
+ streams.start();
+ }
+ };
+ thread.start();
+ }
+
+ public void close() {
+ streams.close();
+ try {
+ thread.join();
+ } catch (Exception ex) {
+ // ignore
+ }
+ }
+
+ private static KafkaStreams createKafkaStreams(File stateDir, String kafka) {
+ Properties props = new Properties();
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, "SmokeTest");
+ props.put(StreamsConfig.STATE_DIR_CONFIG, stateDir.toString());
+ props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
+ props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 3);
+ props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 2);
+ props.put(StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG, 100);
+ props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 2);
+ props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
+ props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+ KStreamBuilder builder = new KStreamBuilder();
+
+ KStream<String, Integer> source = builder.stream(stringSerde, intSerde, "data");
+
+ source.to(stringSerde, intSerde, "echo");
+
+ KStream<String, Integer> data = source.filter(new Predicate<String, Integer>() {
+ @Override
+ public boolean test(String key, Integer value) {
+ return value == null || value != END;
+ }
+ });
+
+ data.process(SmokeTestUtil.printProcessorSupplier("data"));
+
+ // min
+ KGroupedStream<String, Integer>
+ groupedData =
+ data.groupByKey(stringSerde, intSerde);
+
+ groupedData.aggregate(
+ new Initializer<Integer>() {
+ public Integer apply() {
+ return Integer.MAX_VALUE;
+ }
+ },
+ new Aggregator<String, Integer, Integer>() {
+ @Override
+ public Integer apply(String aggKey, Integer value, Integer aggregate) {
+ return (value < aggregate) ? value : aggregate;
+ }
+ },
+ TimeWindows.of(TimeUnit.DAYS.toMillis(1)),
+ intSerde, "uwin-min"
+ ).toStream().map(
+ new Unwindow<String, Integer>()
+ ).to(stringSerde, intSerde, "min");
+
+ KTable<String, Integer> minTable = builder.table(stringSerde, intSerde, "min", "minStoreName");
+ minTable.toStream().process(SmokeTestUtil.printProcessorSupplier("min"));
+
+ // max
+ groupedData.aggregate(
+ new Initializer<Integer>() {
+ public Integer apply() {
+ return Integer.MIN_VALUE;
+ }
+ },
+ new Aggregator<String, Integer, Integer>() {
+ @Override
+ public Integer apply(String aggKey, Integer value, Integer aggregate) {
+ return (value > aggregate) ? value : aggregate;
+ }
+ },
+ TimeWindows.of(TimeUnit.DAYS.toMillis(2)),
+ intSerde, "uwin-max"
+ ).toStream().map(
+ new Unwindow<String, Integer>()
+ ).to(stringSerde, intSerde, "max");
+
+ KTable<String, Integer> maxTable = builder.table(stringSerde, intSerde, "max", "maxStoreName");
+ maxTable.toStream().process(SmokeTestUtil.printProcessorSupplier("max"));
+
+ // sum
+ groupedData.aggregate(
+ new Initializer<Long>() {
+ public Long apply() {
+ return 0L;
+ }
+ },
+ new Aggregator<String, Integer, Long>() {
+ @Override
+ public Long apply(String aggKey, Integer value, Long aggregate) {
+ return (long) value + aggregate;
+ }
+ },
+ TimeWindows.of(TimeUnit.DAYS.toMillis(2)),
+ longSerde, "win-sum"
+ ).toStream().map(
+ new Unwindow<String, Long>()
+ ).to(stringSerde, longSerde, "sum");
+
+
+ KTable<String, Long> sumTable = builder.table(stringSerde, longSerde, "sum", "sumStoreName");
+ sumTable.toStream().process(SmokeTestUtil.printProcessorSupplier("sum"));
+
+ // cnt
+ groupedData.count(TimeWindows.of(TimeUnit.DAYS.toMillis(2)), "uwin-cnt")
+ .toStream().map(
+ new Unwindow<String, Long>()
+ ).to(stringSerde, longSerde, "cnt");
+
+ KTable<String, Long> cntTable = builder.table(stringSerde, longSerde, "cnt", "cntStoreName");
+ cntTable.toStream().process(SmokeTestUtil.printProcessorSupplier("cnt"));
+
+ // dif
+ maxTable.join(minTable,
+ new ValueJoiner<Integer, Integer, Integer>() {
+ public Integer apply(Integer value1, Integer value2) {
+ return value1 - value2;
+ }
+ }
+ ).to(stringSerde, intSerde, "dif");
+
+ // avg
+ sumTable.join(
+ cntTable,
+ new ValueJoiner<Long, Long, Double>() {
+ public Double apply(Long value1, Long value2) {
+ return (double) value1 / (double) value2;
+ }
+ }
+ ).to(stringSerde, doubleSerde, "avg");
+
+ // test repartition
+ Agg agg = new Agg();
+ cntTable.groupBy(agg.selector(),
+ stringSerde,
+ longSerde
+ ).aggregate(agg.init(),
+ agg.adder(),
+ agg.remover(),
+ longSerde,
+ "cntByCnt"
+ ).to(stringSerde, longSerde, "tagg");
+
+ return new KafkaStreams(builder, props);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/448c1a41/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java
new file mode 100644
index 0000000..6f29e00
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java
@@ -0,0 +1,596 @@
+/**
+ * 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.streams.tests;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.test.TestUtils;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+public class SmokeTestDriver extends SmokeTestUtil {
+
+ public static final int MAX_RECORD_EMPTY_RETRIES = 60;
+
+ private static class ValueList {
+ public final String key;
+ private final int[] values;
+ private int index;
+
+ ValueList(int min, int max) {
+ this.key = min + "-" + max;
+
+ this.values = new int[max - min + 1];
+ for (int i = 0; i < this.values.length; i++) {
+ this.values[i] = min + i;
+ }
+ // We want to randomize the order of data to test not completely predictable processing order
+ // However, values are also use as a timestamp of the record. (TODO: separate data and timestamp)
+ // We keep some correlation of time and order. Thus, the shuffling is done with a sliding window
+ shuffle(this.values, 10);
+
+ this.index = 0;
+ }
+
+ int next() {
+ return (index < values.length) ? values[index++] : -1;
+ }
+ }
+
+ // This main() is not used by the system test. It is intended to be used for local debugging.
+ public static void main(String[] args) throws Exception {
+ final String kafka = "localhost:9092";
+ final String zookeeper = "localhost:2181";
+ final File stateDir = TestUtils.tempDirectory();
+
+ final int numKeys = 20;
+ final int maxRecordsPerKey = 1000;
+
+ Thread driver = new Thread() {
+ public void run() {
+ try {
+ Map<String, Set<Integer>> allData = generate(kafka, numKeys, maxRecordsPerKey);
+ verify(kafka, allData, maxRecordsPerKey);
+ } catch (Exception ex) {
+ ex.printStackTrace();
+ }
+ }
+ };
+
+ SmokeTestClient streams1 = new SmokeTestClient(createDir(stateDir, "1"), kafka);
+ SmokeTestClient streams2 = new SmokeTestClient(createDir(stateDir, "2"), kafka);
+ SmokeTestClient streams3 = new SmokeTestClient(createDir(stateDir, "3"), kafka);
+ SmokeTestClient streams4 = new SmokeTestClient(createDir(stateDir, "4"), kafka);
+
+ System.out.println("starting the driver");
+ driver.start();
+
+ System.out.println("starting the first and second client");
+ streams1.start();
+ streams2.start();
+
+ sleep(10000);
+
+ System.out.println("starting the third client");
+ streams3.start();
+
+ System.out.println("closing the first client");
+ streams1.close();
+ System.out.println("closed the first client");
+
+ sleep(10000);
+
+ System.out.println("starting the forth client");
+ streams4.start();
+
+ driver.join();
+
+ System.out.println("driver stopped");
+ streams2.close();
+ streams3.close();
+ streams4.close();
+
+ System.out.println("shutdown");
+ }
+
+ public static Map<String, Set<Integer>> generate(String kafka, final int numKeys, final int maxRecordsPerKey) throws Exception {
+ Properties props = new Properties();
+ props.put(ProducerConfig.CLIENT_ID_CONFIG, "SmokeTest");
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+
+ KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(props);
+
+ int numRecordsProduced = 0;
+
+ Map<String, Set<Integer>> allData = new HashMap<>();
+ ValueList[] data = new ValueList[numKeys];
+ for (int i = 0; i < numKeys; i++) {
+ data[i] = new ValueList(i, i + maxRecordsPerKey - 1);
+ allData.put(data[i].key, new HashSet<Integer>());
+ }
+ Random rand = new Random();
+
+ int remaining = data.length;
+
+ while (remaining > 0) {
+ int index = rand.nextInt(remaining);
+ String key = data[index].key;
+ int value = data[index].next();
+
+ if (value < 0) {
+ remaining--;
+ data[index] = data[remaining];
+ } else {
+
+ ProducerRecord<byte[], byte[]> record =
+ new ProducerRecord<>("data", stringSerde.serializer().serialize("", key), intSerde.serializer().serialize("", value));
+
+ producer.send(record, new Callback() {
+ @Override
+ public void onCompletion(final RecordMetadata metadata, final Exception exception) {
+ if (exception != null) {
+ exception.printStackTrace();
+ System.exit(-1);
+ }
+ }
+ });
+
+
+ numRecordsProduced++;
+ allData.get(key).add(value);
+ if (numRecordsProduced % 100 == 0)
+ System.out.println(numRecordsProduced + " records produced");
+ Utils.sleep(2);
+
+ }
+ }
+ producer.close();
+ return Collections.unmodifiableMap(allData);
+ }
+
+ private static void shuffle(int[] data, int windowSize) {
+ Random rand = new Random();
+ for (int i = 0; i < data.length; i++) {
+ // we shuffle data within windowSize
+ int j = rand.nextInt(Math.min(data.length - i, windowSize)) + i;
+
+ // swap
+ int tmp = data[i];
+ data[i] = data[j];
+ data[j] = tmp;
+ }
+ }
+
+ public static void verify(String kafka, Map<String, Set<Integer>> allData, int maxRecordsPerKey) {
+ Properties props = new Properties();
+ props.put(ConsumerConfig.CLIENT_ID_CONFIG, "verifier");
+ props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+
+ KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props);
+ List<TopicPartition> partitions = getAllPartitions(consumer, "echo", "max", "min", "dif", "sum", "cnt", "avg", "wcnt", "tagg");
+ consumer.assign(partitions);
+ consumer.seekToBeginning(partitions);
+
+ final int recordsGenerated = allData.size() * maxRecordsPerKey;
+ int recordsProcessed = 0;
+
+ HashMap<String, Integer> max = new HashMap<>();
+ HashMap<String, Integer> min = new HashMap<>();
+ HashMap<String, Integer> dif = new HashMap<>();
+ HashMap<String, Long> sum = new HashMap<>();
+ HashMap<String, Long> cnt = new HashMap<>();
+ HashMap<String, Double> avg = new HashMap<>();
+ HashMap<String, Long> wcnt = new HashMap<>();
+ HashMap<String, Long> tagg = new HashMap<>();
+
+ HashSet<String> keys = new HashSet<>();
+ HashMap<String, Set<Integer>> received = new HashMap<>();
+ for (String key : allData.keySet()) {
+ keys.add(key);
+ received.put(key, new HashSet<Integer>());
+ }
+ int retry = 0;
+ final long start = System.currentTimeMillis();
+ while (System.currentTimeMillis() - start < TimeUnit.MINUTES.toMillis(3)) {
+ ConsumerRecords<byte[], byte[]> records = consumer.poll(500);
+ if (records.isEmpty() && recordsProcessed >= recordsGenerated) {
+ if (verifyMin(min, allData, false)
+ && verifyMax(max, allData, false)
+ && verifyDif(dif, allData, false)
+ && verifySum(sum, allData, false)
+ && verifyCnt(cnt, allData, false)
+ && verifyAvg(avg, allData, false)
+ && verifyTAgg(tagg, allData, false)) {
+ break;
+ }
+ if (retry++ > MAX_RECORD_EMPTY_RETRIES) {
+ break;
+ }
+ } else {
+ for (ConsumerRecord<byte[], byte[]> record : records) {
+ String key = stringSerde.deserializer().deserialize("", record.key());
+ switch (record.topic()) {
+ case "echo":
+ Integer value = intSerde.deserializer().deserialize("", record.value());
+ recordsProcessed++;
+ if (recordsProcessed % 100 == 0) {
+ System.out.println("Echo records processed = " + recordsProcessed);
+ }
+ received.get(key).add(value);
+ break;
+ case "min":
+ min.put(key, intSerde.deserializer().deserialize("", record.value()));
+ break;
+ case "max":
+ max.put(key, intSerde.deserializer().deserialize("", record.value()));
+ break;
+ case "dif":
+ dif.put(key, intSerde.deserializer().deserialize("", record.value()));
+ break;
+ case "sum":
+ sum.put(key, longSerde.deserializer().deserialize("", record.value()));
+ break;
+ case "cnt":
+ cnt.put(key, longSerde.deserializer().deserialize("", record.value()));
+ break;
+ case "avg":
+ avg.put(key, doubleSerde.deserializer().deserialize("", record.value()));
+ break;
+ case "wcnt":
+ wcnt.put(key, longSerde.deserializer().deserialize("", record.value()));
+ break;
+ case "tagg":
+ tagg.put(key, longSerde.deserializer().deserialize("", record.value()));
+ break;
+ default:
+ System.out.println("unknown topic: " + record.topic());
+ }
+ }
+ }
+ }
+ consumer.close();
+ final long finished = System.currentTimeMillis() - start;
+ System.out.println("Verification time=" + finished);
+ System.out.println("-------------------");
+ System.out.println("Result Verification");
+ System.out.println("-------------------");
+ System.out.println("recordGenerated=" + recordsGenerated);
+ System.out.println("recordProcessed=" + recordsProcessed);
+
+ if (recordsProcessed > recordsGenerated) {
+ System.out.println("PROCESSED-MORE-THAN-GENERATED");
+ } else if (recordsProcessed < recordsGenerated) {
+ System.out.println("PROCESSED-LESS-THAN-GENERATED");
+ }
+
+ boolean success;
+ success = allData.equals(received);
+
+ if (success) {
+ System.out.println("ALL-RECORDS-DELIVERED");
+ } else {
+ int missedCount = 0;
+ for (Map.Entry<String, Set<Integer>> entry : allData.entrySet()) {
+ missedCount += received.get(entry.getKey()).size();
+ }
+ System.out.println("missedRecords=" + missedCount);
+ }
+
+ success &= verifyMin(min, allData, true);
+ success &= verifyMax(max, allData, true);
+ success &= verifyDif(dif, allData, true);
+ success &= verifySum(sum, allData, true);
+ success &= verifyCnt(cnt, allData, true);
+ success &= verifyAvg(avg, allData, true);
+ success &= verifyTAgg(tagg, allData, true);
+
+ System.out.println(success ? "SUCCESS" : "FAILURE");
+ }
+
+ private static boolean verifyMin(Map<String, Integer> map, Map<String, Set<Integer>> allData, final boolean print) {
+ if (map.isEmpty()) {
+ if (print) {
+ System.out.println("min is empty");
+ }
+ return false;
+ } else {
+ if (print) {
+ System.out.println("verifying min");
+ }
+
+ if (map.size() != allData.size()) {
+ if (print) {
+ System.out.println("fail: resultCount=" + map.size() + " expectedCount=" + allData.size());
+ }
+ return false;
+ }
+ for (Map.Entry<String, Integer> entry : map.entrySet()) {
+ int expected = getMin(entry.getKey());
+ if (expected != entry.getValue()) {
+ if (print) {
+ System.out.println("fail: key=" + entry.getKey() + " min=" + entry.getValue() + " expected=" + expected);
+ }
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ private static boolean verifyMax(Map<String, Integer> map, Map<String, Set<Integer>> allData, final boolean print) {
+ if (map.isEmpty()) {
+ if (print) {
+ System.out.println("max is empty");
+ }
+ return false;
+ } else {
+ if (print) {
+ System.out.println("verifying max");
+ }
+
+ if (map.size() != allData.size()) {
+ if (print) {
+ System.out.println("fail: resultCount=" + map.size() + " expectedCount=" + allData.size());
+ }
+ return false;
+ }
+ for (Map.Entry<String, Integer> entry : map.entrySet()) {
+ int expected = getMax(entry.getKey());
+ if (expected != entry.getValue()) {
+ if (print) {
+ System.out.println("fail: key=" + entry.getKey() + " max=" + entry.getValue() + " expected=" + expected);
+ }
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ private static boolean verifyDif(Map<String, Integer> map, Map<String, Set<Integer>> allData, final boolean print) {
+ if (map.isEmpty()) {
+ if (print) {
+ System.out.println("dif is empty");
+ }
+ return false;
+ } else {
+ if (print) {
+ System.out.println("verifying dif");
+ }
+
+ if (map.size() != allData.size()) {
+ if (print) {
+ System.out.println("fail: resultCount=" + map.size() + " expectedCount=" + allData.size());
+ }
+ return false;
+ }
+ for (Map.Entry<String, Integer> entry : map.entrySet()) {
+ int min = getMin(entry.getKey());
+ int max = getMax(entry.getKey());
+ int expected = max - min;
+ if (entry.getValue() == null || expected != entry.getValue()) {
+ if (print) {
+ System.out.println("fail: key=" + entry.getKey() + " dif=" + entry.getValue() + " expected=" + expected);
+ }
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ private static boolean verifyCnt(Map<String, Long> map, Map<String, Set<Integer>> allData, final boolean print) {
+ if (map.isEmpty()) {
+ if (print) {
+ System.out.println("cnt is empty");
+ }
+ return false;
+ } else {
+ if (print) {
+ System.out.println("verifying cnt");
+ }
+
+ if (map.size() != allData.size()) {
+ if (print) {
+ System.out.println("fail: resultCount=" + map.size() + " expectedCount=" + allData.size());
+ }
+ return false;
+ }
+ for (Map.Entry<String, Long> entry : map.entrySet()) {
+ int min = getMin(entry.getKey());
+ int max = getMax(entry.getKey());
+ long expected = (max - min) + 1L;
+ if (expected != entry.getValue()) {
+ if (print) {
+ System.out.println("fail: key=" + entry.getKey() + " cnt=" + entry.getValue() + " expected=" + expected);
+ }
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ private static boolean verifySum(Map<String, Long> map, Map<String, Set<Integer>> allData, final boolean print) {
+ if (map.isEmpty()) {
+ if (print) {
+ System.out.println("sum is empty");
+ }
+ return false;
+ } else {
+ if (print) {
+ System.out.println("verifying sum");
+ }
+
+ if (map.size() != allData.size()) {
+ if (print) {
+ System.out.println("fail: resultCount=" + map.size() + " expectedCount=" + allData.size());
+ }
+ return false;
+ }
+ for (Map.Entry<String, Long> entry : map.entrySet()) {
+ int min = getMin(entry.getKey());
+ int max = getMax(entry.getKey());
+ long expected = ((long) min + (long) max) * (max - min + 1L) / 2L;
+ if (expected != entry.getValue()) {
+ if (print) {
+ System.out.println("fail: key=" + entry.getKey() + " sum=" + entry.getValue() + " expected=" + expected);
+ }
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ private static boolean verifyAvg(Map<String, Double> map, Map<String, Set<Integer>> allData, final boolean print) {
+ if (map.isEmpty()) {
+ if (print) {
+ System.out.println("avg is empty");
+ }
+ return false;
+ } else {
+ if (print) {
+ System.out.println("verifying avg");
+ }
+
+ if (map.size() != allData.size()) {
+ if (print) {
+ System.out.println("fail: resultCount=" + map.size() + " expectedCount=" + allData.size());
+ }
+ return false;
+ }
+ for (Map.Entry<String, Double> entry : map.entrySet()) {
+ int min = getMin(entry.getKey());
+ int max = getMax(entry.getKey());
+ double expected = ((long) min + (long) max) / 2.0;
+
+ if (entry.getValue() == null || expected != entry.getValue()) {
+ if (print) {
+ System.out.println("fail: key=" + entry.getKey() + " avg=" + entry.getValue() + " expected=" + expected);
+ }
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+
+ private static boolean verifyTAgg(Map<String, Long> map, Map<String, Set<Integer>> allData, final boolean print) {
+ if (map.isEmpty()) {
+ if (print) {
+ System.out.println("tagg is empty");
+ }
+ return false;
+ } else {
+ if (print) {
+ System.out.println("verifying tagg");
+ }
+
+ // generate expected answer
+ Map<String, Long> expected = new HashMap<>();
+ for (String key : allData.keySet()) {
+ int min = getMin(key);
+ int max = getMax(key);
+ String cnt = Long.toString(max - min + 1L);
+
+ if (expected.containsKey(cnt)) {
+ expected.put(cnt, expected.get(cnt) + 1L);
+ } else {
+ expected.put(cnt, 1L);
+ }
+ }
+
+ // check the result
+ for (Map.Entry<String, Long> entry : map.entrySet()) {
+ String key = entry.getKey();
+ Long expectedCount = expected.remove(key);
+ if (expectedCount == null)
+ expectedCount = 0L;
+
+ if (entry.getValue() != expectedCount) {
+ if (print) {
+ System.out.println("fail: key=" + key + " tagg=" + entry.getValue() + " expected=" + expected.get(key));
+ }
+ return false;
+ }
+ }
+
+ }
+ return true;
+ }
+
+ private static int getMin(String key) {
+ return Integer.parseInt(key.split("-")[0]);
+ }
+
+ private static int getMax(String key) {
+ return Integer.parseInt(key.split("-")[1]);
+ }
+
+ private static int getMinFromWKey(String key) {
+ return getMin(key.split("@")[0]);
+ }
+
+ private static int getMaxFromWKey(String key) {
+ return getMax(key.split("@")[0]);
+ }
+
+ private static long getStartFromWKey(String key) {
+ return Long.parseLong(key.split("@")[1]);
+ }
+
+ private static List<TopicPartition> getAllPartitions(KafkaConsumer<?, ?> consumer, String... topics) {
+ ArrayList<TopicPartition> partitions = new ArrayList<>();
+
+ for (String topic : topics) {
+ for (PartitionInfo info : consumer.partitionsFor(topic)) {
+ partitions.add(new TopicPartition(info.topic(), info.partition()));
+ }
+ }
+ return partitions;
+ }
+
+}