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 4 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
8 changes: 7 additions & 1 deletion 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 @@ -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;
Expand Down Expand Up @@ -414,7 +415,9 @@ private ListOffsetResult fetchOffsetsByTimes(Map<TopicPartition, Long> timestamp
if (value.partitionsToRetry.isEmpty())
return result;

remainingToSearch.keySet().removeAll(result.fetchedOffsets.keySet());
remainingToSearch = timestampsToSearch.entrySet().stream()

@hachikuji hachikuji Sep 10, 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.

I think this can be replaced with remainingToSearch.keySet().retainAll(value.partitionsToRetry).

.filter(entry -> value.partitionsToRetry.contains(entry.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
} else if (!future.isRetriable()) {
throw future.exception();
}
Expand Down Expand Up @@ -632,9 +635,11 @@ private RequestFuture<ListOffsetResult> sendListOffsetsRequests(final Map<TopicP
final RequestFuture<ListOffsetResult> listOffsetRequestsFuture = new RequestFuture<>();
final Map<TopicPartition, OffsetData> fetchedTimestampOffsets = new HashMap<>();
final Set<TopicPartition> partitionsToRetry = new HashSet<>();
final Set<TopicPartition> partitionsRequireMetadataUpdate = new HashSet<>(timestampsToSearch.keySet());

@hachikuji hachikuji Sep 10, 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.

Another idea might be to pass partitionsToRetry into groupListOffsetRequests.

final AtomicInteger remainingResponses = new AtomicInteger(timestampsToSearchByNode.size());

for (Map.Entry<Node, Map<TopicPartition, ListOffsetRequest.PartitionData>> entry : timestampsToSearchByNode.entrySet()) {
partitionsRequireMetadataUpdate.removeAll(entry.getValue().keySet());
RequestFuture<ListOffsetResult> future =
sendListOffsetRequest(entry.getKey(), entry.getValue(), requireTimestamps);
future.addListener(new RequestFutureListener<ListOffsetResult>() {
Expand All @@ -645,6 +650,7 @@ 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);
}
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