showuon commented on a change in pull request #11493:
URL: https://github.com/apache/kafka/pull/11493#discussion_r810748411
##########
File path:
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -997,6 +1002,152 @@ public void testAssignWithStandbyReplicas() {
assertEquals(standbyPartitionsByHost, info20.standbyPartitionByHost());
}
+ @Test
+ public void testAssignWithStandbyReplicasBalanceSparse() {
+ builder.addSource(null, "source1", null, null, null, "topic1");
+ builder.addProcessor("processor", new MockApiProcessorSupplier<>(),
"source1");
+ builder.addStateStore(new MockKeyValueStoreBuilder("store1", false),
"processor");
+
+ final List<String> topics = asList("topic1");
+
+ createMockTaskManager(EMPTY_TASKS, EMPTY_TASKS);
+ adminClient =
createMockAdminClientForAssignor(getTopicPartitionOffsetsMap(
+ singletonList(APPLICATION_ID + "-store1-changelog"),
+ singletonList(3))
+ );
+
configurePartitionAssignorWith(Collections.singletonMap(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG,
1));
+
+ final List<String> client1Consumers = asList("consumer10",
"consumer11", "consumer12", "consumer13");
+ final List<String> client2Consumers = asList("consumer20",
"consumer21", "consumer22");
+
+ for (final String consumerId : client1Consumers) {
+ subscriptions.put(consumerId,
+ new Subscription(
+ topics,
+ getInfo(UUID_1, EMPTY_TASKS, EMPTY_TASKS,
USER_END_POINT).encode()));
+ }
+ for (final String consumerId : client2Consumers) {
+ subscriptions.put(consumerId,
+ new Subscription(
+ topics,
+ getInfo(UUID_2, EMPTY_TASKS, EMPTY_TASKS,
USER_END_POINT).encode()));
+ }
+
+ final Map<String, Assignment> assignments =
+ partitionAssignor.assign(metadata, new
GroupSubscription(subscriptions)).groupAssignment();
+
+ // Consumers
+ final AssignmentInfo info10 =
AssignmentInfo.decode(assignments.get("consumer10").userData());
+ final AssignmentInfo info11 =
AssignmentInfo.decode(assignments.get("consumer11").userData());
+ final AssignmentInfo info12 =
AssignmentInfo.decode(assignments.get("consumer12").userData());
+ final AssignmentInfo info13 =
AssignmentInfo.decode(assignments.get("consumer13").userData());
+ final AssignmentInfo info20 =
AssignmentInfo.decode(assignments.get("consumer20").userData());
+ final AssignmentInfo info21 =
AssignmentInfo.decode(assignments.get("consumer21").userData());
+ final AssignmentInfo info22 =
AssignmentInfo.decode(assignments.get("consumer22").userData());
+
+ // Check each consumer has no more than 1 task
+ assertTrue(info10.activeTasks().size() + info10.standbyTasks().size()
<= 1);
+ assertTrue(info11.activeTasks().size() + info11.standbyTasks().size()
<= 1);
+ assertTrue(info12.activeTasks().size() + info12.standbyTasks().size()
<= 1);
+ assertTrue(info13.activeTasks().size() + info13.standbyTasks().size()
<= 1);
+ assertTrue(info20.activeTasks().size() + info20.standbyTasks().size()
<= 1);
+ assertTrue(info21.activeTasks().size() + info21.standbyTasks().size()
<= 1);
+ assertTrue(info22.activeTasks().size() + info22.standbyTasks().size()
<= 1);
+ }
+
+ @Test
+ public void testAssignWithStandbyReplicasBalanceDense() {
+ builder.addSource(null, "source1", null, null, null, "topic1");
+ builder.addProcessor("processor", new MockApiProcessorSupplier<>(),
"source1");
+ builder.addStateStore(new MockKeyValueStoreBuilder("store1", false),
"processor");
+
+ final List<String> topics = asList("topic1");
+
+ createMockTaskManager(EMPTY_TASKS, EMPTY_TASKS);
+ adminClient =
createMockAdminClientForAssignor(getTopicPartitionOffsetsMap(
+ singletonList(APPLICATION_ID + "-store1-changelog"),
+ singletonList(3))
+ );
+
configurePartitionAssignorWith(Collections.singletonMap(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG,
1));
+
+ subscriptions.put("consumer10",
+ new Subscription(
+ topics,
+ getInfo(UUID_1, EMPTY_TASKS, EMPTY_TASKS,
USER_END_POINT).encode()));
+ subscriptions.put("consumer20",
+ new Subscription(
+ topics,
+ getInfo(UUID_2, EMPTY_TASKS, EMPTY_TASKS,
USER_END_POINT).encode()));
+
+ final Map<String, Assignment> assignments =
+ partitionAssignor.assign(metadata, new
GroupSubscription(subscriptions)).groupAssignment();
+
+ // Consumers
+ final AssignmentInfo info10 =
AssignmentInfo.decode(assignments.get("consumer10").userData());
+ final AssignmentInfo info20 =
AssignmentInfo.decode(assignments.get("consumer20").userData());
+
+ // Check each consumer has 3 tasks
+ assertEquals(3, info10.activeTasks().size() +
info10.standbyTasks().size());
+ assertEquals(3, info20.activeTasks().size() +
info20.standbyTasks().size());
+ // Check that not all the actives are on one node
+ assertTrue(info10.activeTasks().size() < 3);
+ assertTrue(info20.activeTasks().size() < 3);
+ }
+
+ @Test
+ public void testAssignWithStandbyReplicasBalanceWithStatelessTasks() {
+ builder.addSource(null, "source1", null, null, null, "topic1");
+ builder.addProcessor("processor_with_state", new
MockApiProcessorSupplier<>(), "source1");
+ builder.addProcessor("processor", new MockApiProcessorSupplier<>(),
"source1");
+ builder.addStateStore(new MockKeyValueStoreBuilder("store1", false),
"processor_with_state");
+
+ final List<String> topics = asList("topic1");
+
+ createMockTaskManager(EMPTY_TASKS, EMPTY_TASKS);
+ adminClient =
createMockAdminClientForAssignor(getTopicPartitionOffsetsMap(
+ singletonList(APPLICATION_ID + "-store1-changelog"),
+ singletonList(3))
+ );
+
configurePartitionAssignorWith(Collections.singletonMap(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG,
1));
+
+ subscriptions.put("consumer10",
+ new Subscription(
+ topics,
+ getInfo(UUID_1, EMPTY_TASKS, EMPTY_TASKS,
USER_END_POINT).encode()));
+ subscriptions.put("consumer11",
+ new Subscription(
+ topics,
+ getInfo(UUID_1, EMPTY_TASKS, EMPTY_TASKS,
USER_END_POINT).encode()));
+ subscriptions.put("consumer20",
+ new Subscription(
+ topics,
+ getInfo(UUID_2, EMPTY_TASKS, EMPTY_TASKS,
USER_END_POINT).encode()));
+ subscriptions.put("consumer21",
+ new Subscription(
+ topics,
+ getInfo(UUID_2, EMPTY_TASKS, EMPTY_TASKS,
USER_END_POINT).encode()));
+
+ final Map<String, Assignment> assignments =
+ partitionAssignor.assign(metadata, new
GroupSubscription(subscriptions)).groupAssignment();
Review comment:
I ran it and found there are no stateless tasks included in the
assignment. Could you check again?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
