You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2020/12/10 03:09:31 UTC
[kafka] branch trunk updated: MINOR: remove duplicate code from
resetByDuration (#9699)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new db79f86 MINOR: remove duplicate code from resetByDuration (#9699)
db79f86 is described below
commit db79f86025683346e7a960fbf1ebc7b3b86b3c65
Author: bertber <75...@users.noreply.github.com>
AuthorDate: Thu Dec 10 11:08:17 2020 +0800
MINOR: remove duplicate code from resetByDuration (#9699)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
core/src/main/scala/kafka/tools/StreamsResetter.java | 14 +-------------
1 file changed, 1 insertion(+), 13 deletions(-)
diff --git a/core/src/main/scala/kafka/tools/StreamsResetter.java b/core/src/main/scala/kafka/tools/StreamsResetter.java
index d356681..f72a3b6 100644
--- a/core/src/main/scala/kafka/tools/StreamsResetter.java
+++ b/core/src/main/scala/kafka/tools/StreamsResetter.java
@@ -489,19 +489,7 @@ public class StreamsResetter {
private void resetByDuration(final Consumer<byte[], byte[]> client,
final Set<TopicPartition> inputTopicPartitions,
final Duration duration) {
- final Instant now = Instant.now();
- final long timestamp = now.minus(duration).toEpochMilli();
-
- final Map<TopicPartition, Long> topicPartitionsAndTimes = new HashMap<>(inputTopicPartitions.size());
- for (final TopicPartition topicPartition : inputTopicPartitions) {
- topicPartitionsAndTimes.put(topicPartition, timestamp);
- }
-
- final Map<TopicPartition, OffsetAndTimestamp> topicPartitionsAndOffset = client.offsetsForTimes(topicPartitionsAndTimes);
-
- for (final TopicPartition topicPartition : inputTopicPartitions) {
- client.seek(topicPartition, topicPartitionsAndOffset.get(topicPartition).offset());
- }
+ resetToDatetime(client, inputTopicPartitions, Instant.now().minus(duration).toEpochMilli());
}
private void resetToDatetime(final Consumer<byte[], byte[]> client,