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

Jump to bottom

KAFKA-15022: [8/N] more tests for HAAssignor #14164

Merged

mjsax merged 5 commits into apache:trunk from lihaosky:rack-aware-more-tests

Aug 9, 2023

Collaborator

lihaosky commented Aug 7, 2023 •

edited

Description

More tests for HighAvailabilityTaskAssignor
Remove null and optional check for RackAwareTaskAssignor
Pass rack aware assignor configs to getMainConsumerConfigs so that they can be picked up in rebalance protocol
Change STATELESS_NON_OVERLAP_COST to 0. It was a mistake to be 1. Stateless tasks should be moved without this cost.

Tests

Update of existing tests


          KAFKA-15022: [8/N] more tests for HAAssignor

294528b

lihaosky force-pushed the rack-aware-more-tests branch from 3749564 to 294528b Compare

August 8, 2023 16:43


          change stateless non_over_lap cost to 0

add66aa

mjsax reviewed

View reviewed changes

...t/java/org/apache/kafka/streams/integration/HighAvailabilityTaskAssignorIntegrationTest.java Outdated

                               // Increasing the number of threads to ensure that a rebalance happens each time a consumer sends a rejoin (KAFKA-10455)
                               mkEntry(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 40),
                               mkEntry(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName()),
-                              mkEntry(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName())
+                              mkEntry(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName()),
+                              mkEntry(CommonClientConfigs.CLIENT_RACK_CONFIG, AssignmentTestUtils.RACK_0),

Member

mjsax Aug 8, 2023

Why is this hard-coded? Should we use different racks for different KS instances?

mjsax reviewed

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java Outdated

                       effectiveConfig.put(KafkaConfig.MessageMaxBytesProp(), 1000000);
                       effectiveConfig.put(KafkaConfig.ControlledShutdownEnableProp(), true);
                       effectiveConfig.put(KafkaConfig.ZkSessionTimeoutMsProp(), 10000);
+                      effectiveConfig.put(KafkaConfig.RackProp(), "rack0");

Member

mjsax Aug 8, 2023

why do we need this overwrite?

Collaborator Author

lihaosky Aug 8, 2023

Good catch. Might be some left over change. Let me delete it

mjsax reviewed

View reviewed changes

...src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java Outdated

+                      if (topicPartitionInfo != null) {
+                          lenient().when(mockInternalTopicManager.getTopicPartitionInfo(anySet())).thenAnswer(
+                              i -> {

Member

mjsax Aug 8, 2023

So i is the input parameter passed into getTopicPartitionInfo ? -- Can we find a better name?

mjsax reviewed

View reviewed changes

...src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java Outdated

+                                          return tp;
+                                      }
+                                  }
+                                  return null;

Member

mjsax Aug 8, 2023

Is null right here? Or should it be an empty Map ?

Collaborator Author

lihaosky Aug 8, 2023

We can return empty Map.

mjsax reviewed

View reviewed changes

...src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java Outdated

-                      configureDefault();
+                      createDefaultMockTaskManager();
+                      final List<Map<String, List<TopicPartitionInfo>>> changelogTopicPartitionInfo = getTopicPartitionInfo(3,

Member

mjsax Aug 8, 2023

nit: move first parameter 3, into it's own line below (hard to read the code otherwise)

mjsax reviewed

View reviewed changes

...src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java Outdated

@@ @@ -1213,12 +1376,18 @@ public void testAssignWithInternalTopics() { @@
                       final List<String> topics = asList("topic1", APPLICATION_ID + "-topicX");
                       final Set<TaskId> allTasks = mkSet(TASK_0_0, TASK_0_1, TASK_0_2);
-                      final MockInternalTopicManager internalTopicManager = configureDefault();
+                      createDefaultMockTaskManager();
+                      final List<Map<String, List<TopicPartitionInfo>>> changelogTopicPartitionInfo = getTopicPartitionInfo(4,

Member

mjsax Aug 8, 2023

nit: move first parameter to next line

mjsax reviewed

View reviewed changes

...src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java Outdated

                       final List<String> topics = asList("topic1", APPLICATION_ID + "-topicX", APPLICATION_ID + "-topicZ");
                       final Set<TaskId> allTasks = mkSet(TASK_0_0, TASK_0_1, TASK_0_2);
-                      final MockInternalTopicManager internalTopicManager = configureDefault();
+                      createDefaultMockTaskManager();
+                      final List<Map<String, List<TopicPartitionInfo>>> changelogTopicPartitionInfo = getTopicPartitionInfo(4,

Member

mjsax Aug 8, 2023

as above

mjsax reviewed

View reviewed changes

...src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java Outdated

		@@ -1768,16 +1993,22 @@ public void shouldReturnInterleavedAssignmentWithUnrevokedPartitionsRemovedWhenN
		final Set<TaskId> allTasks = mkSet(TASK_0_0, TASK_0_1, TASK_0_2);

		subscriptions.put(CONSUMER_1,

Member

mjsax Aug 8, 2023 •

edited

nit: move CONSUMER_1 to next line, too (same below)

mjsax reviewed

View reviewed changes

...src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java Outdated

+                          KeyValue.pair(NODE_3, asList(REPLICA_3))
+                      );
+                      final List<Map<String, List<TopicPartitionInfo>>> ret = new ArrayList<>();

Member

mjsax Aug 8, 2023

ret is not a good variable name

mjsax reviewed

View reviewed changes

...src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java Outdated

+                          ret.add(new HashMap<>());
+                          for (final String topic : topics) {
+                              final List<TopicPartitionInfo> topicPartitionInfoList = new ArrayList<>();
+                              ret.get(ret.size() - 1).put(topic, topicPartitionInfoList);

Member

mjsax Aug 8, 2023 •

edited

Should we add a local variable instead of getting the map from the list? -- Similar to topicPartitionInfoList?

mjsax reviewed

View reviewed changes

...ava/org/apache/kafka/streams/processor/internals/assignment/TaskAssignorConvergenceTest.java Outdated

+                                  changelogPartitionsForTask.put(taskId, mkSet(new TopicPartition(changelogTopicName, i)));
+                                  tasksForTopicGroup.computeIfAbsent(new Subtopology(subtopology, null), k -> new HashSet<>()).add(taskId);
+                                  final Random random = new Random();

Member

mjsax Aug 8, 2023 •

edited

We should init Random with a value that we log (to allow us to reproduce the same test)

final long seed = System.currentTimeMillis():
log.info(seed); // or just use println
final Random random = new Random(seed);

mjsax reviewed

View reviewed changes

...ava/org/apache/kafka/streams/processor/internals/assignment/TaskAssignorConvergenceTest.java Outdated

                               final UUID uuid = uuidForInt(i);
                               clientStates.put(uuid, emptyInstance(uuid, statefulTaskEndOffsetSums));
+                              final Random random = new Random();

Member

mjsax Aug 8, 2023 •

edited

as above: maybe setup Random inside @BeforeTest instead and share?

mjsax reviewed

View reviewed changes

...ava/org/apache/kafka/streams/processor/internals/assignment/TaskAssignorConvergenceTest.java Outdated

-                                                                              EMPTY_RACK_AWARE_ASSIGNMENT_TAGS);
+                                                                              EMPTY_RACK_AWARE_ASSIGNMENT_TAGS,
+                                                  null,
+                                              null,

Member

mjsax Aug 8, 2023

fix indention

mjsax reviewed

View reviewed changes

...ava/org/apache/kafka/streams/processor/internals/assignment/TaskAssignorConvergenceTest.java Outdated

-                                                                              EMPTY_RACK_AWARE_ASSIGNMENT_TAGS);
+                                                                              EMPTY_RACK_AWARE_ASSIGNMENT_TAGS,
+                                                  null,
+                                              null,

Member

mjsax Aug 8, 2023

fix indention

lihaosky added 3 commits

August 8, 2023 16:30


          comments

ea51477


          fix indentation

99fd3d9


          relax check

6f5fbac

lihaosky mentioned this pull request

KAFKA-15022: [9/N] use RackAwareTaskAssignor in StickyTaskAssignor #14178

Merged

mjsax merged commit 8730816 into apache:trunk

1 check was pending

lihaosky deleted the rack-aware-more-tests branch

August 10, 2023 04:12

jeqo pushed a commit to aiven/kafka that referenced this pull request


          KAFKA-15022: [8/N] more tests for HAAssignor (apache#14164)

6ba9332

Part of KIP-925.

- Add more tests for HighAvailabilityTaskAssignor
- Remove null and optional check for RackAwareTaskAssignor
- Pass rack aware assignor configs to getMainConsumerConfigs so that they can be picked up in rebalance protocol
- Change STATELESS_NON_OVERLAP_COST to 0. It was a mistake to be 1. Stateless tasks should be moved without this cost.
- Update of existing tests

Reviewers: Matthias J. Sax <matthias@confluent.io>

jeqo pushed a commit to jeqo/kafka that referenced this pull request


          KAFKA-15022: [8/N] more tests for HAAssignor (apache#14164)

76b8a4e

Part of KIP-925.

- Add more tests for HighAvailabilityTaskAssignor
- Remove null and optional check for RackAwareTaskAssignor
- Pass rack aware assignor configs to getMainConsumerConfigs so that they can be picked up in rebalance protocol
- Change STATELESS_NON_OVERLAP_COST to 0. It was a mistake to be 1. Stateless tasks should be moved without this cost.
- Update of existing tests

Reviewers: Matthias J. Sax <matthias@confluent.io>

jeqo pushed a commit to jeqo/kafka that referenced this pull request


          KAFKA-15022: [8/N] more tests for HAAssignor (apache#14164)

2da9a49

Part of KIP-925.

- Add more tests for HighAvailabilityTaskAssignor
- Remove null and optional check for RackAwareTaskAssignor
- Pass rack aware assignor configs to getMainConsumerConfigs so that they can be picked up in rebalance protocol
- Change STATELESS_NON_OVERLAP_COST to 0. It was a mistake to be 1. Stateless tasks should be moved without this cost.
- Update of existing tests

Reviewers: Matthias J. Sax <matthias@confluent.io>

rreddy-22 pushed a commit to rreddy-22/kafka-rreddy that referenced this pull request


          KAFKA-15022: [8/N] more tests for HAAssignor (apache#14164)

23e9686

Part of KIP-925.

- Add more tests for HighAvailabilityTaskAssignor
- Remove null and optional check for RackAwareTaskAssignor
- Pass rack aware assignor configs to getMainConsumerConfigs so that they can be picked up in rebalance protocol
- Change STATELESS_NON_OVERLAP_COST to 0. It was a mistake to be 1. Stateless tasks should be moved without this cost.
- Update of existing tests

Reviewers: Matthias J. Sax <matthias@confluent.io>

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment