Skip to content

KAFKA-12926: ConsumerGroupCommand's java.lang.NullPointerException at negative offsets while running kafka-consumer-groups.sh #10858

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
dajac merged 8 commits into from
Jun 29, 2021
+86 −7
Merged

KAFKA-12926: ConsumerGroupCommand's java.lang.NullPointerException at negative offsets while running kafka-consumer-groups.sh #10858

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
b7f857a
KAFKA-12926: Filter null values when getting offsets to avoid NullPoi…
IgnacioAcunaF Jun 15, 2021
154ba07
KAFKA-12926: Replace isolated changes on code with a unique function …
IgnacioAcunaF Jun 16, 2021
edaf354
KAFKA-12926: Pass filtered partitionsOffsets directly to collectConsu…
IgnacioAcunaF Jun 17, 2021
53a84cd
KAFKA-12926: Fix mockito's ambiguous reference on unit test
IgnacioAcunaF Jun 17, 2021
d0a3b8b
KAFKA-12926: Improvement of unitary test testAdminRequestsForDescribe…
IgnacioAcunaF Jun 23, 2021
0a22095
KAFKA-12926: Improvement of the unitary test
IgnacioAcunaF Jun 25, 2021
34bebc6
KAFKA-12926: Replace filterKeys with filter because compileTestScala …
IgnacioAcunaF Jun 25, 2021
4e3b237
KAFKA-12926: Improvement of unitary test. Remove comments and spacing…
IgnacioAcunaF Jun 28, 2021
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
11 changes: 4 additions & 7 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 @@ -560,27 +560,24 @@ object ConsumerGroupCommand extends Logging {
val groupOffsets = TreeMap[String, (Option[String], Option[Seq[PartitionAssignmentState]])]() ++ (for ((groupId, consumerGroup) <- consumerGroups) yield {
val state = consumerGroup.state
val committedOffsets = getCommittedOffsets(groupId)
// The admin client returns `null` as a value to indicate that there is not committed offset for a partition.
def getPartitionOffset(tp: TopicPartition): Option[Long] = committedOffsets.get(tp).filter(_ != null).map(_.offset)
var assignedTopicPartitions = ListBuffer[TopicPartition]()
val rowsWithConsumer = consumerGroup.members.asScala.filter(!_.assignment.topicPartitions.isEmpty).toSeq
.sortWith(_.assignment.topicPartitions.size > _.assignment.topicPartitions.size).flatMap { consumerSummary =>
val topicPartitions = consumerSummary.assignment.topicPartitions.asScala
assignedTopicPartitions = assignedTopicPartitions ++ topicPartitions
val partitionOffsets = consumerSummary.assignment.topicPartitions.asScala
.map { topicPartition =>
topicPartition -> committedOffsets.get(topicPartition).map(_.offset)
}.toMap
collectConsumerAssignment(groupId, Option(consumerGroup.coordinator), topicPartitions.toList,
partitionOffsets, Some(s"${consumerSummary.consumerId}"), Some(s"${consumerSummary.host}"),
getPartitionOffset, Some(s"${consumerSummary.consumerId}"), Some(s"${consumerSummary.host}"),
Some(s"${consumerSummary.clientId}"))
}

val unassignedPartitions = committedOffsets.filterNot { case (tp, _) => assignedTopicPartitions.contains(tp) }
val rowsWithoutConsumer = if (unassignedPartitions.nonEmpty) {
collectConsumerAssignment(
groupId,
Option(consumerGroup.coordinator),
unassignedPartitions.keySet.toSeq,
unassignedPartitions.map { case (tp, offset) => tp -> Some(offset.offset) },
getPartitionOffset,
Some(MISSING_COLUMN_VALUE),
Some(MISSING_COLUMN_VALUE),
Some(MISSING_COLUMN_VALUE)).toSeq
Expand Down
83 changes: 83 additions & 0 deletions core/src/test/scala/unit/kafka/admin/ConsumerGroupServiceTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ import org.junit.jupiter.api.Test
import org.mockito.ArgumentMatchers
import org.mockito.ArgumentMatchers._
import org.mockito.Mockito._
import org.mockito.ArgumentMatcher

import scala.jdk.CollectionConverters._

Expand Down Expand Up @@ -62,6 +63,88 @@ class ConsumerGroupServiceTest {
verify(admin, times(1)).listOffsets(offsetsArgMatcher, any())
}

@Test
def testAdminRequestsForDescribeNegativeOffsets(): Unit = {
val args = Array("--bootstrap-server", "localhost:9092", "--group", group, "--describe", "--offsets")
val groupService = consumerGroupService(args)

val testTopicPartition0 = new TopicPartition("testTopic1", 0);
val testTopicPartition1 = new TopicPartition("testTopic1", 1);
val testTopicPartition2 = new TopicPartition("testTopic1", 2);
val testTopicPartition3 = new TopicPartition("testTopic2", 0);
val testTopicPartition4 = new TopicPartition("testTopic2", 1);
val testTopicPartition5 = new TopicPartition("testTopic2", 2);

// Some topic's partitions gets valid OffsetAndMetada values, other gets nulls values (negative integers) and others aren't defined
val commitedOffsets = Map(
testTopicPartition1 -> new OffsetAndMetadata(100),
testTopicPartition2 -> null,
testTopicPartition3 -> new OffsetAndMetadata(100),
testTopicPartition4 -> new OffsetAndMetadata(100),
testTopicPartition5 -> null,
).asJava
Comment on lines +78 to +85

@dajac dajac Jun 25, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Actually, it seems that we should always have null or an OffsetAndMetadata here for each partition as the API always provided an answer for the requested partitions.

commitedOffsets -> committedOffsets

@IgnacioAcunaF IgnacioAcunaF Jun 25, 2021 •
edited
Loading

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.

Actually, put that case based on what I saw on that consumer-group:

This was the consumer-group state:
(groupId=order-validations, isSimpleConsumerGroup=false, members=(memberId=order-validations-d5fbca62-ab2b-48d7-96ba-0ae72dff72a6, groupInstanceId=null, clientId=order-validations, host=/127.0.0.1, assignment=(topicPartitions=rtl_orderReceive-0,rtl_orderReceive-1,rtl_orderReceive-2,rtl_orderReceive-3,rtl_orderReceive-4,rtl_orderReceive-5,rtl_orderReceive-6,rtl_orderReceive-7,rtl_orderReceive-8,rtl_orderReceive-9)), partitionAssignor=RoundRobinAssigner, state=Stable, coordinator=f0527.cluster.cl:31047 (id: 1 rack: null), authorizedOperations=[])

It has assigned all the partitions to rtl_orderReceive, but when getting the commited offsets:

Map(rtl_orderReceive-0 -> null, rtl_orderReceive-1 -> OffsetAndMetadata{offset=39, leaderEpoch=null, metadata=''}, rtl_orderReceive-2 -> null, rtl_orderReceive-3 -> OffsetAndMetadata{offset=33, leaderEpoch=null, metadata=''}, rtl_orderReceive-4 -> null, rtl_orderReceive-5 -> null, rtl_orderReceive-7 -> null, rtl_orderReceive-8 -> null)

Even that partition 6 was assigned, the aren't values commited for it (even not a -1).
That is for the case of assigned partitions, but for unassigned partitions, thinking it now, as it is the subsets of commited offsets that aren't assigned, it makes no sense to have non defined commited unassigned partition (because if there weren't a commited partition, then wouldn't exist the unassigned partition).

So I am thinking on:

val commitedOffsets = Map(
      testTopicPartition1 -> new OffsetAndMetadata(100),
      testTopicPartition2 -> null,
      testTopicPartition3 -> new OffsetAndMetadata(100),
      testTopicPartition4 -> new OffsetAndMetadata(100),
      testTopicPartition5 -> null,
    ).asJava

Just letting testTopicPartition0 as undefined, but defining testTopicPartition3 (unassigned).
Makes sense?

@dajac dajac Jun 25, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Thanks for the clarification. In this case, it might be better to keep committing in both cases in oder to be robust and to adjust the rest of the test accordingly.


val resultInfo = new ListOffsetsResult.ListOffsetsResultInfo(100, System.currentTimeMillis, Optional.of(1))
val endOffsets = Map(
testTopicPartition0 -> KafkaFuture.completedFuture(resultInfo),
testTopicPartition1 -> KafkaFuture.completedFuture(resultInfo),
testTopicPartition2 -> KafkaFuture.completedFuture(resultInfo),
testTopicPartition3 -> KafkaFuture.completedFuture(resultInfo),
testTopicPartition4 -> KafkaFuture.completedFuture(resultInfo),
testTopicPartition5 -> KafkaFuture.completedFuture(resultInfo),
)
val assignedTopicPartitions = Set(testTopicPartition0, testTopicPartition1, testTopicPartition2)
val unassignedTopicPartitions = Set(testTopicPartition3, testTopicPartition4, testTopicPartition5)

val consumerGroupDescription = new ConsumerGroupDescription(group,
true,
Collections.singleton(new MemberDescription("member1", Optional.of("instance1"), "client1", "host1", new MemberAssignment(assignedTopicPartitions.asJava))),
classOf[RangeAssignor].getName,
ConsumerGroupState.STABLE,
new Node(1, "localhost", 9092))

def offsetsArgMatcher(expectedPartitions: Set[TopicPartition]): ArgumentMatcher[util.Map[TopicPartition, OffsetSpec]] = {
topicPartitionOffsets => topicPartitionOffsets != null && topicPartitionOffsets.keySet.asScala.equals(expectedPartitions)
}

when(admin.describeConsumerGroups(ArgumentMatchers.eq(Collections.singletonList(group)), any()))
.thenReturn(new DescribeConsumerGroupsResult(Collections.singletonMap(group, KafkaFuture.completedFuture(consumerGroupDescription))))
when(admin.listConsumerGroupOffsets(ArgumentMatchers.eq(group), any()))
.thenReturn(AdminClientTestUtils.listConsumerGroupOffsetsResult(commitedOffsets))
when(admin.listOffsets(
ArgumentMatchers.argThat(offsetsArgMatcher(assignedTopicPartitions)),
any()
)).thenReturn(new ListOffsetsResult(endOffsets.filter{ case (tp, _) => assignedTopicPartitions.contains(tp) }.asJava))
Comment thread
IgnacioAcunaF marked this conversation as resolved.
Outdated
when(admin.listOffsets(
ArgumentMatchers.argThat(offsetsArgMatcher(unassignedTopicPartitions)),
any()
)).thenReturn(new ListOffsetsResult(endOffsets.filter{ case (tp, _) => unassignedTopicPartitions.contains(tp) }.asJava))
Comment thread
IgnacioAcunaF marked this conversation as resolved.
Outdated

val (state, assignments) = groupService.collectGroupOffsets(group)
val returnedOffsets = assignments.map { results =>
results.map { assignment =>
new TopicPartition(assignment.topic.get, assignment.partition.get) -> assignment.offset
}.toMap
}.getOrElse(Map.empty)
// Results should have information for all assigned topic partition (even if there is not Offset's information at all, because they get fills with None)
// Results should have information only for unassigned topic partitions if and only if there is information about them (including with null values)
Comment thread
IgnacioAcunaF marked this conversation as resolved.
Outdated
val expectedOffsets = Map(
testTopicPartition0 -> None,
testTopicPartition1 -> Some(100),
testTopicPartition2 -> None,
testTopicPartition3 -> Some(100),
testTopicPartition4 -> Some(100),
testTopicPartition5 -> None
)
assertEquals(Some("Stable"), state)
assertEquals(expectedOffsets, returnedOffsets)

verify(admin, times(1)).describeConsumerGroups(ArgumentMatchers.eq(Collections.singletonList(group)), any())
verify(admin, times(1)).listConsumerGroupOffsets(ArgumentMatchers.eq(group), any())
verify(admin, times(1)).listOffsets(ArgumentMatchers.argThat(offsetsArgMatcher(assignedTopicPartitions)), any())
verify(admin, times(1)).listOffsets(ArgumentMatchers.argThat(offsetsArgMatcher(unassignedTopicPartitions)), any())
}

@Test
def testAdminRequestsForResetOffsets(): Unit = {
val args = Seq("--bootstrap-server", "localhost:9092", "--group", group, "--reset-offsets", "--to-latest")
Expand Down