You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2020/02/08 00:48:19 UTC
[kafka] branch 2.5 updated: KAFKA-9507;
AdminClient should check for missing committed offsets (#8057)
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch 2.5
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.5 by this push:
new 64f8023 KAFKA-9507; AdminClient should check for missing committed offsets (#8057)
64f8023 is described below
commit 64f8023818d3af15c3c7324cf1ae31ead3a300ef
Author: David Mao <47...@users.noreply.github.com>
AuthorDate: Fri Feb 7 16:43:52 2020 -0800
KAFKA-9507; AdminClient should check for missing committed offsets (#8057)
Addresses exception being thrown by `AdminClient` when `listConsumerGroupOffsets` returns a negative offset. A negative offset indicates the absence of a committed offset for a requested partition, and should result in a null in the returned offset map.
Reviewers: Anna Povzner <an...@confluent.io>, Jason Gustafson <ja...@confluent.io>
---
.../main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java | 7 ++++++-
.../apache/kafka/clients/admin/ListConsumerGroupOffsetsResult.java | 1 +
.../java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java | 7 ++++++-
3 files changed, 13 insertions(+), 2 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 6b8e962..52332f8 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -3100,7 +3100,12 @@ public class KafkaAdminClient extends AdminClient {
final Long offset = partitionData.offset;
final String metadata = partitionData.metadata;
final Optional<Integer> leaderEpoch = partitionData.leaderEpoch;
- groupOffsetsListing.put(topicPartition, new OffsetAndMetadata(offset, leaderEpoch, metadata));
+ // Negative offset indicates that the group has no committed offset for this partition
+ if (offset < 0) {
+ groupOffsetsListing.put(topicPartition, null);
+ } else {
+ groupOffsetsListing.put(topicPartition, new OffsetAndMetadata(offset, leaderEpoch, metadata));
+ }
} else {
log.warn("Skipping return offset for {} due to error {}.", topicPartition, error);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsResult.java
index ea51934..48f4531 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsResult.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsResult.java
@@ -40,6 +40,7 @@ public class ListConsumerGroupOffsetsResult {
/**
* Return a future which yields a map of topic partitions to OffsetAndMetadata objects.
+ * If the group does not have a committed offset for this partition, the corresponding value in the returned map will be null.
*/
public KafkaFuture<Map<TopicPartition, OffsetAndMetadata>> partitionsToOffsetAndMetadata() {
return future;
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
index cba9b48..ed3e8a6 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
@@ -1494,6 +1494,7 @@ public class KafkaAdminClientTest {
TopicPartition myTopicPartition0 = new TopicPartition("my_topic", 0);
TopicPartition myTopicPartition1 = new TopicPartition("my_topic", 1);
TopicPartition myTopicPartition2 = new TopicPartition("my_topic", 2);
+ TopicPartition myTopicPartition3 = new TopicPartition("my_topic", 3);
final Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData = new HashMap<>();
responseData.put(myTopicPartition0, new OffsetFetchResponse.PartitionData(10,
@@ -1502,15 +1503,19 @@ public class KafkaAdminClientTest {
Optional.empty(), "", Errors.NONE));
responseData.put(myTopicPartition2, new OffsetFetchResponse.PartitionData(20,
Optional.empty(), "", Errors.NONE));
+ responseData.put(myTopicPartition3, new OffsetFetchResponse.PartitionData(OffsetFetchResponse.INVALID_OFFSET,
+ Optional.empty(), "", Errors.NONE));
env.kafkaClient().prepareResponse(new OffsetFetchResponse(Errors.NONE, responseData));
final ListConsumerGroupOffsetsResult result = env.adminClient().listConsumerGroupOffsets("group-0");
final Map<TopicPartition, OffsetAndMetadata> partitionToOffsetAndMetadata = result.partitionsToOffsetAndMetadata().get();
- assertEquals(3, partitionToOffsetAndMetadata.size());
+ assertEquals(4, partitionToOffsetAndMetadata.size());
assertEquals(10, partitionToOffsetAndMetadata.get(myTopicPartition0).offset());
assertEquals(0, partitionToOffsetAndMetadata.get(myTopicPartition1).offset());
assertEquals(20, partitionToOffsetAndMetadata.get(myTopicPartition2).offset());
+ assertTrue(partitionToOffsetAndMetadata.containsKey(myTopicPartition3));
+ assertNull(partitionToOffsetAndMetadata.get(myTopicPartition3));
}
}