From 474bfb40c0c229f7c67c63c26602eb1b111a7708 Mon Sep 17 00:00:00 2001 From: Anna Povzner Date: Sun, 9 Sep 2018 00:35:58 -0700 Subject: [PATCH 1/6] KAFKA-7044: Fix early return from fetchOffsetsByTimes and fix NPE in describe consumer group --- .../clients/consumer/internals/Fetcher.java | 36 ++++++++++++---- .../consumer/internals/FetcherTest.java | 43 +++++++++++++++++++ .../kafka/admin/ConsumerGroupCommand.scala | 20 ++++----- 3 files changed, 80 insertions(+), 19 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index dc0daa233abb6..0544a26f76043 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -411,10 +411,11 @@ private ListOffsetResult fetchOffsetsByTimes(Map timestamp if (future.succeeded()) { ListOffsetResult value = future.value(); result.fetchedOffsets.putAll(value.fetchedOffsets); - if (value.partitionsToRetry.isEmpty()) - return result; remainingToSearch.keySet().removeAll(result.fetchedOffsets.keySet()); + remainingToSearch.keySet().removeAll(value.partitionsWithUnknownOffset); + if (value.partitionsToRetry.isEmpty() && remainingToSearch.isEmpty()) + return result; } else if (!future.isRetriable()) { throw future.exception(); } @@ -440,6 +441,7 @@ private Map beginningOrEndOffset(Collection timestampsToSearch = new HashMap<>(); + for (TopicPartition tp : partitions) timestampsToSearch.put(tp, timestamp); Map offsets = new HashMap<>(); @@ -632,6 +634,7 @@ private RequestFuture sendListOffsetsRequests(final Map listOffsetRequestsFuture = new RequestFuture<>(); final Map fetchedTimestampOffsets = new HashMap<>(); final Set partitionsToRetry = new HashSet<>(); + final Set partitionsWithUnknownOffset = new HashSet<>(); final AtomicInteger remainingResponses = new AtomicInteger(timestampsToSearchByNode.size()); for (Map.Entry> entry : timestampsToSearchByNode.entrySet()) { @@ -643,9 +646,11 @@ public void onSuccess(ListOffsetResult partialResult) { synchronized (listOffsetRequestsFuture) { fetchedTimestampOffsets.putAll(partialResult.fetchedOffsets); partitionsToRetry.addAll(partialResult.partitionsToRetry); + partitionsWithUnknownOffset.addAll(partialResult.partitionsWithUnknownOffset); if (remainingResponses.decrementAndGet() == 0 && !listOffsetRequestsFuture.isDone()) { - ListOffsetResult result = new ListOffsetResult(fetchedTimestampOffsets, partitionsToRetry); + ListOffsetResult result = new ListOffsetResult( + fetchedTimestampOffsets, partitionsToRetry, partitionsWithUnknownOffset); listOffsetRequestsFuture.complete(result); } } @@ -740,6 +745,7 @@ private void handleListOffsetResponse(Map future) { Map fetchedOffsets = new HashMap<>(); Set partitionsToRetry = new HashSet<>(); + Set partitionsWithUnknownOffset = new HashSet<>(); Set unauthorizedTopics = new HashSet<>(); for (Map.Entry entry : timestampsToSearch.entrySet()) { @@ -764,6 +770,8 @@ private void handleListOffsetResponse(Map fetchedOffsets; private final Set partitionsToRetry; + private final Set partitionsWithUnknownOffset; + + public ListOffsetResult() { + this(new HashMap<>(), new HashSet<>(), new HashSet<>()); + } public ListOffsetResult(Map fetchedOffsets, Set partitionsNeedingRetry) { - this.fetchedOffsets = fetchedOffsets; - this.partitionsToRetry = partitionsNeedingRetry; + this(fetchedOffsets, partitionsNeedingRetry, new HashSet<>()); } - public ListOffsetResult() { - this.fetchedOffsets = new HashMap<>(); - this.partitionsToRetry = new HashSet<>(); + public ListOffsetResult(Map fetchedOffsets, + Set partitionsNeedingRetry, + Set partitionsWithUnknownOffset) { + this.fetchedOffsets = fetchedOffsets; + this.partitionsToRetry = partitionsNeedingRetry; + this.partitionsWithUnknownOffset = partitionsWithUnknownOffset; } } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java index d314a4d8c9feb..5edffa9dd939c 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java @@ -108,6 +108,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -1919,6 +1920,48 @@ public void testGetOffsetsForTimes() { testGetOffsetsForTimesWithError(Errors.BROKER_NOT_AVAILABLE, Errors.NONE, 10L, 100L, 10L, 100L); } + @Test + public void testGetOffsetsForTimesWhenSomeTopicPartitionLeadersNotKnownInitially() { + final String anotherTopic = "another-topic"; + final TopicPartition t2p0 = new TopicPartition(anotherTopic, 0); + + client.reset(); + + // Metadata initially has one topic + Cluster cluster = TestUtils.clusterWith(3, topicName, 2); + metadata.update(cluster, Collections.emptySet(), time.milliseconds()); + + // The first metadata refresh should contain one topic + client.prepareMetadataUpdate(cluster, Collections.emptySet(), false); + client.prepareResponseFrom(listOffsetResponse(tp0, Errors.NONE, 1000L, 11L), cluster.leaderFor(tp0)); + client.prepareResponseFrom(listOffsetResponse(tp1, Errors.NONE, 1000L, 32L), cluster.leaderFor(tp1)); + + // Second metadata refresh should contain two topics + Map partitionNumByTopic = new HashMap<>(); + partitionNumByTopic.put(topicName, 2); + partitionNumByTopic.put(anotherTopic, 1); + Cluster updatedCluster = TestUtils.clusterWith(3, partitionNumByTopic); + client.prepareMetadataUpdate(updatedCluster, Collections.emptySet(), false); + client.prepareResponseFrom(listOffsetResponse(t2p0, Errors.NONE, 1000L, 54L), cluster.leaderFor(t2p0)); + + Map timestampToSearch = new HashMap<>(); + timestampToSearch.put(tp0, ListOffsetRequest.LATEST_TIMESTAMP); + timestampToSearch.put(tp1, ListOffsetRequest.LATEST_TIMESTAMP); + timestampToSearch.put(t2p0, ListOffsetRequest.LATEST_TIMESTAMP); + Map offsetAndTimestampMap = + fetcher.offsetsByTimes(timestampToSearch, time.timer(Long.MAX_VALUE)); + + assertNotNull("Expect Fetcher.offsetsByTimes() to return non-null result for " + tp0, + offsetAndTimestampMap.get(tp0)); + assertNotNull("Expect Fetcher.offsetsByTimes() to return non-null result for " + tp1, + offsetAndTimestampMap.get(tp1)); + assertNotNull("Expect Fetcher.offsetsByTimes() to return non-null result for " + t2p0, + offsetAndTimestampMap.get(t2p0)); + assertEquals(11L, offsetAndTimestampMap.get(tp0).offset()); + assertEquals(32L, offsetAndTimestampMap.get(tp1).offset()); + assertEquals(54L, offsetAndTimestampMap.get(t2p0).offset()); + } + @Test(expected = TimeoutException.class) public void testBatchedListOffsetsMetadataErrors() { Map partitionData = new HashMap<>(); diff --git a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala index 1d61720bfa92d..c0f6797eddd07 100755 --- a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala +++ b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala @@ -292,12 +292,8 @@ object ConsumerGroupCommand extends Logging { } getLogEndOffsets(topicPartitions).map { - logEndOffsetResult => - logEndOffsetResult._2 match { - case LogOffsetResult.LogOffset(logEndOffset) => getDescribePartitionResult(logEndOffsetResult._1, Some(logEndOffset)) - case LogOffsetResult.Unknown => getDescribePartitionResult(logEndOffsetResult._1, None) - case LogOffsetResult.Ignore => null - } + case (topicPartition, LogOffsetResult.LogOffset(offset)) => getDescribePartitionResult(topicPartition, Some(offset)) + case (topicPartition, _) => getDescribePartitionResult(topicPartition, None) }.toArray } @@ -399,16 +395,20 @@ object ConsumerGroupCommand extends Logging { private def getLogEndOffsets(topicPartitions: Seq[TopicPartition]): Map[TopicPartition, LogOffsetResult] = { val offsets = getConsumer.endOffsets(topicPartitions.asJava) topicPartitions.map { topicPartition => - val logEndOffset = offsets.get(topicPartition) - topicPartition -> LogOffsetResult.LogOffset(logEndOffset) + Option(offsets.get(topicPartition)) match { + case Some(logEndOffset) => topicPartition -> LogOffsetResult.LogOffset(logEndOffset) + case _ => topicPartition -> LogOffsetResult.Unknown + } }.toMap } private def getLogStartOffsets(topicPartitions: Seq[TopicPartition]): Map[TopicPartition, LogOffsetResult] = { val offsets = getConsumer.beginningOffsets(topicPartitions.asJava) topicPartitions.map { topicPartition => - val logStartOffset = offsets.get(topicPartition) - topicPartition -> LogOffsetResult.LogOffset(logStartOffset) + Option(offsets.get(topicPartition)) match { + case Some(logStartOffset) => topicPartition -> LogOffsetResult.LogOffset(logStartOffset) + case _ => topicPartition -> LogOffsetResult.Unknown + } }.toMap } From 33068498816b318508f4f0afa96e2f799e8896c7 Mon Sep 17 00:00:00 2001 From: Anna Povzner Date: Sun, 9 Sep 2018 18:52:17 -0700 Subject: [PATCH 2/6] New test exceeded default max number of lines in the FetcherTest class --- checkstyle/suppressions.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index 3ed4a9c1f3f67..2fa499c21655e 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -70,7 +70,7 @@ files="MockAdminClient.java"/> + files="RequestResponseTest.java|FetcherTest.java"/> From 4f1a6a05c091c68088e49e9cd1185ae71289389e Mon Sep 17 00:00:00 2001 From: Anna Povzner Date: Sun, 9 Sep 2018 19:57:06 -0700 Subject: [PATCH 3/6] addressed review comments --- .../clients/consumer/internals/Fetcher.java | 44 +++++++------------ 1 file changed, 17 insertions(+), 27 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 0544a26f76043..41639d20ab288 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -88,6 +88,7 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; import static java.util.Collections.emptyList; import static org.apache.kafka.common.serialization.ExtendedDeserializer.Wrapper.ensureExtended; @@ -411,11 +412,12 @@ private ListOffsetResult fetchOffsetsByTimes(Map timestamp if (future.succeeded()) { ListOffsetResult value = future.value(); result.fetchedOffsets.putAll(value.fetchedOffsets); - - remainingToSearch.keySet().removeAll(result.fetchedOffsets.keySet()); - remainingToSearch.keySet().removeAll(value.partitionsWithUnknownOffset); - if (value.partitionsToRetry.isEmpty() && remainingToSearch.isEmpty()) + if (value.partitionsToRetry.isEmpty()) return result; + + remainingToSearch = timestampsToSearch.entrySet().stream() + .filter(entry -> value.partitionsToRetry.contains(entry.getKey())) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); } else if (!future.isRetriable()) { throw future.exception(); } @@ -634,10 +636,11 @@ private RequestFuture sendListOffsetsRequests(final Map listOffsetRequestsFuture = new RequestFuture<>(); final Map fetchedTimestampOffsets = new HashMap<>(); final Set partitionsToRetry = new HashSet<>(); - final Set partitionsWithUnknownOffset = new HashSet<>(); + final Set partitionsRequireMetadataUpdate = new HashSet<>(timestampsToSearch.keySet()); final AtomicInteger remainingResponses = new AtomicInteger(timestampsToSearchByNode.size()); for (Map.Entry> entry : timestampsToSearchByNode.entrySet()) { + partitionsRequireMetadataUpdate.removeAll(entry.getValue().keySet()); RequestFuture future = sendListOffsetRequest(entry.getKey(), entry.getValue(), requireTimestamps); future.addListener(new RequestFutureListener() { @@ -646,11 +649,10 @@ public void onSuccess(ListOffsetResult partialResult) { synchronized (listOffsetRequestsFuture) { fetchedTimestampOffsets.putAll(partialResult.fetchedOffsets); partitionsToRetry.addAll(partialResult.partitionsToRetry); - partitionsWithUnknownOffset.addAll(partialResult.partitionsWithUnknownOffset); if (remainingResponses.decrementAndGet() == 0 && !listOffsetRequestsFuture.isDone()) { - ListOffsetResult result = new ListOffsetResult( - fetchedTimestampOffsets, partitionsToRetry, partitionsWithUnknownOffset); + partitionsToRetry.addAll(partitionsRequireMetadataUpdate); + ListOffsetResult result = new ListOffsetResult(fetchedTimestampOffsets, partitionsToRetry); listOffsetRequestsFuture.complete(result); } } @@ -665,6 +667,7 @@ public void onFailure(RuntimeException e) { } }); } + return listOffsetRequestsFuture; } @@ -745,7 +748,6 @@ private void handleListOffsetResponse(Map future) { Map fetchedOffsets = new HashMap<>(); Set partitionsToRetry = new HashSet<>(); - Set partitionsWithUnknownOffset = new HashSet<>(); Set unauthorizedTopics = new HashSet<>(); for (Map.Entry entry : timestampsToSearch.entrySet()) { @@ -770,8 +772,6 @@ private void handleListOffsetResponse(Map fetchedOffsets; private final Set partitionsToRetry; - private final Set partitionsWithUnknownOffset; - - public ListOffsetResult() { - this(new HashMap<>(), new HashSet<>(), new HashSet<>()); - } public ListOffsetResult(Map fetchedOffsets, Set partitionsNeedingRetry) { - this(fetchedOffsets, partitionsNeedingRetry, new HashSet<>()); - } - - public ListOffsetResult(Map fetchedOffsets, - Set partitionsNeedingRetry, - Set partitionsWithUnknownOffset) { this.fetchedOffsets = fetchedOffsets; this.partitionsToRetry = partitionsNeedingRetry; - this.partitionsWithUnknownOffset = partitionsWithUnknownOffset; + } + + public ListOffsetResult() { + this.fetchedOffsets = new HashMap<>(); + this.partitionsToRetry = new HashSet<>(); } } From aec7385a253d930c0490a707a3f1382c13612356 Mon Sep 17 00:00:00 2001 From: Anna Povzner Date: Sun, 9 Sep 2018 19:59:59 -0700 Subject: [PATCH 4/6] cleanup --- .../org/apache/kafka/clients/consumer/internals/Fetcher.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 41639d20ab288..87fe0f432ff46 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -443,7 +443,6 @@ private Map beginningOrEndOffset(Collection timestampsToSearch = new HashMap<>(); - for (TopicPartition tp : partitions) timestampsToSearch.put(tp, timestamp); Map offsets = new HashMap<>(); @@ -667,7 +666,6 @@ public void onFailure(RuntimeException e) { } }); } - return listOffsetRequestsFuture; } From e71e6f4ef84738849da7ec56d2a7894f60ecc081 Mon Sep 17 00:00:00 2001 From: Anna Povzner Date: Mon, 10 Sep 2018 14:51:00 -0700 Subject: [PATCH 5/6] passing partitions to retry to groupListOffsetRequests --- .../clients/consumer/internals/Fetcher.java | 34 ++++++++++++------- 1 file changed, 21 insertions(+), 13 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 87fe0f432ff46..8537755ce5941 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -88,7 +88,6 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; -import java.util.stream.Collectors; import static java.util.Collections.emptyList; import static org.apache.kafka.common.serialization.ExtendedDeserializer.Wrapper.ensureExtended; @@ -415,9 +414,7 @@ private ListOffsetResult fetchOffsetsByTimes(Map timestamp if (value.partitionsToRetry.isEmpty()) return result; - remainingToSearch = timestampsToSearch.entrySet().stream() - .filter(entry -> value.partitionsToRetry.contains(entry.getKey())) - .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + remainingToSearch.keySet().retainAll(value.partitionsToRetry); } else if (!future.isRetriable()) { throw future.exception(); } @@ -577,8 +574,11 @@ private void resetOffsetsAsync(Map partitionResetTimestamp for (TopicPartition tp : partitionResetTimestamps.keySet()) metadata.add(tp.topic()); + final Set partitionsToRetry = new HashSet<>(); Map> timestampsToSearchByNode = - groupListOffsetRequests(partitionResetTimestamps); + groupListOffsetRequests(partitionResetTimestamps, partitionsToRetry); + if (!partitionsToRetry.isEmpty()) + log.info("Skipping reset of partitions {} that need metadata update or re-connect to leader", partitionsToRetry); for (Map.Entry> entry : timestampsToSearchByNode.entrySet()) { Node node = entry.getKey(); final Map resetTimestamps = entry.getValue(); @@ -627,21 +627,19 @@ private RequestFuture sendListOffsetsRequests(final Map partitionsToRetry = new HashSet<>(); Map> timestampsToSearchByNode = - groupListOffsetRequests(timestampsToSearch); + groupListOffsetRequests(timestampsToSearch, partitionsToRetry); if (timestampsToSearchByNode.isEmpty()) return RequestFuture.failure(new StaleMetadataException()); final RequestFuture listOffsetRequestsFuture = new RequestFuture<>(); final Map fetchedTimestampOffsets = new HashMap<>(); - final Set partitionsToRetry = new HashSet<>(); - final Set partitionsRequireMetadataUpdate = new HashSet<>(timestampsToSearch.keySet()); final AtomicInteger remainingResponses = new AtomicInteger(timestampsToSearchByNode.size()); for (Map.Entry> entry : timestampsToSearchByNode.entrySet()) { - partitionsRequireMetadataUpdate.removeAll(entry.getValue().keySet()); RequestFuture future = - sendListOffsetRequest(entry.getKey(), entry.getValue(), requireTimestamps); + sendListOffsetRequest(entry.getKey(), entry.getValue(), requireTimestamps); future.addListener(new RequestFutureListener() { @Override public void onSuccess(ListOffsetResult partialResult) { @@ -650,7 +648,6 @@ public void onSuccess(ListOffsetResult partialResult) { partitionsToRetry.addAll(partialResult.partitionsToRetry); if (remainingResponses.decrementAndGet() == 0 && !listOffsetRequestsFuture.isDone()) { - partitionsToRetry.addAll(partitionsRequireMetadataUpdate); ListOffsetResult result = new ListOffsetResult(fetchedTimestampOffsets, partitionsToRetry); listOffsetRequestsFuture.complete(result); } @@ -669,8 +666,16 @@ public void onFailure(RuntimeException e) { return listOffsetRequestsFuture; } + /** + * Groups timestamps to search by node for topic partitions in `timestampsToSearch` that have + * leaders available. Topic partitions from `timestampsToSearch` that do not have their leader + * available are added to `partitionsToRetry` + * @param timestampsToSearch The mapping from partitions ot the target timestamps + * @param partitionsToRetry A set of topic partitions that will be extended with partitions + * that need metadata update or re-connect to the leader. + */ private Map> groupListOffsetRequests( - Map timestampsToSearch) { + Map timestampsToSearch, Set partitionsToRetry) { final Map> timestampsToSearchByNode = new HashMap<>(); for (Map.Entry entry: timestampsToSearch.entrySet()) { TopicPartition tp = entry.getKey(); @@ -679,9 +684,11 @@ private Map> groupLis metadata.add(tp.topic()); log.debug("Leader for partition {} is unknown for fetching offset", tp); metadata.requestUpdate(); + partitionsToRetry.add(tp); } else if (info.leader() == null) { log.debug("Leader for partition {} is unavailable for fetching offset", tp); metadata.requestUpdate(); + partitionsToRetry.add(tp); } else if (client.isUnavailable(info.leader())) { client.maybeThrowAuthFailure(info.leader()); @@ -689,7 +696,8 @@ private Map> groupLis // try again. No need to request a metadata update since the disconnect will have // done so already. log.debug("Leader {} for partition {} is unavailable for fetching offset until reconnect backoff expires", - info.leader(), tp); + info.leader(), tp); + partitionsToRetry.add(tp); } else { Node node = info.leader(); Map topicData = From 1f8fbd8a0a66124cffd3a95a15fe8464dd3ae92b Mon Sep 17 00:00:00 2001 From: Anna Povzner Date: Mon, 10 Sep 2018 16:39:31 -0700 Subject: [PATCH 6/6] removed printing partitions to retry in resetOffsetsAsync --- .../org/apache/kafka/clients/consumer/internals/Fetcher.java | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 8537755ce5941..9fa64462d85d0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -574,11 +574,8 @@ private void resetOffsetsAsync(Map partitionResetTimestamp for (TopicPartition tp : partitionResetTimestamps.keySet()) metadata.add(tp.topic()); - final Set partitionsToRetry = new HashSet<>(); Map> timestampsToSearchByNode = - groupListOffsetRequests(partitionResetTimestamps, partitionsToRetry); - if (!partitionsToRetry.isEmpty()) - log.info("Skipping reset of partitions {} that need metadata update or re-connect to leader", partitionsToRetry); + groupListOffsetRequests(partitionResetTimestamps, new HashSet<>()); for (Map.Entry> entry : timestampsToSearchByNode.entrySet()) { Node node = entry.getKey(); final Map resetTimestamps = entry.getValue();