You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2017/10/02 20:49:58 UTC
kafka git commit: KAFKA-5225;
StreamsResetter doesn't allow custom Consumer properties
Repository: kafka
Updated Branches:
refs/heads/trunk cc84686a4 -> f9865d52e
KAFKA-5225; StreamsResetter doesn't allow custom Consumer properties
Author: Matthias J. Sax <ma...@confluent.io>
Author: Bharat Viswanadham <bh...@us.ibm.com>
Reviewers: Ismael Juma <is...@juma.me.uk>, Damian Guy <da...@gmail.com>
Closes #3970 from mjsax/kafka-5225-streams-resetter-properties
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/f9865d52
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/f9865d52
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/f9865d52
Branch: refs/heads/trunk
Commit: f9865d52e81bbdddb7889d6c3cc7be537e610826
Parents: cc84686
Author: Matthias J. Sax <ma...@confluent.io>
Authored: Mon Oct 2 13:47:45 2017 -0700
Committer: Damian Guy <da...@gmail.com>
Committed: Mon Oct 2 13:47:45 2017 -0700
----------------------------------------------------------------------
build.gradle | 1 +
.../main/scala/kafka/tools/StreamsResetter.java | 60 ++-
.../AbstractResetIntegrationTest.java | 473 +++++++++++++++++++
.../integration/ResetIntegrationTest.java | 352 +-------------
.../ResetIntegrationWithSslTest.java | 96 ++++
.../integration/utils/KafkaEmbedded.java | 5 +-
6 files changed, 615 insertions(+), 372 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/f9865d52/build.gradle
----------------------------------------------------------------------
diff --git a/build.gradle b/build.gradle
index cbae7b0..d7b799b 100644
--- a/build.gradle
+++ b/build.gradle
@@ -893,6 +893,7 @@ project(':streams') {
testCompile project(':core').sourceSets.test.output
testCompile libs.junit
testCompile libs.easymock
+ testCompile libs.bcpkix
testRuntime libs.slf4jlog4j
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/f9865d52/core/src/main/scala/kafka/tools/StreamsResetter.java
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/tools/StreamsResetter.java b/core/src/main/scala/kafka/tools/StreamsResetter.java
index 09d0d75..5539258 100644
--- a/core/src/main/scala/kafka/tools/StreamsResetter.java
+++ b/core/src/main/scala/kafka/tools/StreamsResetter.java
@@ -16,7 +16,11 @@
*/
package kafka.tools;
-
+import joptsimple.OptionException;
+import joptsimple.OptionParser;
+import joptsimple.OptionSet;
+import joptsimple.OptionSpec;
+import joptsimple.OptionSpecBuilder;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DeleteTopicsResult;
import org.apache.kafka.clients.admin.KafkaAdminClient;
@@ -27,9 +31,11 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.common.utils.Utils;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
@@ -38,12 +44,6 @@ import java.util.Properties;
import java.util.Set;
import java.util.concurrent.TimeUnit;
-import joptsimple.OptionException;
-import joptsimple.OptionParser;
-import joptsimple.OptionSet;
-import joptsimple.OptionSpec;
-import joptsimple.OptionSpecBuilder;
-
/**
* {@link StreamsResetter} resets the processing state of a Kafka Streams application so that, for example, you can reprocess its input from scratch.
* <p>
@@ -71,14 +71,13 @@ public class StreamsResetter {
private static final int EXIT_CODE_ERROR = 1;
private static OptionSpec<String> bootstrapServerOption;
- private static OptionSpecBuilder zookeeperOption;
private static OptionSpec<String> applicationIdOption;
private static OptionSpec<String> inputTopicsOption;
private static OptionSpec<String> intermediateTopicsOption;
private static OptionSpecBuilder dryRunOption;
+ private static OptionSpec<String> commandConfigOption;
private OptionSet options = null;
- private final Properties consumerConfig = new Properties();
private final List<String> allTopics = new LinkedList<>();
private boolean dryRun = false;
@@ -86,10 +85,8 @@ public class StreamsResetter {
return run(args, new Properties());
}
- public int run(final String[] args, final Properties config) {
- consumerConfig.clear();
- consumerConfig.putAll(config);
-
+ public int run(final String[] args,
+ final Properties config) {
int exitCode = EXIT_CODE_SUCCESS;
KafkaAdminClient kafkaAdminClient = null;
@@ -99,12 +96,14 @@ public class StreamsResetter {
dryRun = options.has(dryRunOption);
final String groupId = options.valueOf(applicationIdOption);
+ final Properties properties = new Properties();
+ if (options.has(commandConfigOption)) {
+ properties.putAll(Utils.loadProps(options.valueOf(commandConfigOption)));
+ }
+ properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServerOption));
- validateNoActiveConsumers(groupId);
-
- final Properties adminClientProperties = new Properties();
- adminClientProperties.put("bootstrap.servers", options.valueOf(bootstrapServerOption));
- kafkaAdminClient = (KafkaAdminClient) AdminClient.create(adminClientProperties);
+ validateNoActiveConsumers(groupId, properties);
+ kafkaAdminClient = (KafkaAdminClient) AdminClient.create(properties);
allTopics.clear();
allTopics.addAll(kafkaAdminClient.listTopics().names().get(60, TimeUnit.SECONDS));
@@ -112,7 +111,10 @@ public class StreamsResetter {
if (dryRun) {
System.out.println("----Dry run displays the actions which will be performed when running Streams Reset Tool----");
}
- maybeResetInputAndSeekToEndIntermediateTopicOffsets();
+
+ final HashMap<Object, Object> consumerConfig = new HashMap<>(config);
+ consumerConfig.putAll(properties);
+ maybeResetInputAndSeekToEndIntermediateTopicOffsets(consumerConfig);
maybeDeleteInternalTopics(kafkaAdminClient);
} catch (final Throwable e) {
@@ -128,10 +130,11 @@ public class StreamsResetter {
return exitCode;
}
- private void validateNoActiveConsumers(final String groupId) {
+ private void validateNoActiveConsumers(final String groupId,
+ final Properties properties) {
kafka.admin.AdminClient olderAdminClient = null;
try {
- olderAdminClient = kafka.admin.AdminClient.createSimplePlaintext(options.valueOf(bootstrapServerOption));
+ olderAdminClient = kafka.admin.AdminClient.create(properties);
if (!olderAdminClient.describeConsumerGroup(groupId, 0).consumers().get().isEmpty()) {
throw new IllegalStateException("Consumer group '" + groupId + "' is still active. "
+ "Make sure to stop all running application instances before running the reset tool.");
@@ -156,8 +159,6 @@ public class StreamsResetter {
.ofType(String.class)
.defaultsTo("localhost:9092")
.describedAs("urls");
- zookeeperOption = optionParser.accepts("zookeeper", "Zookeeper option is deprecated by bootstrap.servers, as the reset tool would no longer access Zookeeper directly.");
-
inputTopicsOption = optionParser.accepts("input-topics", "Comma-separated list of user input topics. For these topics, the tool will reset the offset to the earliest available offset.")
.withRequiredArg()
.ofType(String.class)
@@ -168,8 +169,15 @@ public class StreamsResetter {
.ofType(String.class)
.withValuesSeparatedBy(',')
.describedAs("list");
+ commandConfigOption = optionParser.accepts("config-file", "Property file containing configs to be passed to admin clients and embedded consumer.")
+ .withRequiredArg()
+ .ofType(String.class)
+ .describedAs("file name");
dryRunOption = optionParser.accepts("dry-run", "Display the actions that would be performed without executing the reset commands.");
+ // TODO: deprecated in 1.0; can be removed eventually
+ optionParser.accepts("zookeeper", "Zookeeper option is deprecated by bootstrap.servers, as the reset tool would no longer access Zookeeper directly.");
+
try {
options = optionParser.parse(args);
} catch (final OptionException e) {
@@ -178,7 +186,7 @@ public class StreamsResetter {
}
}
- private void maybeResetInputAndSeekToEndIntermediateTopicOffsets() {
+ private void maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map consumerConfig) {
final List<String> inputTopics = options.valuesOf(inputTopicsOption);
final List<String> intermediateTopics = options.valuesOf(intermediateTopicsOption);
@@ -219,7 +227,6 @@ public class StreamsResetter {
final Properties config = new Properties();
config.putAll(consumerConfig);
- config.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServerOption));
config.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
config.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
@@ -274,7 +281,8 @@ public class StreamsResetter {
System.out.println("Done.");
}
- private void maybeSeekToEnd(final KafkaConsumer<byte[], byte[]> client, final Set<TopicPartition> intermediateTopicPartitions) {
+ private void maybeSeekToEnd(final KafkaConsumer<byte[], byte[]> client,
+ final Set<TopicPartition> intermediateTopicPartitions) {
final String groupId = options.valueOf(applicationIdOption);
final List<String> intermediateTopics = options.valuesOf(intermediateTopicsOption);
http://git-wip-us.apache.org/repos/asf/kafka/blob/f9865d52/streams/src/test/java/org/apache/kafka/streams/integration/AbstractResetIntegrationTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/AbstractResetIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/AbstractResetIntegrationTest.java
new file mode 100644
index 0000000..6ab7141
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/AbstractResetIntegrationTest.java
@@ -0,0 +1,473 @@
+/*
+ * 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.integration;
+
+import kafka.admin.AdminClient;
+import kafka.tools.StreamsResetter;
+import kafka.utils.MockTime;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.KafkaAdminClient;
+import org.apache.kafka.clients.admin.ListTopicsOptions;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.LongSerializer;
+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.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KeyValueMapper;
+import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.streams.kstream.TimeWindows;
+import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.test.TestCondition;
+import org.apache.kafka.test.TestUtils;
+import org.junit.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+abstract class AbstractResetIntegrationTest {
+ private final static Logger log = LoggerFactory.getLogger(AbstractResetIntegrationTest.class);
+
+ static final int NUM_BROKERS = 1;
+
+ private static final String APP_ID = "cleanup-integration-test";
+ private static final String INPUT_TOPIC = "inputTopic";
+ private static final String OUTPUT_TOPIC = "outputTopic";
+ private static final String OUTPUT_TOPIC_2 = "outputTopic2";
+ private static final String OUTPUT_TOPIC_2_RERUN = "outputTopic2_rerun";
+ private static final String INTERMEDIATE_USER_TOPIC = "userTopic";
+
+ private static final long STREAMS_CONSUMER_TIMEOUT = 2000L;
+ private static final long CLEANUP_CONSUMER_TIMEOUT = 2000L;
+ private static final int TIMEOUT_MULTIPLIER = 5;
+
+ private static AdminClient adminClient = null;
+ private static KafkaAdminClient kafkaAdminClient = null;
+ private static int testNo = 0;
+
+ static EmbeddedKafkaCluster cluster;
+ static String bootstrapServers;
+ static MockTime mockTime;
+
+ private final AbstractResetIntegrationTest.WaitUntilConsumerGroupGotClosed consumerGroupInactive = new AbstractResetIntegrationTest.WaitUntilConsumerGroupGotClosed();
+
+ private class WaitUntilConsumerGroupGotClosed implements TestCondition {
+ @Override
+ public boolean conditionMet() {
+ return adminClient.describeConsumerGroup(APP_ID, 0).consumers().get().isEmpty();
+ }
+ }
+
+ static void afterClassGlobalCleanup() {
+ if (adminClient != null) {
+ adminClient.close();
+ adminClient = null;
+ }
+
+ if (kafkaAdminClient != null) {
+ kafkaAdminClient.close(10, TimeUnit.SECONDS);
+ kafkaAdminClient = null;
+ }
+ }
+
+ void beforePrepareTest() throws Exception {
+ ++testNo;
+ bootstrapServers = cluster.bootstrapServers();
+ mockTime = cluster.time;
+
+ Properties sslConfig = getClientSslConfig();
+ if (sslConfig == null) {
+ sslConfig = new Properties();
+ sslConfig.put("bootstrap.servers", bootstrapServers);
+ }
+
+ if (adminClient == null) {
+ adminClient = AdminClient.create(sslConfig);
+ }
+
+ if (kafkaAdminClient == null) {
+ kafkaAdminClient = (KafkaAdminClient) org.apache.kafka.clients.admin.AdminClient.create(sslConfig);
+ }
+
+ // busy wait until cluster (ie, ConsumerGroupCoordinator) is available
+ while (true) {
+ Thread.sleep(50);
+
+ try {
+ TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT,
+ "Test consumer group active even after waiting " + (TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT) + " ms.");
+ } catch (final TimeoutException e) {
+ continue;
+ }
+ break;
+ }
+
+ prepareInputData();
+ }
+
+ Properties getClientSslConfig() {
+ return null;
+ }
+
+ void testReprocessingFromScratchAfterResetWithoutIntermediateUserTopic() throws Exception {
+ final Properties sslConfig = getClientSslConfig();
+ final Properties streamsConfiguration = prepareTest();
+
+ final Properties resultTopicConsumerConfig = new Properties();
+ if (sslConfig != null) {
+ resultTopicConsumerConfig.putAll(sslConfig);
+ }
+ resultTopicConsumerConfig.putAll(TestUtils.consumerConfig(
+ bootstrapServers,
+ APP_ID + "-standard-consumer-" + OUTPUT_TOPIC,
+ LongDeserializer.class,
+ LongDeserializer.class));
+
+ // RUN
+ KafkaStreams streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfiguration);
+ final KafkaStreams handlerReference = streams;
+ streams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+ @Override
+ public void uncaughtException(Thread t, Throwable e) {
+ handlerReference.close(10, TimeUnit.SECONDS);
+ log.error("Streams application failed: ", e);
+ }
+ });
+ streams.start();
+ final List<KeyValue<Long, Long>> result = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
+ resultTopicConsumerConfig,
+ OUTPUT_TOPIC,
+ 10);
+
+ streams.close();
+ TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * STREAMS_CONSUMER_TIMEOUT,
+ "Streams Application consumer group did not time out after " + (TIMEOUT_MULTIPLIER * STREAMS_CONSUMER_TIMEOUT) + " ms.");
+
+ // RESET
+ streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfiguration);
+ final KafkaStreams handlerReference2 = streams;
+ streams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+ @Override
+ public void uncaughtException(Thread t, Throwable e) {
+ handlerReference2.close(10, TimeUnit.SECONDS);
+ log.error("Streams application failed: ", e);
+ }
+ });
+ streams.cleanUp();
+ cleanGlobal(null, sslConfig);
+ TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT,
+ "Reset Tool consumer group did not time out after " + (TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT) + " ms.");
+
+ assertInternalTopicsGotDeleted(null);
+
+ // RE-RUN
+ streams.start();
+ final List<KeyValue<Long, Long>> resultRerun = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
+ resultTopicConsumerConfig,
+ OUTPUT_TOPIC,
+ 10);
+ streams.close();
+
+ assertThat(resultRerun, equalTo(result));
+
+ TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT,
+ "Reset Tool consumer group did not time out after " + (TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT) + " ms.");
+ cleanGlobal(null, sslConfig);
+ }
+
+ void testReprocessingFromScratchAfterResetWithIntermediateUserTopic() throws Exception {
+ cluster.createTopic(INTERMEDIATE_USER_TOPIC);
+
+ final Properties sslConfig = getClientSslConfig();
+ final Properties streamsConfiguration = prepareTest();
+
+ final Properties resultTopicConsumerConfig = new Properties();
+ if (sslConfig != null) {
+ resultTopicConsumerConfig.putAll(sslConfig);
+ }
+ resultTopicConsumerConfig.putAll(TestUtils.consumerConfig(
+ bootstrapServers,
+ APP_ID + "-standard-consumer-" + OUTPUT_TOPIC,
+ LongDeserializer.class,
+ LongDeserializer.class));
+
+ // RUN
+ KafkaStreams streams = new KafkaStreams(setupTopologyWithIntermediateUserTopic(OUTPUT_TOPIC_2), streamsConfiguration);
+ final KafkaStreams handlerReference = streams;
+ streams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+ @Override
+ public void uncaughtException(Thread t, Throwable e) {
+ handlerReference.close(10, TimeUnit.SECONDS);
+ log.error("Streams application failed: ", e);
+ }
+ });
+ streams.start();
+ final List<KeyValue<Long, Long>> result = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
+ resultTopicConsumerConfig,
+ OUTPUT_TOPIC,
+ 10);
+ // receive only first values to make sure intermediate user topic is not consumed completely
+ // => required to test "seekToEnd" for intermediate topics
+ final List<KeyValue<Long, Long>> result2 = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
+ resultTopicConsumerConfig,
+ OUTPUT_TOPIC_2,
+ 40
+ );
+
+ streams.close();
+ TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * STREAMS_CONSUMER_TIMEOUT,
+ "Streams Application consumer group did not time out after " + (TIMEOUT_MULTIPLIER * STREAMS_CONSUMER_TIMEOUT) + " ms.");
+
+ // insert bad record to make sure intermediate user topic gets seekToEnd()
+ mockTime.sleep(1);
+ Properties producerConfig = sslConfig;
+ if (producerConfig == null) {
+ producerConfig = new Properties();
+ }
+ producerConfig.putAll(TestUtils.producerConfig(bootstrapServers, LongSerializer.class, StringSerializer.class));
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ INTERMEDIATE_USER_TOPIC,
+ Collections.singleton(new KeyValue<>(-1L, "badRecord-ShouldBeSkipped")),
+ producerConfig,
+ mockTime.milliseconds());
+
+ // RESET
+ streams = new KafkaStreams(setupTopologyWithIntermediateUserTopic(OUTPUT_TOPIC_2_RERUN), streamsConfiguration);
+ final KafkaStreams handlerReference2 = streams;
+ streams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+ @Override
+ public void uncaughtException(Thread t, Throwable e) {
+ handlerReference2.close(10, TimeUnit.SECONDS);
+ log.error("Streams application failed: ", e);
+ }
+ });
+ streams.cleanUp();
+ cleanGlobal(INTERMEDIATE_USER_TOPIC, sslConfig);
+ TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT,
+ "Reset Tool consumer group did not time out after " + (TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT) + " ms.");
+
+ assertInternalTopicsGotDeleted(INTERMEDIATE_USER_TOPIC);
+
+ // RE-RUN
+ streams.start();
+ final List<KeyValue<Long, Long>> resultRerun = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
+ resultTopicConsumerConfig,
+ OUTPUT_TOPIC,
+ 10);
+ final List<KeyValue<Long, Long>> resultRerun2 = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
+ resultTopicConsumerConfig,
+ OUTPUT_TOPIC_2_RERUN,
+ 40
+ );
+ streams.close();
+
+ assertThat(resultRerun, equalTo(result));
+ assertThat(resultRerun2, equalTo(result2));
+
+ TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT,
+ "Reset Tool consumer group did not time out after " + (TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT) + " ms.");
+ cleanGlobal(INTERMEDIATE_USER_TOPIC, sslConfig);
+
+ cluster.deleteTopicAndWait(INTERMEDIATE_USER_TOPIC);
+ }
+
+ private Properties prepareTest() throws IOException {
+ Properties streamsConfiguration = getClientSslConfig();
+ if (streamsConfiguration == null) {
+ streamsConfiguration = new Properties();
+ }
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, APP_ID + testNo);
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Long().getClass());
+ streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfiguration.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
+ streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);
+ streamsConfiguration.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 100);
+ streamsConfiguration.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "" + STREAMS_CONSUMER_TIMEOUT);
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ streamsConfiguration.put(IntegrationTestUtils.INTERNAL_LEAVE_GROUP_ON_CLOSE, true);
+
+ IntegrationTestUtils.purgeLocalStreamsState(streamsConfiguration);
+
+ return streamsConfiguration;
+ }
+
+ private void prepareInputData() throws Exception {
+ cluster.deleteAndRecreateTopics(INPUT_TOPIC, OUTPUT_TOPIC, OUTPUT_TOPIC_2, OUTPUT_TOPIC_2_RERUN);
+
+ Properties producerConfig = getClientSslConfig();
+ if (producerConfig == null) {
+ producerConfig = new Properties();
+ }
+ producerConfig.putAll(TestUtils.producerConfig(bootstrapServers, LongSerializer.class, StringSerializer.class));
+
+ mockTime.sleep(10);
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(0L, "aaa")), producerConfig, mockTime.milliseconds());
+ mockTime.sleep(10);
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(1L, "bbb")), producerConfig, mockTime.milliseconds());
+ mockTime.sleep(10);
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(0L, "ccc")), producerConfig, mockTime.milliseconds());
+ mockTime.sleep(10);
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(1L, "ddd")), producerConfig, mockTime.milliseconds());
+ mockTime.sleep(10);
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(0L, "eee")), producerConfig, mockTime.milliseconds());
+ mockTime.sleep(10);
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(1L, "fff")), producerConfig, mockTime.milliseconds());
+ mockTime.sleep(1);
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(0L, "ggg")), producerConfig, mockTime.milliseconds());
+ mockTime.sleep(1);
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(1L, "hhh")), producerConfig, mockTime.milliseconds());
+ mockTime.sleep(1);
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(0L, "iii")), producerConfig, mockTime.milliseconds());
+ mockTime.sleep(1);
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(1L, "jjj")), producerConfig, mockTime.milliseconds());
+ }
+
+ private Topology setupTopologyWithIntermediateUserTopic(final String outputTopic2) {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KStream<Long, String> input = builder.stream(INPUT_TOPIC);
+
+ // use map to trigger internal re-partitioning before groupByKey
+ input.map(new KeyValueMapper<Long, String, KeyValue<Long, String>>() {
+ @Override
+ public KeyValue<Long, String> apply(final Long key, final String value) {
+ return new KeyValue<>(key, value);
+ }
+ })
+ .groupByKey()
+ .count()
+ .toStream()
+ .to(OUTPUT_TOPIC, Produced.with(Serdes.Long(), Serdes.Long()));
+
+ input.through(INTERMEDIATE_USER_TOPIC)
+ .groupByKey()
+ .windowedBy(TimeWindows.of(35).advanceBy(10))
+ .count()
+ .toStream()
+ .map(new KeyValueMapper<Windowed<Long>, Long, KeyValue<Long, Long>>() {
+ @Override
+ public KeyValue<Long, Long> apply(final Windowed<Long> key, final Long value) {
+ return new KeyValue<>(key.window().start() + key.window().end(), value);
+ }
+ })
+ .to(outputTopic2, Produced.with(Serdes.Long(), Serdes.Long()));
+
+ return builder.build();
+ }
+
+ private Topology setupTopologyWithoutIntermediateUserTopic() {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KStream<Long, String> input = builder.stream(INPUT_TOPIC);
+
+ // use map to trigger internal re-partitioning before groupByKey
+ input.map(new KeyValueMapper<Long, String, KeyValue<Long, Long>>() {
+ @Override
+ public KeyValue<Long, Long> apply(final Long key, final String value) {
+ return new KeyValue<>(key, key);
+ }
+ }).to(OUTPUT_TOPIC, Produced.with(Serdes.Long(), Serdes.Long()));
+
+ return builder.build();
+ }
+
+ private void cleanGlobal(final String intermediateUserTopic, final Properties sslConfig) throws Exception {
+ // leaving --zookeeper arg here to ensure tool works if users add it
+ final String[] parameters;
+ if (intermediateUserTopic != null) {
+ parameters = new String[]{
+ "--application-id", APP_ID + testNo,
+ "--bootstrap-servers", bootstrapServers,
+ "--input-topics", INPUT_TOPIC,
+ "--intermediate-topics", INTERMEDIATE_USER_TOPIC,
+ "--zookeeper", "localhost:2181"
+ };
+ } else {
+ if (sslConfig != null) {
+ final File configFile = TestUtils.tempFile();
+ final BufferedWriter writer = new BufferedWriter(new FileWriter(configFile));
+ writer.write(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG + "=SSL\n");
+ writer.write(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG + "=" + sslConfig.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG) + "\n");
+ writer.write(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG + "=" + sslConfig.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG) + "\n");
+ writer.close();
+
+ parameters = new String[]{
+ "--application-id", APP_ID + testNo,
+ "--bootstrap-servers", bootstrapServers,
+ "--input-topics", INPUT_TOPIC,
+ "--config-file", configFile.getAbsolutePath()
+ };
+ } else {
+ parameters = new String[]{
+ "--application-id", APP_ID + testNo,
+ "--bootstrap-servers", bootstrapServers,
+ "--input-topics", INPUT_TOPIC
+ };
+ }
+ }
+ final Properties cleanUpConfig = new Properties();
+ cleanUpConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 100);
+ cleanUpConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "" + CLEANUP_CONSUMER_TIMEOUT);
+
+ final int exitCode = new StreamsResetter().run(parameters, cleanUpConfig);
+ Assert.assertEquals(0, exitCode);
+ }
+
+ private void assertInternalTopicsGotDeleted(final String intermediateUserTopic) throws Exception {
+ final Set<String> expectedRemainingTopicsAfterCleanup = new HashSet<>();
+ expectedRemainingTopicsAfterCleanup.add(INPUT_TOPIC);
+ if (intermediateUserTopic != null) {
+ expectedRemainingTopicsAfterCleanup.add(intermediateUserTopic);
+ }
+ expectedRemainingTopicsAfterCleanup.add(OUTPUT_TOPIC);
+ expectedRemainingTopicsAfterCleanup.add(OUTPUT_TOPIC_2);
+ expectedRemainingTopicsAfterCleanup.add(OUTPUT_TOPIC_2_RERUN);
+ expectedRemainingTopicsAfterCleanup.add("__consumer_offsets");
+
+ final Set<String> allTopics = new HashSet<>();
+
+ final ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
+ listTopicsOptions.listInternal(true);
+ allTopics.addAll(kafkaAdminClient.listTopics(listTopicsOptions).names().get(30000, TimeUnit.MILLISECONDS));
+ assertThat(allTopics, equalTo(expectedRemainingTopicsAfterCleanup));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/f9865d52/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java
index d76f5da..549f8f1 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java
@@ -16,57 +16,22 @@
*/
package org.apache.kafka.streams.integration;
-import org.apache.kafka.clients.admin.KafkaAdminClient;
-import org.apache.kafka.clients.admin.ListTopicsOptions;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.common.errors.TimeoutException;
-import org.apache.kafka.common.serialization.LongDeserializer;
-import org.apache.kafka.common.serialization.LongSerializer;
-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.KeyValue;
-import org.apache.kafka.streams.StreamsBuilder;
-import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.Topology;
+import kafka.server.KafkaConfig$;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
-import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
-import org.apache.kafka.streams.kstream.KStream;
-import org.apache.kafka.streams.kstream.KeyValueMapper;
-import org.apache.kafka.streams.kstream.TimeWindows;
-import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.test.IntegrationTest;
-import org.apache.kafka.test.TestCondition;
-import org.apache.kafka.test.TestUtils;
import org.junit.AfterClass;
-import org.junit.Assert;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
-import java.io.IOException;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
import java.util.Properties;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-
-import kafka.admin.AdminClient;
-import kafka.server.KafkaConfig$;
-import kafka.tools.StreamsResetter;
-import kafka.utils.MockTime;
-
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
/**
* Tests local state store and global application cleanup.
*/
@Category({IntegrationTest.class})
-public class ResetIntegrationTest {
- private static final int NUM_BROKERS = 1;
+public class ResetIntegrationTest extends AbstractResetIntegrationTest {
@ClassRule
public static final EmbeddedKafkaCluster CLUSTER;
@@ -80,327 +45,26 @@ public class ResetIntegrationTest {
// otherwise, input records could fall into different windows for different runs depending on the initial mock time
final long alignedTime = (System.currentTimeMillis() / 1000) * 1000;
CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS, props, alignedTime);
+ cluster = CLUSTER;
}
- private static final String APP_ID = "cleanup-integration-test";
- private static final String INPUT_TOPIC = "inputTopic";
- private static final String OUTPUT_TOPIC = "outputTopic";
- private static final String OUTPUT_TOPIC_2 = "outputTopic2";
- private static final String OUTPUT_TOPIC_2_RERUN = "outputTopic2_rerun";
- private static final String INTERMEDIATE_USER_TOPIC = "userTopic";
-
- private static final long STREAMS_CONSUMER_TIMEOUT = 2000L;
- private static final long CLEANUP_CONSUMER_TIMEOUT = 2000L;
- private static final int TIMEOUT_MULTIPLIER = 5;
-
- private static int testNo = 0;
- private static AdminClient adminClient = null;
- private static KafkaAdminClient kafkaAdminClient = null;
-
- private final MockTime mockTime = CLUSTER.time;
- private final WaitUntilConsumerGroupGotClosed consumerGroupInactive = new WaitUntilConsumerGroupGotClosed();
-
@AfterClass
public static void globalCleanup() {
- if (adminClient != null) {
- adminClient.close();
- adminClient = null;
- }
-
- if (kafkaAdminClient != null) {
- kafkaAdminClient.close(10, TimeUnit.SECONDS);
- kafkaAdminClient = null;
- }
+ afterClassGlobalCleanup();
}
@Before
- public void cleanup() throws Exception {
- ++testNo;
-
- if (adminClient == null) {
- adminClient = AdminClient.createSimplePlaintext(CLUSTER.bootstrapServers());
- }
-
- if (kafkaAdminClient == null) {
- Properties props = new Properties();
- props.put("bootstrap.servers", CLUSTER.bootstrapServers());
- kafkaAdminClient = (KafkaAdminClient) org.apache.kafka.clients.admin.AdminClient.create(props);
- }
-
- // busy wait until cluster (ie, ConsumerGroupCoordinator) is available
- while (true) {
- Thread.sleep(50);
-
- try {
- TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT,
- "Test consumer group active even after waiting " + (TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT) + " ms.");
- } catch (final TimeoutException e) {
- continue;
- }
- break;
- }
-
- prepareInputData();
+ public void before() throws Exception {
+ beforePrepareTest();
}
@Test
public void testReprocessingFromScratchAfterResetWithIntermediateUserTopic() throws Exception {
- CLUSTER.createTopic(INTERMEDIATE_USER_TOPIC);
-
- final Properties streamsConfiguration = prepareTest(4);
- final Properties resultTopicConsumerConfig = TestUtils.consumerConfig(
- CLUSTER.bootstrapServers(),
- APP_ID + "-standard-consumer-" + OUTPUT_TOPIC,
- LongDeserializer.class,
- LongDeserializer.class);
-
- // RUN
- KafkaStreams streams = new KafkaStreams(setupTopologyWithIntermediateUserTopic(OUTPUT_TOPIC_2), streamsConfiguration);
- streams.start();
- final List<KeyValue<Long, Long>> result = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
- resultTopicConsumerConfig,
- OUTPUT_TOPIC,
- 10);
- // receive only first values to make sure intermediate user topic is not consumed completely
- // => required to test "seekToEnd" for intermediate topics
- final List<KeyValue<Long, Long>> result2 = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
- resultTopicConsumerConfig,
- OUTPUT_TOPIC_2,
- 40
- );
-
- streams.close();
- TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * STREAMS_CONSUMER_TIMEOUT,
- "Streams Application consumer group did not time out after " + (TIMEOUT_MULTIPLIER * STREAMS_CONSUMER_TIMEOUT) + " ms.");
-
- // insert bad record to make sure intermediate user topic gets seekToEnd()
- mockTime.sleep(1);
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
- INTERMEDIATE_USER_TOPIC,
- Collections.singleton(new KeyValue<>(-1L, "badRecord-ShouldBeSkipped")), TestUtils.producerConfig(CLUSTER.bootstrapServers(), LongSerializer.class, StringSerializer.class), mockTime.milliseconds());
-
- // RESET
- streams = new KafkaStreams(setupTopologyWithIntermediateUserTopic(OUTPUT_TOPIC_2_RERUN), streamsConfiguration);
- streams.cleanUp();
- cleanGlobal(INTERMEDIATE_USER_TOPIC);
- TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT,
- "Reset Tool consumer group did not time out after " + (TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT) + " ms.");
-
- assertInternalTopicsGotDeleted(INTERMEDIATE_USER_TOPIC);
-
- // RE-RUN
- streams.start();
- final List<KeyValue<Long, Long>> resultRerun = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
- resultTopicConsumerConfig,
- OUTPUT_TOPIC,
- 10);
- final List<KeyValue<Long, Long>> resultRerun2 = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
- resultTopicConsumerConfig,
- OUTPUT_TOPIC_2_RERUN,
- 40
- );
- streams.close();
-
- assertThat(resultRerun, equalTo(result));
- assertThat(resultRerun2, equalTo(result2));
-
- TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT,
- "Reset Tool consumer group did not time out after " + (TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT) + " ms.");
- cleanGlobal(INTERMEDIATE_USER_TOPIC);
-
- CLUSTER.deleteTopicAndWait(INTERMEDIATE_USER_TOPIC);
+ super.testReprocessingFromScratchAfterResetWithIntermediateUserTopic();
}
@Test
public void testReprocessingFromScratchAfterResetWithoutIntermediateUserTopic() throws Exception {
- final Properties streamsConfiguration = prepareTest(1);
- final Properties resultTopicConsumerConfig = TestUtils.consumerConfig(
- CLUSTER.bootstrapServers(),
- APP_ID + "-standard-consumer-" + OUTPUT_TOPIC,
- LongDeserializer.class,
- LongDeserializer.class);
-
- // RUN
- KafkaStreams streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfiguration);
- streams.start();
- final List<KeyValue<Long, Long>> result = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
- resultTopicConsumerConfig,
- OUTPUT_TOPIC,
- 10);
-
- streams.close();
- TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * STREAMS_CONSUMER_TIMEOUT,
- "Streams Application consumer group did not time out after " + (TIMEOUT_MULTIPLIER * STREAMS_CONSUMER_TIMEOUT) + " ms.");
-
- // RESET
- streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfiguration);
- streams.cleanUp();
- cleanGlobal(null);
- TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT,
- "Reset Tool consumer group did not time out after " + (TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT) + " ms.");
-
- assertInternalTopicsGotDeleted(null);
-
- // RE-RUN
- streams.start();
- final List<KeyValue<Long, Long>> resultRerun = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
- resultTopicConsumerConfig,
- OUTPUT_TOPIC,
- 10);
- streams.close();
-
- assertThat(resultRerun, equalTo(result));
-
- TestUtils.waitForCondition(consumerGroupInactive, TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT,
- "Reset Tool consumer group did not time out after " + (TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT) + " ms.");
- cleanGlobal(null);
+ super.testReprocessingFromScratchAfterResetWithoutIntermediateUserTopic();
}
-
- private Properties prepareTest(final int threads) throws IOException {
- final Properties streamsConfiguration = new Properties();
- streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, APP_ID + testNo);
- streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
- streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
- streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Long().getClass());
- streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
- streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, threads);
- streamsConfiguration.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
- streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);
- streamsConfiguration.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 100);
- streamsConfiguration.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "" + STREAMS_CONSUMER_TIMEOUT);
- streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
- streamsConfiguration.put(IntegrationTestUtils.INTERNAL_LEAVE_GROUP_ON_CLOSE, true);
-
- IntegrationTestUtils.purgeLocalStreamsState(streamsConfiguration);
-
- return streamsConfiguration;
- }
-
- private void prepareInputData() throws Exception {
- CLUSTER.deleteAndRecreateTopics(INPUT_TOPIC, OUTPUT_TOPIC, OUTPUT_TOPIC_2, OUTPUT_TOPIC_2_RERUN);
-
- final Properties producerConfig = TestUtils.producerConfig(CLUSTER.bootstrapServers(), LongSerializer.class, StringSerializer.class);
-
- mockTime.sleep(10);
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(0L, "aaa")), producerConfig, mockTime.milliseconds());
- mockTime.sleep(10);
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(1L, "bbb")), producerConfig, mockTime.milliseconds());
- mockTime.sleep(10);
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(0L, "ccc")), producerConfig, mockTime.milliseconds());
- mockTime.sleep(10);
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(1L, "ddd")), producerConfig, mockTime.milliseconds());
- mockTime.sleep(10);
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(0L, "eee")), producerConfig, mockTime.milliseconds());
- mockTime.sleep(10);
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(1L, "fff")), producerConfig, mockTime.milliseconds());
- mockTime.sleep(1);
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(0L, "ggg")), producerConfig, mockTime.milliseconds());
- mockTime.sleep(1);
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(1L, "hhh")), producerConfig, mockTime.milliseconds());
- mockTime.sleep(1);
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(0L, "iii")), producerConfig, mockTime.milliseconds());
- mockTime.sleep(1);
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(INPUT_TOPIC, Collections.singleton(new KeyValue<>(1L, "jjj")), producerConfig, mockTime.milliseconds());
- }
-
- private Topology setupTopologyWithIntermediateUserTopic(final String outputTopic2) {
- final StreamsBuilder builder = new StreamsBuilder();
-
- final KStream<Long, String> input = builder.stream(INPUT_TOPIC);
-
- // use map to trigger internal re-partitioning before groupByKey
- input.map(new KeyValueMapper<Long, String, KeyValue<Long, String>>() {
- @Override
- public KeyValue<Long, String> apply(final Long key, final String value) {
- return new KeyValue<>(key, value);
- }
- })
- .groupByKey()
- .count("global-count")
- .to(Serdes.Long(), Serdes.Long(), OUTPUT_TOPIC);
-
- input.through(INTERMEDIATE_USER_TOPIC)
- .groupByKey()
- .count(TimeWindows.of(35).advanceBy(10), "count")
- .toStream()
- .map(new KeyValueMapper<Windowed<Long>, Long, KeyValue<Long, Long>>() {
- @Override
- public KeyValue<Long, Long> apply(final Windowed<Long> key, final Long value) {
- return new KeyValue<>(key.window().start() + key.window().end(), value);
- }
- })
- .to(Serdes.Long(), Serdes.Long(), outputTopic2);
-
- return builder.build();
- }
-
- private Topology setupTopologyWithoutIntermediateUserTopic() {
- final StreamsBuilder builder = new StreamsBuilder();
-
- final KStream<Long, String> input = builder.stream(INPUT_TOPIC);
-
- // use map to trigger internal re-partitioning before groupByKey
- input.map(new KeyValueMapper<Long, String, KeyValue<Long, Long>>() {
- @Override
- public KeyValue<Long, Long> apply(final Long key, final String value) {
- return new KeyValue<>(key, key);
- }
- }).to(Serdes.Long(), Serdes.Long(), OUTPUT_TOPIC);
-
- return builder.build();
- }
-
- private void cleanGlobal(final String intermediateUserTopic) {
- // leaving --zookeeper arg here to ensure tool works if users add it
- final String[] parameters;
- if (intermediateUserTopic != null) {
- parameters = new String[]{
- "--application-id", APP_ID + testNo,
- "--bootstrap-servers", CLUSTER.bootstrapServers(),
- "--input-topics", INPUT_TOPIC,
- "--intermediate-topics", INTERMEDIATE_USER_TOPIC,
- "--zookeeper", "localhost:2181"
- };
- } else {
- parameters = new String[]{
- "--application-id", APP_ID + testNo,
- "--bootstrap-servers", CLUSTER.bootstrapServers(),
- "--input-topics", INPUT_TOPIC
- };
- }
- final Properties cleanUpConfig = new Properties();
- cleanUpConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 100);
- cleanUpConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "" + CLEANUP_CONSUMER_TIMEOUT);
-
- final int exitCode = new StreamsResetter().run(parameters, cleanUpConfig);
- Assert.assertEquals(0, exitCode);
- }
-
- private void assertInternalTopicsGotDeleted(final String intermediateUserTopic) throws Exception {
- final Set<String> expectedRemainingTopicsAfterCleanup = new HashSet<>();
- expectedRemainingTopicsAfterCleanup.add(INPUT_TOPIC);
- if (intermediateUserTopic != null) {
- expectedRemainingTopicsAfterCleanup.add(intermediateUserTopic);
- }
- expectedRemainingTopicsAfterCleanup.add(OUTPUT_TOPIC);
- expectedRemainingTopicsAfterCleanup.add(OUTPUT_TOPIC_2);
- expectedRemainingTopicsAfterCleanup.add(OUTPUT_TOPIC_2_RERUN);
- expectedRemainingTopicsAfterCleanup.add("__consumer_offsets");
-
- final Set<String> allTopics = new HashSet<>();
-
- final ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
- listTopicsOptions.listInternal(true);
- allTopics.addAll(kafkaAdminClient.listTopics(listTopicsOptions).names().get(30000, TimeUnit.MILLISECONDS));
- assertThat(allTopics, equalTo(expectedRemainingTopicsAfterCleanup));
-
- }
-
- private class WaitUntilConsumerGroupGotClosed implements TestCondition {
- @Override
- public boolean conditionMet() {
- return adminClient.describeConsumerGroup(APP_ID + testNo, 0).consumers().get().isEmpty();
- }
- }
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/f9865d52/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationWithSslTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationWithSslTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationWithSslTest.java
new file mode 100644
index 0000000..d018225
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationWithSslTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.integration;
+
+import kafka.server.KafkaConfig$;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.types.Password;
+import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.test.IntegrationTest;
+import org.apache.kafka.test.TestSslUtils;
+import org.apache.kafka.test.TestUtils;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Tests command line SSL setup for reset tool.
+ */
+@Category({IntegrationTest.class})
+public class ResetIntegrationWithSslTest extends AbstractResetIntegrationTest {
+
+ private static Map<String, Object> sslConfig;
+ static {
+ try {
+ sslConfig = TestSslUtils.createSslConfig(false, true, Mode.SERVER, TestUtils.tempFile(), "testCert");
+ } catch (final Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @ClassRule
+ public static final EmbeddedKafkaCluster CLUSTER;
+ static {
+ final Properties props = new Properties();
+ // we double the value passed to `time.sleep` in each iteration in one of the map functions, so we disable
+ // expiration of connections by the brokers to avoid errors when `AdminClient` sends requests after potentially
+ // very long sleep times
+ props.put(KafkaConfig$.MODULE$.ConnectionsMaxIdleMsProp(), -1L);
+ props.put(KafkaConfig$.MODULE$.ListenersProp(), "SSL://localhost:9092");
+ props.put(KafkaConfig$.MODULE$.InterBrokerListenerNameProp(), "SSL");
+ props.putAll(sslConfig);
+ // we align time to seconds to get clean window boundaries and thus ensure the same result for each run
+ // otherwise, input records could fall into different windows for different runs depending on the initial mock time
+ final long alignedTime = (System.currentTimeMillis() / 1000) * 1000;
+ CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS, props, alignedTime);
+ cluster = CLUSTER;
+ }
+
+ @AfterClass
+ public static void globalCleanup() {
+ afterClassGlobalCleanup();
+ }
+
+ @Before
+ public void before() throws Exception {
+ beforePrepareTest();
+ }
+
+ Properties getClientSslConfig() {
+ final Properties props = new Properties();
+
+ props.put("bootstrap.servers", CLUSTER.bootstrapServers());
+ props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, sslConfig.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG));
+ props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, ((Password) sslConfig.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG)).value());
+ props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
+
+ return props;
+ }
+
+ @Test
+ public void testReprocessingFromScratchAfterResetWithoutIntermediateUserTopic() throws Exception {
+ super.testReprocessingFromScratchAfterResetWithoutIntermediateUserTopic();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/f9865d52/streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java b/streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java
index 18c1995..1863484 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java
@@ -29,7 +29,6 @@ import kafka.utils.ZkUtils;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import org.apache.kafka.common.network.ListenerName;
-import org.apache.kafka.common.protocol.SecurityProtocol;
import org.junit.rules.TemporaryFolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -110,7 +109,9 @@ public class KafkaEmbedded {
* You can use this to tell Kafka producers and consumers how to connect to this instance.
*/
public String brokerList() {
- return kafka.config().hostName() + ":" + kafka.boundPort(ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT));
+ Object listenerConfig = effectiveConfig.get(KafkaConfig$.MODULE$.InterBrokerListenerNameProp());
+ return kafka.config().hostName() + ":" + kafka.boundPort(
+ new ListenerName(listenerConfig != null ? listenerConfig.toString() : "PLAINTEXT"));
}