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
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 5 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
78 changes: 78 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 @@ -62,6 +62,84 @@ 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,
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: util.Map[TopicPartition, OffsetSpec] = {
val expectedOffsetsUnassignedTopics = commitedOffsets.asScala.filter{ case (tp, _) => unassignedTopicPartitions.contains(tp) }.keySet.map(tp => tp -> OffsetSpec.latest).toMap
val expectedOffsetsAssignedTopics = endOffsets.filter{ case (tp, _) => assignedTopicPartitions.contains(tp) }.keySet.map(tp => tp -> OffsetSpec.latest).toMap
ArgumentMatchers.argThat[util.Map[TopicPartition, OffsetSpec]] { map =>
(map.keySet.asScala == expectedOffsetsUnassignedTopics.keySet || map.keySet.asScala == expectedOffsetsAssignedTopics.keySet) && map.values.asScala.forall(_.isInstanceOf[OffsetSpec.LatestSpec])
}
}

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(offsetsArgMatcher, any()))
.thenReturn(new ListOffsetsResult(endOffsets.asJava))

@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.

I was not fully satisfied with providing all the endOffsets in all the cases here so I took a bit of time to play with Mockito. I was able to make it work as follow:

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

    when(admin.listOffsets(
      ArgumentMatchers.argThat(offsetsArgMatcher(assignedTopicPartitions)),
      any()
    )).thenReturn(new ListOffsetsResult(endOffsets.view.filterKeys(assignedTopicPartitions.contains).toMap.asJava))

    when(admin.listOffsets(
      ArgumentMatchers.argThat(offsetsArgMatcher(unassignedTopicPartitions)),
      any()
    )).thenReturn(new ListOffsetsResult(endOffsets.view.filterKeys(unassignedTopicPartitions.contains).toMap.asJava))

Note the topicPartitionOffsets != null condition. It seems that Mockito call the argument matcher once with null somehow.

Interestingly, the test fails when I use this modified version. I think that this is due to the fact that not all the end offsets are returned all the time now and that committedOffsets must contain all the partitions. See my previous comment about this.

What do you think?

@IgnacioAcunaF IgnacioAcunaF Jun 25, 2021

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.

Run the test defining all the topic-partitoins with some value but failed too.

The stack:
ConsumerGroupServiceTest > testAdminRequestsForDescribeNegativeOffsets() FAILED java.lang.NullPointerException at kafka.admin.ConsumerGroupCommand$ConsumerGroupService.getLogEndOffsets(ConsumerGroupCommand.scala:638) at kafka.admin.ConsumerGroupCommand$ConsumerGroupService.$anonfun$collectGroupsOffsets$2(ConsumerGroupCommand.scala:411) at scala.collection.Iterator$$anon$9.next(Iterator.scala:575) at scala.collection.mutable.Growable.addAll(Growable.scala:62) at scala.collection.mutable.Growable.addAll$(Growable.scala:57) at scala.collection.mutable.HashMap.addAll(HashMap.scala:117) at scala.collection.mutable.HashMap$.from(HashMap.scala:589) at scala.collection.mutable.HashMap$.from(HashMap.scala:582) at scala.collection.MapOps$WithFilter.map(Map.scala:381) at kafka.admin.ConsumerGroupCommand$ConsumerGroupService.collectGroupsOffsets(ConsumerGroupCommand.scala:560) at kafka.admin.ConsumerGroupCommand$ConsumerGroupService.collectGroupOffsets(ConsumerGroupCommand.scala:551) at kafka.admin.ConsumerGroupServiceTest.testAdminRequestsForDescribeNegativeOffsets(ConsumerGroupServiceTest.scala:134)

@IgnacioAcunaF IgnacioAcunaF Jun 25, 2021

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.

Going to do some research, but seems like the problem I had before, the second call to when(..).then(..) doesnt get call, so the admin.listOffsets is call directly with no mock results.

Going to take a look.

@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.

You might need to adjust the expected arguments.

@IgnacioAcunaF IgnacioAcunaF Jun 25, 2021

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 yes, is because of the fact that not all the end offsets are returned all the time now, but only for unassigned partitions case.

Because the ConsumerGroupCommand defines unassigned topic partition as:
val unassignedPartitions = committedOffsets.filterNot { case (tp, _) => assignedTopicPartitions.contains(tp) }

The definition that I gave to the test is breaking that logic, because if the topic partition is not in the commitedOffsets, by definition shouldn't be on the unassignedPartitions.
The test, as as I primarily defined, is declaring testTopicPartition3 as an unassigned partition, but that partition is not being defined also on the commitedOffsets. So is not respecting the previous statment val unassignedPartitions = committedOffsets.filterNot { case (tp, _) => assignedTopicPartitions.contains(tp) }. That is the reason that the test is failing.

So the test case should be:

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

Just the testTopicPartition0 (which is assigned) as non defined, because as I seem there is never going to be a case where there is a non defined unassigned partition, because it is a requirement to be defined on commitedOffsets in order to be consider as unasigned.
On the other way, there could be assigned but not commited partitions.

@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.

Once defined that way, test runs OK.
Makes sense?

@IgnacioAcunaF IgnacioAcunaF Jun 25, 2021

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.

And why the test was passing before, was because of this:

 def offsetsArgMatcher: util.Map[TopicPartition, OffsetSpec] = {
      val expectedOffsetsUnassignedTopics = commitedOffsets.asScala.filter{ case (tp, _) => unassignedTopicPartitions.contains(tp) }.keySet.map(tp => tp -> OffsetSpec.latest).toMap
      val expectedOffsetsAssignedTopics = endOffsets.filter{ case (tp, _) => assignedTopicPartitions.contains(tp) }.keySet.map(tp => tp -> OffsetSpec.latest).toMap
      ArgumentMatchers.argThat[util.Map[TopicPartition, OffsetSpec]] { map =>
        (map.keySet.asScala == expectedOffsetsUnassignedTopics.keySet || map.keySet.asScala == expectedOffsetsAssignedTopics.keySet) && map.values.asScala.forall(_.isInstanceOf[OffsetSpec.LatestSpec])
      }
    }

The expectedOffsetsUnassigned was a subset of commitedOffsets (commitedOffsets.asScala.filter{ case (tp, _) => unassignedTopicPartitions.contains(tp) }.keySet.map(tp => tp -> OffsetSpec.latest).toMap)

The expectedOffsetsAssignedTopics was a subset of endOffsets (endOffsets.filter{ case (tp, _) => assignedTopicPartitions.contains(tp) }.keySet.map(tp => tp -> OffsetSpec.latest).toMap)

It was defined that way to preserve the logic of val unassignedPartitions = committedOffsets.filterNot { case (tp, _) => assignedTopicPartitions.contains(tp) }.

But your suggestion of use endOffsets on both as a filter seems cleaner to me. Just needed to adjust the test case.


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,
testTopicPartition4 -> Some(100),
testTopicPartition5 -> None
)
assertEquals(Some("Stable"), state)
assertTrue(assignments.nonEmpty)
Comment thread
IgnacioAcunaF marked this conversation as resolved.
Outdated
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(2)).listOffsets(offsetsArgMatcher, any())
}

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