Skip to content

MINOR: Update usage of deprecated API #6146

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjsax merged 4 commits into from
Jan 29, 2019
Merged

MINOR: Update usage of deprecated API #6146

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0881366
MINOR: Update usage of deprecated API
mjsax Jan 15, 2019
631115c
Github comments
mjsax Jan 22, 2019
40fc3a2
Github comments
mjsax Jan 24, 2019
d0a8b53
Fixed spotbugs error
mjsax Jan 26, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions checkstyle/import-control-core.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,9 +43,6 @@
<allow pkg="kafka.admin" />
<allow pkg="joptsimple" />
<allow pkg="org.apache.kafka.clients.consumer" />
<allow class="javax.xml.datatype.Duration" />
<allow class="javax.xml.datatype.DatatypeFactory" />
<allow class="javax.xml.datatype.DatatypeConfigurationException" />
</subpackage>

<subpackage name="coordinator">
Expand Down
12 changes: 6 additions & 6 deletions core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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._
Expand Down Expand Up @@ -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

@vvcephei vvcephei Jan 23, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this code covered by any tests? Duration and Instant are immutable, so I don't think this code would work.

What do you think about:

Suggested change
val timestamp = now.toEpochMilli
val timestamp = now.minus(durationParsed).toEpochMilli()

(also applies to the same change below)

@mjsax mjsax Jan 24, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this code covered by any tests?

Yes. Jenkins failed (I did not run the tests locally before I pushed...) -- Same issue in StreamsResetter.

val logTimestampOffsets = getLogTimestampOffsets(partitionsToReset, timestamp)
partitionsToReset.map { topicPartition =>
val logTimestampOffset = logTimestampOffsets.get(topicPartition)
Expand Down
54 changes: 34 additions & 20 deletions core/src/main/scala/kafka/tools/StreamsResetter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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.

@mjsax mjsax Jan 23, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just shorting some too long lines here -- not content change.

* {@link StreamsResetter} resets the processing state of a Kafka Streams application so that, for example,
* you can reprocess its input from scratch.
* <p>
* <strong>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.</strong>
* <strong>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.</strong>
* <p>
* Resetting the processing state of an application includes the following actions:
* <ol>
Expand All @@ -71,14 +73,17 @@
* <li>deleting any topics created internally by Kafka Streams for this application</li>
* </ol>
* <p>
* Do only use this tool if <strong>no</strong> 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 <strong>no</strong> application instance is running.
* Otherwise, the application will get into an invalid state and crash or produce wrong results.
* <p>
* 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.
* <p>
* 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 {
Expand Down Expand Up @@ -164,15 +169,17 @@ public int run(final String[] args,
e.printStackTrace(System.err);
} finally {
if (kafkaAdminClient != null) {
kafkaAdminClient.close(60, TimeUnit.SECONDS);
kafkaAdminClient.close(Duration.ofSeconds(60));
}
}

return exitCode;
}

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<MemberDescription> members =
Expand Down Expand Up @@ -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<String> inputTopics = options.valuesOf(inputTopicsOption);
final List<String> intermediateTopics = options.valuesOf(intermediateTopicsOption);
int topicNotFound = EXIT_CODE_SUCCESS;
Expand Down Expand Up @@ -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<byte[], byte[]> client = new KafkaConsumer<>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer())) {
try (final KafkaConsumer<byte[], byte[]> client =
new KafkaConsumer<>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer())) {

final Collection<TopicPartition> partitions = topicsToSubscribe.stream().map(client::partitionsFor)
.flatMap(Collection::stream)
.map(info -> new TopicPartition(info.topic(), info.partition()))
Expand Down Expand Up @@ -392,7 +404,7 @@ public void maybeSeekToEnd(final String groupId,
private void maybeReset(final String groupId,
final Consumer<byte[], byte[]> client,
final Set<TopicPartition> 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 + ")");
Expand All @@ -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));

@mjsax mjsax Jan 23, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the actual update to move to java.time.Duration

} else if (options.has(fromFileOption)) {
final String resetPlanPath = options.valueOf(fromFileOption);
final Map<TopicPartition, Long> topicPartitionsAndOffset = getTopicPartitionOffsetFromResetPlan(resetPlanPath);
final Map<TopicPartition, Long> topicPartitionsAndOffset =
getTopicPartitionOffsetFromResetPlan(resetPlanPath);
resetOffsetsFromResetPlan(client, inputTopicPartitions, topicPartitionsAndOffset);
} else {
client.seekToBeginning(inputTopicPartitions);
Expand All @@ -441,17 +453,19 @@ public void resetOffsetsFromResetPlan(final Consumer<byte[], byte[]> client,
}
}

private Map<TopicPartition, Long> getTopicPartitionOffsetFromResetPlan(final String resetPlanPath) throws IOException, ParseException {
private Map<TopicPartition, Long> getTopicPartitionOffsetFromResetPlan(final String resetPlanPath)
throws IOException, ParseException {

final String resetPlanCsv = Utils.readFileAsString(resetPlanPath);
return parseResetPlan(resetPlanCsv);
}

private void resetByDuration(final Consumer<byte[], byte[]> client,
final Set<TopicPartition> 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();

@mjsax mjsax Jan 23, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Part of actual update.


final Map<TopicPartition, Long> topicPartitionsAndTimes = new HashMap<>(inputTopicPartitions.size());
for (final TopicPartition topicPartition : inputTopicPartitions) {
Expand Down Expand Up @@ -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);
}
Expand Down