Skip to content

KAFKA-7044: Fix Fetcher.fetchOffsetsByTimes and NPE in describe consumer group #5627

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
hachikuji merged 6 commits into from
Sep 11, 2018
Merged

KAFKA-7044: Fix Fetcher.fetchOffsetsByTimes and NPE in describe consumer group #5627

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
474bfb4
KAFKA-7044: Fix early return from fetchOffsetsByTimes and fix NPE in …
apovzner Sep 9, 2018
3306849
New test exceeded default max number of lines in the FetcherTest class
apovzner Sep 10, 2018
4f1a6a0
addressed review comments
apovzner Sep 10, 2018
aec7385
cleanup
apovzner Sep 10, 2018
e71e6f4
passing partitions to retry to groupListOffsetRequests
apovzner Sep 10, 2018
1f8fbd8
removed printing partitions to retry in resetOffsetsAsync
apovzner Sep 10, 2018
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
2 changes: 1 addition & 1 deletion checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@
files="MockAdminClient.java"/>

<suppress checks="JavaNCSS"
files="RequestResponseTest.java"/>
files="RequestResponseTest.java|FetcherTest.java"/>

<suppress checks="NPathComplexity"
files="MemoryRecordsTest|MetricsTest"/>
Expand Down
25 changes: 18 additions & 7 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -414,7 +414,7 @@ private ListOffsetResult fetchOffsetsByTimes(Map<TopicPartition, Long> timestamp
if (value.partitionsToRetry.isEmpty())
return result;

remainingToSearch.keySet().removeAll(result.fetchedOffsets.keySet());
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
} else if (!future.isRetriable()) {
throw future.exception();
}
Expand Down Expand Up @@ -575,7 +575,7 @@ private void resetOffsetsAsync(Map<TopicPartition, Long> partitionResetTimestamp
metadata.add(tp.topic());

Map<Node, Map<TopicPartition, ListOffsetRequest.PartitionData>> timestampsToSearchByNode =
groupListOffsetRequests(partitionResetTimestamps);
groupListOffsetRequests(partitionResetTimestamps, new HashSet<>());
for (Map.Entry<Node, Map<TopicPartition, ListOffsetRequest.PartitionData>> entry : timestampsToSearchByNode.entrySet()) {
Node node = entry.getKey();
final Map<TopicPartition, ListOffsetRequest.PartitionData> resetTimestamps = entry.getValue();
Expand Down Expand Up @@ -624,19 +624,19 @@ private RequestFuture<ListOffsetResult> sendListOffsetsRequests(final Map<TopicP
for (TopicPartition tp : timestampsToSearch.keySet())
metadata.add(tp.topic());

final Set<TopicPartition> partitionsToRetry = new HashSet<>();
Map<Node, Map<TopicPartition, ListOffsetRequest.PartitionData>> timestampsToSearchByNode =
groupListOffsetRequests(timestampsToSearch);
groupListOffsetRequests(timestampsToSearch, partitionsToRetry);
if (timestampsToSearchByNode.isEmpty())
return RequestFuture.failure(new StaleMetadataException());

final RequestFuture<ListOffsetResult> listOffsetRequestsFuture = new RequestFuture<>();
final Map<TopicPartition, OffsetData> fetchedTimestampOffsets = new HashMap<>();
final Set<TopicPartition> partitionsToRetry = new HashSet<>();
final AtomicInteger remainingResponses = new AtomicInteger(timestampsToSearchByNode.size());

for (Map.Entry<Node, Map<TopicPartition, ListOffsetRequest.PartitionData>> entry : timestampsToSearchByNode.entrySet()) {
RequestFuture<ListOffsetResult> future =
sendListOffsetRequest(entry.getKey(), entry.getValue(), requireTimestamps);
sendListOffsetRequest(entry.getKey(), entry.getValue(), requireTimestamps);
future.addListener(new RequestFutureListener<ListOffsetResult>() {
@Override
public void onSuccess(ListOffsetResult partialResult) {
Expand All @@ -663,8 +663,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<Node, Map<TopicPartition, ListOffsetRequest.PartitionData>> groupListOffsetRequests(
Map<TopicPartition, Long> timestampsToSearch) {
Map<TopicPartition, Long> timestampsToSearch, Set<TopicPartition> partitionsToRetry) {
final Map<Node, Map<TopicPartition, ListOffsetRequest.PartitionData>> timestampsToSearchByNode = new HashMap<>();
for (Map.Entry<TopicPartition, Long> entry: timestampsToSearch.entrySet()) {
TopicPartition tp = entry.getKey();
Expand All @@ -673,17 +681,20 @@ private Map<Node, Map<TopicPartition, ListOffsetRequest.PartitionData>> 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());

// The connection has failed and we need to await the blackout period before we can
// 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);

@apovzner apovzner Sep 10, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I moved the logic for getting the list of partitions that do not have available leader to groupListOffsetRequests . I was contemplating whether we should add partition for which we have a failed connection to leader to partitionsToRetry. Seems like we should, because we may be able to reconnect before the list offsets timeout.

} else {
Node node = info.leader();
Map<TopicPartition, ListOffsetRequest.PartitionData> topicData =
Expand Down
43 changes: 43 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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.<String>emptySet(), time.milliseconds());

// The first metadata refresh should contain one topic
client.prepareMetadataUpdate(cluster, Collections.<String>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<String, Integer> partitionNumByTopic = new HashMap<>();
partitionNumByTopic.put(topicName, 2);
partitionNumByTopic.put(anotherTopic, 1);
Cluster updatedCluster = TestUtils.clusterWith(3, partitionNumByTopic);
client.prepareMetadataUpdate(updatedCluster, Collections.<String>emptySet(), false);
client.prepareResponseFrom(listOffsetResponse(t2p0, Errors.NONE, 1000L, 54L), cluster.leaderFor(t2p0));

Map<TopicPartition, Long> timestampToSearch = new HashMap<>();
timestampToSearch.put(tp0, ListOffsetRequest.LATEST_TIMESTAMP);
timestampToSearch.put(tp1, ListOffsetRequest.LATEST_TIMESTAMP);
timestampToSearch.put(t2p0, ListOffsetRequest.LATEST_TIMESTAMP);
Map<TopicPartition, OffsetAndTimestamp> 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<TopicPartition, ListOffsetResponse.PartitionData> partitionData = new HashMap<>();
Expand Down
20 changes: 10 additions & 10 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 @@ -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

@hachikuji hachikuji Sep 11, 2018

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.

As far as I can tell, this was the only use of LogOffsetResult.Ignore. Maybe we can get rid of it?

If we did that, then we could probably also get rid of LogOffsetResult and replace it with a simple Option[Long], but we can leave that for a separate PR.

}
case (topicPartition, LogOffsetResult.LogOffset(offset)) => getDescribePartitionResult(topicPartition, Some(offset))
case (topicPartition, _) => getDescribePartitionResult(topicPartition, None)
}.toArray
}

Expand Down Expand Up @@ -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
}

Expand Down