HOTIFX: streams system test do not start up correctly Author: Matthias J. Sax <[email protected]>
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/e0eefdd3 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/e0eefdd3 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/e0eefdd3 Branch: refs/heads/0.10.2 Commit: e0eefdd36ca2606286a14cb033879ff4a9f56d58 Parents: 87bd754 Author: Matthias J. Sax <[email protected]> Authored: Tue Jan 24 19:36:08 2017 -0800 Committer: Guozhang Wang <[email protected]> Committed: Tue Jan 24 19:53:14 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/e0eefdd3/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/e0eefdd3/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/e0eefdd3/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/e0eefdd3/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/e0eefdd3/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/e0eefdd3/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/e0eefdd3/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/e0eefdd3/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/e0eefdd3/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; + } + +}
