From 08813668b85421088cf0e0300b8574a0279eaf0b Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Tue, 15 Jan 2019 09:32:04 -0800 Subject: [PATCH 1/4] MINOR: Update usage of deprecated API --- core/src/main/scala/kafka/tools/StreamsResetter.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/tools/StreamsResetter.java b/core/src/main/scala/kafka/tools/StreamsResetter.java index 40d285afe846b..5f91deafd6979 100644 --- a/core/src/main/scala/kafka/tools/StreamsResetter.java +++ b/core/src/main/scala/kafka/tools/StreamsResetter.java @@ -164,7 +164,7 @@ public int run(final String[] args, e.printStackTrace(System.err); } finally { if (kafkaAdminClient != null) { - kafkaAdminClient.close(60, TimeUnit.SECONDS); + kafkaAdminClient.close(java.time.Duration.ofSeconds(60)); } } From 631115c977f643ba045550f8252fab9ccad6fb7c Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Mon, 21 Jan 2019 18:07:00 -0800 Subject: [PATCH 2/4] Github comments --- checkstyle/import-control-core.xml | 3 -- .../kafka/admin/ConsumerGroupCommand.scala | 12 ++--- .../scala/kafka/tools/StreamsResetter.java | 54 ++++++++++++------- 3 files changed, 40 insertions(+), 29 deletions(-) diff --git a/checkstyle/import-control-core.xml b/checkstyle/import-control-core.xml index f8b1c46ff384c..4898c0b50a73b 100644 --- a/checkstyle/import-control-core.xml +++ b/checkstyle/import-control-core.xml @@ -43,9 +43,6 @@ - - - diff --git a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala index f27f2c3021cdd..834cb31b40164 100755 --- a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala +++ b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala @@ -18,10 +18,10 @@ package kafka.admin import java.text.{ParseException, SimpleDateFormat} +import java.time.{Duration, Instant} import java.util -import java.util.{Date, Properties} +import java.util.Properties -import javax.xml.datatype.DatatypeFactory import joptsimple.OptionSpec import kafka.utils._ import org.apache.kafka.clients.admin._ @@ -553,10 +553,10 @@ object ConsumerGroupCommand extends Logging { }.toMap } else if (opts.options.has(opts.resetByDurationOpt)) { val duration = opts.options.valueOf(opts.resetByDurationOpt) - val durationParsed = DatatypeFactory.newInstance().newDuration(duration) - val now = new Date() - durationParsed.negate().addTo(now) - val timestamp = now.getTime + val durationParsed = Duration.parse(duration) + val now = Instant.now() + durationParsed.negated().addTo(now) + val timestamp = now.toEpochMilli val logTimestampOffsets = getLogTimestampOffsets(partitionsToReset, timestamp) partitionsToReset.map { topicPartition => val logTimestampOffset = logTimestampOffsets.get(topicPartition) diff --git a/core/src/main/scala/kafka/tools/StreamsResetter.java b/core/src/main/scala/kafka/tools/StreamsResetter.java index 5f91deafd6979..78b2706360fea 100644 --- a/core/src/main/scala/kafka/tools/StreamsResetter.java +++ b/core/src/main/scala/kafka/tools/StreamsResetter.java @@ -39,11 +39,11 @@ import org.apache.kafka.common.utils.Exit; import org.apache.kafka.common.utils.Utils; -import javax.xml.datatype.DatatypeFactory; -import javax.xml.datatype.Duration; import java.io.IOException; import java.text.ParseException; import java.text.SimpleDateFormat; +import java.time.Duration; +import java.time.Instant; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -60,9 +60,11 @@ import java.util.stream.Collectors; /** - * {@link StreamsResetter} resets the processing state of a Kafka Streams application so that, for example, you can reprocess its input from scratch. + * {@link StreamsResetter} resets the processing state of a Kafka Streams application so that, for example, + * you can reprocess its input from scratch. *

- * This class is not part of public API. For backward compatibility, use the provided script in "bin/" instead of calling this class directly from your code. + * This class is not part of public API. For backward compatibility, + * use the provided script in "bin/" instead of calling this class directly from your code. *

* Resetting the processing state of an application includes the following actions: *

    @@ -71,14 +73,17 @@ *
  1. deleting any topics created internally by Kafka Streams for this application
    2. *
*

- * Do only use this tool if no application instance is running. Otherwise, the application will get into an invalid state and crash or produce wrong results. + * Do only use this tool if no application instance is running. + * Otherwise, the application will get into an invalid state and crash or produce wrong results. *

* If you run multiple application instances, running this tool once is sufficient. - * However, you need to call {@code KafkaStreams#cleanUp()} before re-starting any instance (to clean local state store directory). + * However, you need to call {@code KafkaStreams#cleanUp()} before re-starting any instance + * (to clean local state store directory). * Otherwise, your application is in an invalid state. *

* User output topics will not be deleted or modified by this tool. - * If downstream applications consume intermediate or output topics, it is the user's responsibility to adjust those applications manually if required. + * If downstream applications consume intermediate or output topics, + * it is the user's responsibility to adjust those applications manually if required. */ @InterfaceStability.Unstable public class StreamsResetter { @@ -164,7 +169,7 @@ public int run(final String[] args, e.printStackTrace(System.err); } finally { if (kafkaAdminClient != null) { - kafkaAdminClient.close(java.time.Duration.ofSeconds(60)); + kafkaAdminClient.close(Duration.ofSeconds(60)); } } @@ -172,7 +177,9 @@ public int run(final String[] args, } private void validateNoActiveConsumers(final String groupId, - final AdminClient adminClient) throws ExecutionException, InterruptedException { + final AdminClient adminClient) + throws ExecutionException, InterruptedException { + final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(Collections.singleton(groupId), (new DescribeConsumerGroupsOptions()).timeoutMs(10 * 1000)); final List members = @@ -268,7 +275,10 @@ private void parseArguments(final String[] args) throws IOException { CommandLineUtils.checkInvalidArgs(optionParser, options, shiftByOption, allScenarioOptions.$minus(shiftByOption)); } - private int maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map consumerConfig, final boolean dryRun) throws Exception { + private int maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map consumerConfig, + final boolean dryRun) + throws IOException, ParseException { + final List inputTopics = options.valuesOf(inputTopicsOption); final List intermediateTopics = options.valuesOf(intermediateTopicsOption); int topicNotFound = EXIT_CODE_SUCCESS; @@ -334,7 +344,9 @@ private int maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map consum config.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId); config.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); - try (final KafkaConsumer client = new KafkaConsumer<>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer())) { + try (final KafkaConsumer client = + new KafkaConsumer<>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer())) { + final Collection partitions = topicsToSubscribe.stream().map(client::partitionsFor) .flatMap(Collection::stream) .map(info -> new TopicPartition(info.topic(), info.partition())) @@ -392,7 +404,7 @@ public void maybeSeekToEnd(final String groupId, private void maybeReset(final String groupId, final Consumer client, final Set inputTopicPartitions) - throws Exception { + throws IOException, ParseException { if (inputTopicPartitions.size() > 0) { System.out.println("Following input topics offsets will be reset to (for consumer group " + groupId + ")"); @@ -410,11 +422,11 @@ private void maybeReset(final String groupId, resetToDatetime(client, inputTopicPartitions, timestamp); } else if (options.has(byDurationOption)) { final String duration = options.valueOf(byDurationOption); - final Duration durationParsed = DatatypeFactory.newInstance().newDuration(duration); - resetByDuration(client, inputTopicPartitions, durationParsed); + resetByDuration(client, inputTopicPartitions, Duration.parse(duration)); } else if (options.has(fromFileOption)) { final String resetPlanPath = options.valueOf(fromFileOption); - final Map topicPartitionsAndOffset = getTopicPartitionOffsetFromResetPlan(resetPlanPath); + final Map topicPartitionsAndOffset = + getTopicPartitionOffsetFromResetPlan(resetPlanPath); resetOffsetsFromResetPlan(client, inputTopicPartitions, topicPartitionsAndOffset); } else { client.seekToBeginning(inputTopicPartitions); @@ -441,7 +453,9 @@ public void resetOffsetsFromResetPlan(final Consumer client, } } - private Map getTopicPartitionOffsetFromResetPlan(final String resetPlanPath) throws IOException, ParseException { + private Map getTopicPartitionOffsetFromResetPlan(final String resetPlanPath) + throws IOException, ParseException { + final String resetPlanCsv = Utils.readFileAsString(resetPlanPath); return parseResetPlan(resetPlanCsv); } @@ -449,9 +463,9 @@ private Map getTopicPartitionOffsetFromResetPlan(final Str private void resetByDuration(final Consumer client, final Set inputTopicPartitions, final Duration duration) { - final Date now = new Date(); - duration.negate().addTo(now); - final long timestamp = now.getTime(); + final Instant now = Instant.now(); + duration.negated().addTo(now); + final long timestamp = now.toEpochMilli(); final Map topicPartitionsAndTimes = new HashMap<>(inputTopicPartitions.size()); for (final TopicPartition topicPartition : inputTopicPartitions) { @@ -647,7 +661,7 @@ private boolean isInternalTopic(final String topicName) { && (topicName.endsWith("-changelog") || topicName.endsWith("-repartition")); } - private void printHelp(OptionParser parser) throws IOException { + private void printHelp(final OptionParser parser) throws IOException { System.err.println(usage); parser.printHelpOn(System.err); } From 40fc3a276d6fde96546a5e9f9bf8b7863700d6be Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Thu, 24 Jan 2019 10:31:22 -0800 Subject: [PATCH 3/4] Github comments --- core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala | 3 +-- core/src/main/scala/kafka/tools/StreamsResetter.java | 5 ++--- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala index 834cb31b40164..b40a33b980d87 100755 --- a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala +++ b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala @@ -555,8 +555,7 @@ object ConsumerGroupCommand extends Logging { val duration = opts.options.valueOf(opts.resetByDurationOpt) val durationParsed = Duration.parse(duration) val now = Instant.now() - durationParsed.negated().addTo(now) - val timestamp = now.toEpochMilli + val timestamp = now.minus(durationParsed).toEpochMilli val logTimestampOffsets = getLogTimestampOffsets(partitionsToReset, timestamp) partitionsToReset.map { topicPartition => val logTimestampOffset = logTimestampOffsets.get(topicPartition) diff --git a/core/src/main/scala/kafka/tools/StreamsResetter.java b/core/src/main/scala/kafka/tools/StreamsResetter.java index 78b2706360fea..e79ccda1d648d 100644 --- a/core/src/main/scala/kafka/tools/StreamsResetter.java +++ b/core/src/main/scala/kafka/tools/StreamsResetter.java @@ -257,7 +257,7 @@ private void parseArguments(final String[] args) throws IOException { CommandLineUtils.printUsageAndDie(optionParser, "Only one of --dry-run and --execute can be specified"); } - final scala.collection.immutable.HashSet> allScenarioOptions = new scala.collection.immutable.HashSet<>(); + final scala.collection.immutable.HashSet> allScenarioOptions = new scala.collection.immutable.HashSet>(); allScenarioOptions.$plus(toOffsetOption); allScenarioOptions.$plus(toDatetimeOption); allScenarioOptions.$plus(byDurationOption); @@ -464,8 +464,7 @@ private void resetByDuration(final Consumer client, final Set inputTopicPartitions, final Duration duration) { final Instant now = Instant.now(); - duration.negated().addTo(now); - final long timestamp = now.toEpochMilli(); + final long timestamp = now.minus(duration).toEpochMilli(); final Map topicPartitionsAndTimes = new HashMap<>(inputTopicPartitions.size()); for (final TopicPartition topicPartition : inputTopicPartitions) { From d0a8b53084ae1893a658712c53f91d0664fac785 Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Sat, 26 Jan 2019 00:13:40 -0800 Subject: [PATCH 4/4] Fixed spotbugs error --- core/src/main/scala/kafka/tools/StreamsResetter.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/tools/StreamsResetter.java b/core/src/main/scala/kafka/tools/StreamsResetter.java index e79ccda1d648d..ddb140612d66f 100644 --- a/core/src/main/scala/kafka/tools/StreamsResetter.java +++ b/core/src/main/scala/kafka/tools/StreamsResetter.java @@ -377,7 +377,7 @@ private int maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map consum } client.commitSync(); } - } catch (final Exception e) { + } catch (final IOException | ParseException e) { System.err.println("ERROR: Resetting offsets failed."); throw e; }