DL1231 commented on code in PR #19983:
URL: https://github.com/apache/kafka/pull/19983#discussion_r2153469078
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java:
##########
@@ -113,6 +116,54 @@ public void partitionAssignmentChangeOnTopicSubscription()
{
assertEquals(0, state.numAssignedPartitions());
}
+ @Test
+ public void testIsFetchableOnManualAssignment() {
+ state.assignFromUser(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ @Test
+ public void testIsFetchableOnAutoAssignment() {
+ state.subscribe(Set.of(topic), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(Set.of(tp0, tp1));
+ assertAssignedPartitionIsFetchable();
+ }
+
+ private void assertAssignedPartitionIsFetchable() {
+ assertEquals(2, state.assignedPartitions().size());
+ assertTrue(state.assignedPartitions().contains(tp0));
+ assertTrue(state.assignedPartitions().contains(tp1));
+
+ assertFalse(state.isFetchable(tp0), "Should not be fetchable without a
valid position");
+ assertFalse(state.isFetchable(tp1), "Should not be fetchable without a
valid position");
+
+ state.seek(tp0, 1);
+ state.seek(tp1, 1);
+
+ assertTrue(state.isFetchable(tp0));
+ assertTrue(state.isFetchable(tp1));
+ }
+
+ @Test
+ public void testIsFetchableConsidersExplicitTopicSubscription() {
+ state.subscribe(Set.of(topic1), Optional.of(rebalanceListener));
+ state.assignFromSubscribed(singleton(t1p0));
+ state.seek(t1p0, 1);
+
+ assertEquals(Set.of(t1p0), state.assignedPartitions());
+ assertTrue(state.isFetchable(t1p0));
+
+ // Change subscription. Assigned partitions should remain unchanged
but not fetchable.
+ state.subscribe(singleton(topic), Optional.of(rebalanceListener));
Review Comment:
Thanks for the PR, just out of curiosity, is there any difference between
singleton() and Set.of() for this??
--
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
