lucasbru commented on code in PR #15271:
URL: https://github.com/apache/kafka/pull/15271#discussion_r1472843719
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -552,8 +555,63 @@ public void
testDelayedReconciliationResultAppliedWhenTargetChangedWithMetadataU
// Pending assignment that was discovered in metadata should be ready
to reconcile in the
// next reconciliation loop.
- Set<TopicIdPartition> topic2Assignment =
topicIdPartitionsSet(topicId2, topic2, 1, 2);
- assertEquals(topic2Assignment,
membershipManager.assignmentReadyToReconcile());
+ Map<Uuid, SortedSet<Integer>> topic2Assignment =
topicIdPartitionsMap(topicId2, 1, 2);
+ assertEquals(topic2Assignment,
membershipManager.topicPartitionsAwaitingReconciliation());
+ }
+
+ // Tests the case where topic metadata is not available at the time of the
assignment,
+ // but is made available later.
+ @Test
+ public void testDelayedMetadataUsedToCompleteAssignment() {
Review Comment:
If I understand your ask correctly, I think the test wouldn't pass yet. If
the metadata arrives while the first topic is being reconciled, we'll not
retrigger reconciliation, unless we get another assignment via the heartbeat. I
will fix this and implement the test you are suggesting in
[KAFKA-15832](https://issues.apache.org/jira/browse/KAFKA-15832)), where we'll
check for the need to reconcile regularly during poll.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -552,8 +555,63 @@ public void
testDelayedReconciliationResultAppliedWhenTargetChangedWithMetadataU
// Pending assignment that was discovered in metadata should be ready
to reconcile in the
// next reconciliation loop.
- Set<TopicIdPartition> topic2Assignment =
topicIdPartitionsSet(topicId2, topic2, 1, 2);
- assertEquals(topic2Assignment,
membershipManager.assignmentReadyToReconcile());
+ Map<Uuid, SortedSet<Integer>> topic2Assignment =
topicIdPartitionsMap(topicId2, 1, 2);
+ assertEquals(topic2Assignment,
membershipManager.topicPartitionsAwaitingReconciliation());
+ }
+
+ // Tests the case where topic metadata is not available at the time of the
assignment,
+ // but is made available later.
+ @Test
+ public void testDelayedMetadataUsedToCompleteAssignment() {
+
+ Uuid topicId1 = Uuid.randomUuid();
+ String topic1 = "topic1";
+ final TopicIdPartition topicId1Partition0 = new
TopicIdPartition(topicId1, new TopicPartition(topic1, 0));
+
+ Uuid topicId2 = Uuid.randomUuid();
+ String topic2 = "topic2";
+ final TopicIdPartition topicId2Partition0 = new
TopicIdPartition(topicId2, new TopicPartition(topic2, 0));
+
+ // Receive assignment with only topic1-0, entering STABLE state.
+ MembershipManagerImpl membershipManager =
+ mockMemberSuccessfullyReceivesAndAcksAssignment(topicId1, topic1,
Collections.singletonList(0));
+
+ membershipManager.onHeartbeatRequestSent();
+
+ assertEquals(MemberState.STABLE, membershipManager.state());
+
when(subscriptionState.assignedPartitions()).thenReturn(getTopicPartitions(Collections.singleton(topicId1Partition0)));
+ clearInvocations(membershipManager, subscriptionState);
+
+ // New assignment adding a new topic2-0 (not in metadata).
+ // No reconciliation triggered, because new topic in assignment is
waiting for metadata.
+
+ Map<Uuid, SortedSet<Integer>> newAssignment =
+ mkMap(
+ mkEntry(topicId1, mkSortedSet(0)),
+ mkEntry(topicId2, mkSortedSet(0))
+ );
+
+ receiveAssignment(newAssignment, membershipManager);
+
+ verifyReconciliationNotTriggered(membershipManager);
+ assertEquals(MemberState.RECONCILING, membershipManager.state());
+ assertEquals(Collections.singleton(topicId2),
membershipManager.topicsAwaitingReconciliation());
+ verify(metadata).requestUpdate(anyBoolean());
+ clearInvocations(membershipManager);
+ clearInvocations(commitRequestManager);
Review Comment:
Done
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -552,8 +555,63 @@ public void
testDelayedReconciliationResultAppliedWhenTargetChangedWithMetadataU
// Pending assignment that was discovered in metadata should be ready
to reconcile in the
// next reconciliation loop.
- Set<TopicIdPartition> topic2Assignment =
topicIdPartitionsSet(topicId2, topic2, 1, 2);
- assertEquals(topic2Assignment,
membershipManager.assignmentReadyToReconcile());
+ Map<Uuid, SortedSet<Integer>> topic2Assignment =
topicIdPartitionsMap(topicId2, 1, 2);
+ assertEquals(topic2Assignment,
membershipManager.topicPartitionsAwaitingReconciliation());
+ }
+
+ // Tests the case where topic metadata is not available at the time of the
assignment,
+ // but is made available later.
+ @Test
+ public void testDelayedMetadataUsedToCompleteAssignment() {
+
+ Uuid topicId1 = Uuid.randomUuid();
+ String topic1 = "topic1";
+ final TopicIdPartition topicId1Partition0 = new
TopicIdPartition(topicId1, new TopicPartition(topic1, 0));
+
+ Uuid topicId2 = Uuid.randomUuid();
+ String topic2 = "topic2";
+ final TopicIdPartition topicId2Partition0 = new
TopicIdPartition(topicId2, new TopicPartition(topic2, 0));
+
+ // Receive assignment with only topic1-0, entering STABLE state.
+ MembershipManagerImpl membershipManager =
+ mockMemberSuccessfullyReceivesAndAcksAssignment(topicId1, topic1,
Collections.singletonList(0));
+
+ membershipManager.onHeartbeatRequestSent();
+
+ assertEquals(MemberState.STABLE, membershipManager.state());
+
when(subscriptionState.assignedPartitions()).thenReturn(getTopicPartitions(Collections.singleton(topicId1Partition0)));
+ clearInvocations(membershipManager, subscriptionState);
+
+ // New assignment adding a new topic2-0 (not in metadata).
+ // No reconciliation triggered, because new topic in assignment is
waiting for metadata.
+
+ Map<Uuid, SortedSet<Integer>> newAssignment =
+ mkMap(
+ mkEntry(topicId1, mkSortedSet(0)),
+ mkEntry(topicId2, mkSortedSet(0))
+ );
+
+ receiveAssignment(newAssignment, membershipManager);
+
+ verifyReconciliationNotTriggered(membershipManager);
+ assertEquals(MemberState.RECONCILING, membershipManager.state());
+ assertEquals(Collections.singleton(topicId2),
membershipManager.topicsAwaitingReconciliation());
+ verify(metadata).requestUpdate(anyBoolean());
+ clearInvocations(membershipManager);
+ clearInvocations(commitRequestManager);
+
+ // Metadata discovered for topic2. Should trigger reconciliation to
complete the assignment,
+ // with membership manager entering ACKNOWLEDGING state.
+
+ Map<Uuid, String> fullTopicMetadata = mkMap(
+ mkEntry(topicId1, topic1),
+ mkEntry(topicId2, topic2)
+ );
+ when(metadata.topicNames()).thenReturn(fullTopicMetadata);
+
+ membershipManager.onUpdate(null);
+
+ verifyReconciliationTriggeredAndCompleted(membershipManager,
Arrays.asList(topicId1Partition0, topicId2Partition0));
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
