lihaosky commented on code in PR #13851:
URL: https://github.com/apache/kafka/pull/13851#discussion_r1244450143
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -409,6 +411,43 @@ private String getBrokerSideConfigValue(final Config
brokerSideTopicConfig,
return brokerSideConfigEntry.value();
}
+ public Map<String, List<TopicPartitionInfo>> getTopicPartitionInfo(final
Set<String> topics) {
+ log.debug("Starting to describe topics {} in partition assignor.",
topics);
+
+ long currentWallClockMs = time.milliseconds();
+ final long deadlineMs = currentWallClockMs + retryTimeoutMs;
+
+ Set<String> topicsToDescribe = new HashSet<>(topics);
+ final Map<String, List<TopicPartitionInfo>> topicPartitionInfo = new
HashMap<>();
+
+ while (!topicsToDescribe.isEmpty()) {
+ final Map<String, List<TopicPartitionInfo>> existed =
getTopicPartitionInfo(topicsToDescribe, null);
+ topicPartitionInfo.putAll(existed);
+ topicsToDescribe.removeAll(topicPartitionInfo.keySet());
+ if (!topicsToDescribe.isEmpty()) {
+ currentWallClockMs = time.milliseconds();
+
+ if (currentWallClockMs >= deadlineMs) {
+ final String timeoutError = String.format(
+ "Could not create topics within %d milliseconds. " +
+ "This can happen if the Kafka cluster is
temporarily not available.",
+ retryTimeoutMs);
+ log.error(timeoutError);
Review Comment:
Is this mimicking the `makeReady` function logs.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -409,6 +411,43 @@ private String getBrokerSideConfigValue(final Config
brokerSideTopicConfig,
return brokerSideConfigEntry.value();
}
+ public Map<String, List<TopicPartitionInfo>> getTopicPartitionInfo(final
Set<String> topics) {
+ log.debug("Starting to describe topics {} in partition assignor.",
topics);
+
+ long currentWallClockMs = time.milliseconds();
+ final long deadlineMs = currentWallClockMs + retryTimeoutMs;
+
+ Set<String> topicsToDescribe = new HashSet<>(topics);
+ final Map<String, List<TopicPartitionInfo>> topicPartitionInfo = new
HashMap<>();
+
+ while (!topicsToDescribe.isEmpty()) {
+ final Map<String, List<TopicPartitionInfo>> existed =
getTopicPartitionInfo(topicsToDescribe, null);
+ topicPartitionInfo.putAll(existed);
+ topicsToDescribe.removeAll(topicPartitionInfo.keySet());
+ if (!topicsToDescribe.isEmpty()) {
+ currentWallClockMs = time.milliseconds();
+
+ if (currentWallClockMs >= deadlineMs) {
+ final String timeoutError = String.format(
+ "Could not create topics within %d milliseconds. " +
+ "This can happen if the Kafka cluster is
temporarily not available.",
+ retryTimeoutMs);
+ log.error(timeoutError);
+ throw new TimeoutException(timeoutError);
+ }
+ log.info(
Review Comment:
Is this mimicking the `makeReady` function logs.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RackAwareTaskAssignor.java:
##########
@@ -0,0 +1,164 @@
+package org.apache.kafka.streams.processor.internals;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackAwareTaskAssignor {
+ private static final Logger log =
LoggerFactory.getLogger(RackAwareTaskAssignor.class);
+
+ private final Cluster fullMetadata;
+ private final Map<TaskId, Set<TopicPartition>> partitionsForTask;
+ private final Map<UUID, Map<String, Optional<String>>> processRacks;
+ private final AssignmentConfigs assignmentConfigs;
+ private final Map<TopicPartition, Set<String>> racksForPartition;
+ private final InternalTopicManager internalTopicManager;
+ private Boolean canEnableForActive;
+
+ public RackAwareTaskAssignor(final Cluster fullMetadata,
+ final Map<TaskId, Set<TopicPartition>>
partitionsForTask,
+ final Map<Subtopology, Set<TaskId>>
tasksForTopicGroup,
+ final Map<UUID, Map<String,
Optional<String>>> processRacks,
+ final InternalTopicManager
internalTopicManager,
+ final AssignmentConfigs assignmentConfigs) {
+ this.fullMetadata = fullMetadata;
+ this.partitionsForTask = partitionsForTask;
+ this.processRacks = processRacks;
+ this.internalTopicManager = internalTopicManager;
+ this.assignmentConfigs = assignmentConfigs;
+ this.racksForPartition = new HashMap<>();
+ }
+
+ public synchronized boolean canEnableForActive() {
+ if (canEnableForActive != null) {
+ return canEnableForActive;
+ }
+
+ /*
+ TODO: enable this after we add the config
+ if
(StreamsConfig.RACK_AWARE_ASSSIGNMENT_STRATEGY_NONE.equals(assignmentConfigs.rackAwareAssignmentStrategy))
{
+ canEnableForActive = false;
+ return false;
+ }
+ */
+
+ if (!validateClientRack()) {
+ canEnableForActive = false;
+ return false;
+ }
+
+ canEnableForActive = validateTopicPartitionRack();
+ return canEnableForActive;
+ }
+
+ public boolean canEnableForStandby() {
+ // TODO
+ return false;
+ }
+
+ private boolean validateTopicPartitionRack() {
+ // Make sure rackId exist for all TopicPartitions needed
Review Comment:
The reason I want to enforce all topic partition racks exist to enable the
assignor is to avoid too much assignment changes if some racks appear or
disappear later somehow.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RackAwareTaskAssignor.java:
##########
@@ -0,0 +1,164 @@
+package org.apache.kafka.streams.processor.internals;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackAwareTaskAssignor {
+ private static final Logger log =
LoggerFactory.getLogger(RackAwareTaskAssignor.class);
+
+ private final Cluster fullMetadata;
+ private final Map<TaskId, Set<TopicPartition>> partitionsForTask;
+ private final Map<UUID, Map<String, Optional<String>>> processRacks;
+ private final AssignmentConfigs assignmentConfigs;
+ private final Map<TopicPartition, Set<String>> racksForPartition;
+ private final InternalTopicManager internalTopicManager;
+ private Boolean canEnableForActive;
+
+ public RackAwareTaskAssignor(final Cluster fullMetadata,
+ final Map<TaskId, Set<TopicPartition>>
partitionsForTask,
+ final Map<Subtopology, Set<TaskId>>
tasksForTopicGroup,
+ final Map<UUID, Map<String,
Optional<String>>> processRacks,
+ final InternalTopicManager
internalTopicManager,
+ final AssignmentConfigs assignmentConfigs) {
+ this.fullMetadata = fullMetadata;
+ this.partitionsForTask = partitionsForTask;
+ this.processRacks = processRacks;
+ this.internalTopicManager = internalTopicManager;
+ this.assignmentConfigs = assignmentConfigs;
+ this.racksForPartition = new HashMap<>();
+ }
+
+ public synchronized boolean canEnableForActive() {
+ if (canEnableForActive != null) {
+ return canEnableForActive;
+ }
+
+ /*
+ TODO: enable this after we add the config
+ if
(StreamsConfig.RACK_AWARE_ASSSIGNMENT_STRATEGY_NONE.equals(assignmentConfigs.rackAwareAssignmentStrategy))
{
+ canEnableForActive = false;
+ return false;
+ }
+ */
+
+ if (!validateClientRack()) {
+ canEnableForActive = false;
+ return false;
+ }
+
+ canEnableForActive = validateTopicPartitionRack();
+ return canEnableForActive;
+ }
+
+ public boolean canEnableForStandby() {
+ // TODO
+ return false;
+ }
+
+ private boolean validateTopicPartitionRack() {
+ // Make sure rackId exist for all TopicPartitions needed
+ final Set<String> topicsToDescribe = new HashSet<>();
+ for (final Set<TopicPartition> topicPartitions :
partitionsForTask.values()) {
+ for (TopicPartition topicPartition : topicPartitions) {
+ final PartitionInfo partitionInfo =
fullMetadata.partition(topicPartition);
+ if (partitionInfo == null) {
+ log.error("TopicPartition {} doesn't exist in cluster",
topicPartition);
+ return false;
+ }
+ final Node[] replica = partitionInfo.replicas();
+ if (replica == null || replica.length == 0) {
+ topicsToDescribe.add(topicPartition.topic());
+ continue;
+ }
+ for (final Node node : replica) {
+ if (node.hasRack()) {
+ racksForPartition.computeIfAbsent(topicPartition, k ->
new HashSet<>()).add(node.rack());
+ }
+ }
+ }
+ }
+
+ if (!topicsToDescribe.isEmpty()) {
+ log.info("Fetching PartitionInfo for topics {}", topicsToDescribe);
Review Comment:
Sure
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/RackAwareTaskAssignorTest.java:
##########
@@ -0,0 +1,264 @@
+package org.apache.kafka.streams.processor.internals.assignment;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import org.apache.kafka.clients.admin.MockAdminClient;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.StreamsConfig.InternalConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.internals.RackAwareTaskAssignor;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.apache.kafka.test.MockClientSupplier;
+import org.apache.kafka.test.MockInternalTopicManager;
+import org.junit.Test;
+
+public class RackAwareTaskAssignorTest {
+ private final static String TOPIC0 = "topic0";
+ private final static String TOPIC1 = "topic1";
+ private static final String USER_END_POINT = "localhost:8080";
+ private static final String APPLICATION_ID =
"stream-partition-assignor-test";
+
+ private final Node node0 = new Node(0, "node0", 1, "rack1");
+ private final Node node1 = new Node(1, "node1", 1, "rack2");
+ private final Node node2 = new Node(2, "node2", 1, "rack3");
+ private final Node[] replicas = new Node[] {node0, node1, node2};
+
+ private final PartitionInfo partitionInfo00 = new PartitionInfo(TOPIC0, 0,
node0, replicas, replicas);
+ private final PartitionInfo partitionInfo01 = new PartitionInfo(TOPIC0, 1,
node0, replicas, replicas);
+
+ private final TopicPartition tp00 = new TopicPartition(TOPIC0, 0);
+ private final TopicPartition tp01 = new TopicPartition(TOPIC0, 1);
+ private final TopicPartition tp10 = new TopicPartition(TOPIC1, 0);
+
+ private final UUID process0UUID = UUID.randomUUID();
+ private final UUID process1UUID = UUID.randomUUID();
+
+ private final Subtopology subtopology1 = new Subtopology(1, "topology1");
+
+ private final MockTime time = new MockTime();
+ private final StreamsConfig streamsConfig = new
StreamsConfig(configProps());
+ private final MockClientSupplier mockClientSupplier = new
MockClientSupplier();
+
+ private final MockInternalTopicManager mockInternalTopicManager = new
MockInternalTopicManager(
+ time,
+ streamsConfig,
+ mockClientSupplier.restoreConsumer,
+ false
+ );
+
+ private Map<String, Object> configProps() {
+ final Map<String, Object> configurationMap = new HashMap<>();
+ configurationMap.put(StreamsConfig.APPLICATION_ID_CONFIG,
APPLICATION_ID);
+ configurationMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
USER_END_POINT);
+
+ final ReferenceContainer referenceContainer = new ReferenceContainer();
+ /*
+ referenceContainer.mainConsumer = consumer;
+ referenceContainer.adminClient = adminClient;
+ referenceContainer.taskManager = taskManager;
+ referenceContainer.streamsMetadataState = streamsMetadataState;
+ referenceContainer.time = time;
+ */
+
configurationMap.put(InternalConfig.REFERENCE_CONTAINER_PARTITION_ASSIGNOR,
referenceContainer);
+ return configurationMap;
+ }
+
+ @Test
+ public void disableActiveSinceMissingClusterInfo() {
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2)),
+ new HashSet<>(Arrays.asList(partitionInfo00, partitionInfo01)),
+ Collections.emptySet(),
+ Collections.emptySet()
+ );
+
+ final Map<UUID, Map<String, Optional<String>>> processRacks = new
HashMap<>();
+
+ processRacks.put(process0UUID , Collections.singletonMap("consumer1",
Optional.of("rack1")));
+
+ final RackAwareTaskAssignor assignor = new RackAwareTaskAssignor(
+ metadata,
+ Collections.singletonMap(new TaskId(1, 1), new
HashSet<>(Arrays.asList(tp00, tp10))),
+ Collections.singletonMap(subtopology1, Collections.singleton(new
TaskId(1, 1))),
+ processRacks,
+ mockInternalTopicManager,
+ new
AssignorConfiguration(streamsConfig.originals()).assignmentConfigs()
+ );
+
+ // False since tp10 is missing in cluster metadata
+ assertFalse(assignor.canEnableForActive());
+ }
+
+ @Test
+ public void disableActiveSinceRackMissingInClient() {
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2)),
+ new HashSet<>(Arrays.asList(partitionInfo00, partitionInfo01)),
+ Collections.emptySet(),
+ Collections.emptySet()
+ );
+
+ final Map<UUID, Map<String, Optional<String>>> processRacks = new
HashMap<>();
+
+ // Missing rackId config in client
+ processRacks.put(process0UUID , Collections.singletonMap("consumer1",
Optional.empty()));
+
+ final RackAwareTaskAssignor assignor = new RackAwareTaskAssignor(
+ metadata,
+ Collections.singletonMap(new TaskId(1, 1),
Collections.singleton(tp00)),
+ Collections.singletonMap(subtopology1, Collections.singleton(new
TaskId(1, 1))),
+ processRacks,
+ mockInternalTopicManager,
+ new
AssignorConfiguration(streamsConfig.originals()).assignmentConfigs()
+ );
+
+ // False since process1 doesn't have rackId
+ assertFalse(assignor.canEnableForActive());
+ }
+
+ @Test
+ public void disableActiveSinceRackDiffersInSameProcess() {
Review Comment:
I'm ok deleting this. Was trying to get test coverage.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/RackAwareTaskAssignorTest.java:
##########
@@ -0,0 +1,264 @@
+package org.apache.kafka.streams.processor.internals.assignment;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import org.apache.kafka.clients.admin.MockAdminClient;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.StreamsConfig.InternalConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.internals.RackAwareTaskAssignor;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.apache.kafka.test.MockClientSupplier;
+import org.apache.kafka.test.MockInternalTopicManager;
+import org.junit.Test;
+
+public class RackAwareTaskAssignorTest {
+ private final static String TOPIC0 = "topic0";
+ private final static String TOPIC1 = "topic1";
+ private static final String USER_END_POINT = "localhost:8080";
+ private static final String APPLICATION_ID =
"stream-partition-assignor-test";
+
+ private final Node node0 = new Node(0, "node0", 1, "rack1");
+ private final Node node1 = new Node(1, "node1", 1, "rack2");
+ private final Node node2 = new Node(2, "node2", 1, "rack3");
+ private final Node[] replicas = new Node[] {node0, node1, node2};
+
+ private final PartitionInfo partitionInfo00 = new PartitionInfo(TOPIC0, 0,
node0, replicas, replicas);
+ private final PartitionInfo partitionInfo01 = new PartitionInfo(TOPIC0, 1,
node0, replicas, replicas);
+
+ private final TopicPartition tp00 = new TopicPartition(TOPIC0, 0);
+ private final TopicPartition tp01 = new TopicPartition(TOPIC0, 1);
+ private final TopicPartition tp10 = new TopicPartition(TOPIC1, 0);
+
+ private final UUID process0UUID = UUID.randomUUID();
+ private final UUID process1UUID = UUID.randomUUID();
+
+ private final Subtopology subtopology1 = new Subtopology(1, "topology1");
+
+ private final MockTime time = new MockTime();
+ private final StreamsConfig streamsConfig = new
StreamsConfig(configProps());
+ private final MockClientSupplier mockClientSupplier = new
MockClientSupplier();
+
+ private final MockInternalTopicManager mockInternalTopicManager = new
MockInternalTopicManager(
+ time,
+ streamsConfig,
+ mockClientSupplier.restoreConsumer,
+ false
+ );
+
+ private Map<String, Object> configProps() {
+ final Map<String, Object> configurationMap = new HashMap<>();
+ configurationMap.put(StreamsConfig.APPLICATION_ID_CONFIG,
APPLICATION_ID);
+ configurationMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
USER_END_POINT);
+
+ final ReferenceContainer referenceContainer = new ReferenceContainer();
+ /*
+ referenceContainer.mainConsumer = consumer;
+ referenceContainer.adminClient = adminClient;
+ referenceContainer.taskManager = taskManager;
+ referenceContainer.streamsMetadataState = streamsMetadataState;
+ referenceContainer.time = time;
+ */
+
configurationMap.put(InternalConfig.REFERENCE_CONTAINER_PARTITION_ASSIGNOR,
referenceContainer);
+ return configurationMap;
+ }
+
+ @Test
+ public void disableActiveSinceMissingClusterInfo() {
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2)),
+ new HashSet<>(Arrays.asList(partitionInfo00, partitionInfo01)),
+ Collections.emptySet(),
+ Collections.emptySet()
+ );
+
+ final Map<UUID, Map<String, Optional<String>>> processRacks = new
HashMap<>();
+
+ processRacks.put(process0UUID , Collections.singletonMap("consumer1",
Optional.of("rack1")));
+
+ final RackAwareTaskAssignor assignor = new RackAwareTaskAssignor(
+ metadata,
+ Collections.singletonMap(new TaskId(1, 1), new
HashSet<>(Arrays.asList(tp00, tp10))),
+ Collections.singletonMap(subtopology1, Collections.singleton(new
TaskId(1, 1))),
+ processRacks,
+ mockInternalTopicManager,
+ new
AssignorConfiguration(streamsConfig.originals()).assignmentConfigs()
+ );
+
+ // False since tp10 is missing in cluster metadata
+ assertFalse(assignor.canEnableForActive());
+ }
+
+ @Test
+ public void disableActiveSinceRackMissingInClient() {
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2)),
+ new HashSet<>(Arrays.asList(partitionInfo00, partitionInfo01)),
+ Collections.emptySet(),
+ Collections.emptySet()
+ );
+
+ final Map<UUID, Map<String, Optional<String>>> processRacks = new
HashMap<>();
+
+ // Missing rackId config in client
+ processRacks.put(process0UUID , Collections.singletonMap("consumer1",
Optional.empty()));
+
+ final RackAwareTaskAssignor assignor = new RackAwareTaskAssignor(
+ metadata,
+ Collections.singletonMap(new TaskId(1, 1),
Collections.singleton(tp00)),
Review Comment:
This is just to simplify to one task
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RackAwareTaskAssignor.java:
##########
@@ -0,0 +1,164 @@
+package org.apache.kafka.streams.processor.internals;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackAwareTaskAssignor {
+ private static final Logger log =
LoggerFactory.getLogger(RackAwareTaskAssignor.class);
+
+ private final Cluster fullMetadata;
+ private final Map<TaskId, Set<TopicPartition>> partitionsForTask;
+ private final Map<UUID, Map<String, Optional<String>>> processRacks;
+ private final AssignmentConfigs assignmentConfigs;
+ private final Map<TopicPartition, Set<String>> racksForPartition;
+ private final InternalTopicManager internalTopicManager;
+ private Boolean canEnableForActive;
Review Comment:
What you said make sense. I was also on the fence about this.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RackAwareTaskAssignor.java:
##########
@@ -0,0 +1,164 @@
+package org.apache.kafka.streams.processor.internals;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackAwareTaskAssignor {
+ private static final Logger log =
LoggerFactory.getLogger(RackAwareTaskAssignor.class);
+
+ private final Cluster fullMetadata;
+ private final Map<TaskId, Set<TopicPartition>> partitionsForTask;
+ private final Map<UUID, Map<String, Optional<String>>> processRacks;
+ private final AssignmentConfigs assignmentConfigs;
+ private final Map<TopicPartition, Set<String>> racksForPartition;
+ private final InternalTopicManager internalTopicManager;
+ private Boolean canEnableForActive;
+
+ public RackAwareTaskAssignor(final Cluster fullMetadata,
+ final Map<TaskId, Set<TopicPartition>>
partitionsForTask,
+ final Map<Subtopology, Set<TaskId>>
tasksForTopicGroup,
+ final Map<UUID, Map<String,
Optional<String>>> processRacks,
+ final InternalTopicManager
internalTopicManager,
+ final AssignmentConfigs assignmentConfigs) {
+ this.fullMetadata = fullMetadata;
+ this.partitionsForTask = partitionsForTask;
+ this.processRacks = processRacks;
+ this.internalTopicManager = internalTopicManager;
+ this.assignmentConfigs = assignmentConfigs;
+ this.racksForPartition = new HashMap<>();
+ }
+
+ public synchronized boolean canEnableForActive() {
+ if (canEnableForActive != null) {
+ return canEnableForActive;
+ }
+
+ /*
+ TODO: enable this after we add the config
+ if
(StreamsConfig.RACK_AWARE_ASSSIGNMENT_STRATEGY_NONE.equals(assignmentConfigs.rackAwareAssignmentStrategy))
{
+ canEnableForActive = false;
+ return false;
+ }
+ */
+
+ if (!validateClientRack()) {
+ canEnableForActive = false;
+ return false;
+ }
+
+ canEnableForActive = validateTopicPartitionRack();
+ return canEnableForActive;
+ }
+
+ public boolean canEnableForStandby() {
+ // TODO
+ return false;
+ }
+
+ private boolean validateTopicPartitionRack() {
+ // Make sure rackId exist for all TopicPartitions needed
+ final Set<String> topicsToDescribe = new HashSet<>();
+ for (final Set<TopicPartition> topicPartitions :
partitionsForTask.values()) {
Review Comment:
Yes
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RackAwareTaskAssignor.java:
##########
@@ -0,0 +1,164 @@
+package org.apache.kafka.streams.processor.internals;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackAwareTaskAssignor {
+ private static final Logger log =
LoggerFactory.getLogger(RackAwareTaskAssignor.class);
+
+ private final Cluster fullMetadata;
+ private final Map<TaskId, Set<TopicPartition>> partitionsForTask;
+ private final Map<UUID, Map<String, Optional<String>>> processRacks;
+ private final AssignmentConfigs assignmentConfigs;
+ private final Map<TopicPartition, Set<String>> racksForPartition;
+ private final InternalTopicManager internalTopicManager;
+ private Boolean canEnableForActive;
+
+ public RackAwareTaskAssignor(final Cluster fullMetadata,
+ final Map<TaskId, Set<TopicPartition>>
partitionsForTask,
+ final Map<Subtopology, Set<TaskId>>
tasksForTopicGroup,
+ final Map<UUID, Map<String,
Optional<String>>> processRacks,
+ final InternalTopicManager
internalTopicManager,
+ final AssignmentConfigs assignmentConfigs) {
+ this.fullMetadata = fullMetadata;
+ this.partitionsForTask = partitionsForTask;
+ this.processRacks = processRacks;
+ this.internalTopicManager = internalTopicManager;
+ this.assignmentConfigs = assignmentConfigs;
+ this.racksForPartition = new HashMap<>();
+ }
+
+ public synchronized boolean canEnableForActive() {
+ if (canEnableForActive != null) {
+ return canEnableForActive;
+ }
+
+ /*
+ TODO: enable this after we add the config
+ if
(StreamsConfig.RACK_AWARE_ASSSIGNMENT_STRATEGY_NONE.equals(assignmentConfigs.rackAwareAssignmentStrategy))
{
+ canEnableForActive = false;
+ return false;
+ }
+ */
+
+ if (!validateClientRack()) {
+ canEnableForActive = false;
+ return false;
+ }
+
+ canEnableForActive = validateTopicPartitionRack();
+ return canEnableForActive;
+ }
+
+ public boolean canEnableForStandby() {
+ // TODO
+ return false;
+ }
+
+ private boolean validateTopicPartitionRack() {
+ // Make sure rackId exist for all TopicPartitions needed
+ final Set<String> topicsToDescribe = new HashSet<>();
+ for (final Set<TopicPartition> topicPartitions :
partitionsForTask.values()) {
+ for (TopicPartition topicPartition : topicPartitions) {
+ final PartitionInfo partitionInfo =
fullMetadata.partition(topicPartition);
+ if (partitionInfo == null) {
+ log.error("TopicPartition {} doesn't exist in cluster",
topicPartition);
+ return false;
+ }
+ final Node[] replica = partitionInfo.replicas();
+ if (replica == null || replica.length == 0) {
+ topicsToDescribe.add(topicPartition.topic());
Review Comment:
If an external topic doesn't have node information, we will also try to
describe it
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RackAwareTaskAssignor.java:
##########
@@ -0,0 +1,164 @@
+package org.apache.kafka.streams.processor.internals;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackAwareTaskAssignor {
+ private static final Logger log =
LoggerFactory.getLogger(RackAwareTaskAssignor.class);
+
+ private final Cluster fullMetadata;
+ private final Map<TaskId, Set<TopicPartition>> partitionsForTask;
+ private final Map<UUID, Map<String, Optional<String>>> processRacks;
+ private final AssignmentConfigs assignmentConfigs;
+ private final Map<TopicPartition, Set<String>> racksForPartition;
+ private final InternalTopicManager internalTopicManager;
+ private Boolean canEnableForActive;
+
+ public RackAwareTaskAssignor(final Cluster fullMetadata,
+ final Map<TaskId, Set<TopicPartition>>
partitionsForTask,
+ final Map<Subtopology, Set<TaskId>>
tasksForTopicGroup,
+ final Map<UUID, Map<String,
Optional<String>>> processRacks,
+ final InternalTopicManager
internalTopicManager,
+ final AssignmentConfigs assignmentConfigs) {
+ this.fullMetadata = fullMetadata;
+ this.partitionsForTask = partitionsForTask;
+ this.processRacks = processRacks;
+ this.internalTopicManager = internalTopicManager;
+ this.assignmentConfigs = assignmentConfigs;
+ this.racksForPartition = new HashMap<>();
+ }
+
+ public synchronized boolean canEnableForActive() {
+ if (canEnableForActive != null) {
+ return canEnableForActive;
+ }
+
+ /*
+ TODO: enable this after we add the config
+ if
(StreamsConfig.RACK_AWARE_ASSSIGNMENT_STRATEGY_NONE.equals(assignmentConfigs.rackAwareAssignmentStrategy))
{
+ canEnableForActive = false;
+ return false;
+ }
+ */
+
+ if (!validateClientRack()) {
+ canEnableForActive = false;
+ return false;
+ }
+
+ canEnableForActive = validateTopicPartitionRack();
+ return canEnableForActive;
+ }
+
+ public boolean canEnableForStandby() {
+ // TODO
+ return false;
+ }
+
+ private boolean validateTopicPartitionRack() {
+ // Make sure rackId exist for all TopicPartitions needed
+ final Set<String> topicsToDescribe = new HashSet<>();
+ for (final Set<TopicPartition> topicPartitions :
partitionsForTask.values()) {
+ for (TopicPartition topicPartition : topicPartitions) {
+ final PartitionInfo partitionInfo =
fullMetadata.partition(topicPartition);
+ if (partitionInfo == null) {
+ log.error("TopicPartition {} doesn't exist in cluster",
topicPartition);
+ return false;
+ }
+ final Node[] replica = partitionInfo.replicas();
+ if (replica == null || replica.length == 0) {
Review Comment:
Yes. It's 0 for repartition topics we create:
https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/RepartitionTopics.java#L99
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RackAwareTaskAssignor.java:
##########
@@ -0,0 +1,164 @@
+package org.apache.kafka.streams.processor.internals;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackAwareTaskAssignor {
+ private static final Logger log =
LoggerFactory.getLogger(RackAwareTaskAssignor.class);
+
+ private final Cluster fullMetadata;
+ private final Map<TaskId, Set<TopicPartition>> partitionsForTask;
+ private final Map<UUID, Map<String, Optional<String>>> processRacks;
+ private final AssignmentConfigs assignmentConfigs;
+ private final Map<TopicPartition, Set<String>> racksForPartition;
+ private final InternalTopicManager internalTopicManager;
+ private Boolean canEnableForActive;
+
+ public RackAwareTaskAssignor(final Cluster fullMetadata,
+ final Map<TaskId, Set<TopicPartition>>
partitionsForTask,
+ final Map<Subtopology, Set<TaskId>>
tasksForTopicGroup,
+ final Map<UUID, Map<String,
Optional<String>>> processRacks,
+ final InternalTopicManager
internalTopicManager,
+ final AssignmentConfigs assignmentConfigs) {
+ this.fullMetadata = fullMetadata;
+ this.partitionsForTask = partitionsForTask;
+ this.processRacks = processRacks;
+ this.internalTopicManager = internalTopicManager;
+ this.assignmentConfigs = assignmentConfigs;
+ this.racksForPartition = new HashMap<>();
+ }
+
+ public synchronized boolean canEnableForActive() {
+ if (canEnableForActive != null) {
+ return canEnableForActive;
+ }
+
+ /*
+ TODO: enable this after we add the config
+ if
(StreamsConfig.RACK_AWARE_ASSSIGNMENT_STRATEGY_NONE.equals(assignmentConfigs.rackAwareAssignmentStrategy))
{
+ canEnableForActive = false;
+ return false;
+ }
+ */
+
+ if (!validateClientRack()) {
+ canEnableForActive = false;
+ return false;
+ }
+
+ canEnableForActive = validateTopicPartitionRack();
+ return canEnableForActive;
+ }
+
+ public boolean canEnableForStandby() {
+ // TODO
+ return false;
+ }
+
+ private boolean validateTopicPartitionRack() {
+ // Make sure rackId exist for all TopicPartitions needed
+ final Set<String> topicsToDescribe = new HashSet<>();
+ for (final Set<TopicPartition> topicPartitions :
partitionsForTask.values()) {
+ for (TopicPartition topicPartition : topicPartitions) {
+ final PartitionInfo partitionInfo =
fullMetadata.partition(topicPartition);
+ if (partitionInfo == null) {
+ log.error("TopicPartition {} doesn't exist in cluster",
topicPartition);
+ return false;
+ }
+ final Node[] replica = partitionInfo.replicas();
+ if (replica == null || replica.length == 0) {
+ topicsToDescribe.add(topicPartition.topic());
+ continue;
+ }
+ for (final Node node : replica) {
+ if (node.hasRack()) {
+ racksForPartition.computeIfAbsent(topicPartition, k ->
new HashSet<>()).add(node.rack());
Review Comment:
Good catch
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java:
##########
@@ -305,6 +327,31 @@ public void
shouldThrowTimeoutExceptionIfTopicExistsDuringSetup() {
);
}
+ @Test
+ public void
shouldThrowTimeoutExceptionIfGetPartitionInfoHasTopicDescriptionTimeout() {
+ mockAdminClient.timeoutNextRequest(1);
+
+ final InternalTopicManager internalTopicManager =
+ new InternalTopicManager(time, mockAdminClient, new
StreamsConfig(config));
+ try {
+ final Set<String> topic1set = new
HashSet<>(Collections.singletonList(topic1));
+ internalTopicManager.getTopicPartitionInfo(topic1set, null);
+
+ } catch (final TimeoutException expected) {
+ assertEquals(TimeoutException.class,
expected.getCause().getClass());
+ }
+
+ mockAdminClient.timeoutNextRequest(1);
Review Comment:
This is actually testing `getTopicPartitionInfo(final Set<String> topics,
final Set<String> tempUnknownTopics)`
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RackAwareTaskAssignor.java:
##########
@@ -0,0 +1,164 @@
+package org.apache.kafka.streams.processor.internals;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RackAwareTaskAssignor {
+ private static final Logger log =
LoggerFactory.getLogger(RackAwareTaskAssignor.class);
+
+ private final Cluster fullMetadata;
+ private final Map<TaskId, Set<TopicPartition>> partitionsForTask;
+ private final Map<UUID, Map<String, Optional<String>>> processRacks;
+ private final AssignmentConfigs assignmentConfigs;
+ private final Map<TopicPartition, Set<String>> racksForPartition;
+ private final InternalTopicManager internalTopicManager;
+ private Boolean canEnableForActive;
+
+ public RackAwareTaskAssignor(final Cluster fullMetadata,
+ final Map<TaskId, Set<TopicPartition>>
partitionsForTask,
+ final Map<Subtopology, Set<TaskId>>
tasksForTopicGroup,
+ final Map<UUID, Map<String,
Optional<String>>> processRacks,
+ final InternalTopicManager
internalTopicManager,
+ final AssignmentConfigs assignmentConfigs) {
+ this.fullMetadata = fullMetadata;
+ this.partitionsForTask = partitionsForTask;
+ this.processRacks = processRacks;
+ this.internalTopicManager = internalTopicManager;
+ this.assignmentConfigs = assignmentConfigs;
+ this.racksForPartition = new HashMap<>();
+ }
+
+ public synchronized boolean canEnableForActive() {
+ if (canEnableForActive != null) {
+ return canEnableForActive;
+ }
+
+ /*
+ TODO: enable this after we add the config
+ if
(StreamsConfig.RACK_AWARE_ASSSIGNMENT_STRATEGY_NONE.equals(assignmentConfigs.rackAwareAssignmentStrategy))
{
+ canEnableForActive = false;
+ return false;
+ }
+ */
+
+ if (!validateClientRack()) {
+ canEnableForActive = false;
+ return false;
+ }
+
+ canEnableForActive = validateTopicPartitionRack();
+ return canEnableForActive;
+ }
+
+ public boolean canEnableForStandby() {
+ // TODO
+ return false;
+ }
+
+ private boolean validateTopicPartitionRack() {
+ // Make sure rackId exist for all TopicPartitions needed
+ final Set<String> topicsToDescribe = new HashSet<>();
+ for (final Set<TopicPartition> topicPartitions :
partitionsForTask.values()) {
+ for (TopicPartition topicPartition : topicPartitions) {
+ final PartitionInfo partitionInfo =
fullMetadata.partition(topicPartition);
+ if (partitionInfo == null) {
+ log.error("TopicPartition {} doesn't exist in cluster",
topicPartition);
+ return false;
+ }
+ final Node[] replica = partitionInfo.replicas();
+ if (replica == null || replica.length == 0) {
+ topicsToDescribe.add(topicPartition.topic());
+ continue;
+ }
+ for (final Node node : replica) {
+ if (node.hasRack()) {
+ racksForPartition.computeIfAbsent(topicPartition, k ->
new HashSet<>()).add(node.rack());
+ }
+ }
+ }
+ }
+
+ if (!topicsToDescribe.isEmpty()) {
+ log.info("Fetching PartitionInfo for topics {}", topicsToDescribe);
+ try {
+ final Map<String, List<TopicPartitionInfo>> topicPartitionInfo =
internalTopicManager.getTopicPartitionInfo(topicsToDescribe);
+ if (topicsToDescribe.size() > topicPartitionInfo.size()) {
+ topicsToDescribe.removeAll(topicPartitionInfo.keySet());
Review Comment:
To get the ones which fail to be described. It's used again in log on next
line.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java:
##########
@@ -279,6 +279,28 @@ public CreateTopicsResult createTopics(final
Collection<NewTopic> newTopics,
+ " You can change the replication.factor config or upgrade
your brokers to version 2.4 or newer to avoid this error."));
}
+ @Test
+ public void shouldThrowTimeoutExceptionInGetPartitionInfo() {
+ setupTopicInMockAdminClient(topic1, Collections.emptyMap());
+ final MockTime time = new MockTime(
+ (Integer)
config.get(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG))
/ 15
Review Comment:
I think this is `autoTickMs`, if set to 0, time won't increase?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/RackAwareTaskAssignorTest.java:
##########
@@ -0,0 +1,264 @@
+package org.apache.kafka.streams.processor.internals.assignment;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import org.apache.kafka.clients.admin.MockAdminClient;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.StreamsConfig.InternalConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.internals.RackAwareTaskAssignor;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.apache.kafka.test.MockClientSupplier;
+import org.apache.kafka.test.MockInternalTopicManager;
+import org.junit.Test;
+
+public class RackAwareTaskAssignorTest {
+ private final static String TOPIC0 = "topic0";
+ private final static String TOPIC1 = "topic1";
+ private static final String USER_END_POINT = "localhost:8080";
+ private static final String APPLICATION_ID =
"stream-partition-assignor-test";
+
+ private final Node node0 = new Node(0, "node0", 1, "rack1");
+ private final Node node1 = new Node(1, "node1", 1, "rack2");
+ private final Node node2 = new Node(2, "node2", 1, "rack3");
+ private final Node[] replicas = new Node[] {node0, node1, node2};
+
+ private final PartitionInfo partitionInfo00 = new PartitionInfo(TOPIC0, 0,
node0, replicas, replicas);
+ private final PartitionInfo partitionInfo01 = new PartitionInfo(TOPIC0, 1,
node0, replicas, replicas);
+
+ private final TopicPartition tp00 = new TopicPartition(TOPIC0, 0);
+ private final TopicPartition tp01 = new TopicPartition(TOPIC0, 1);
+ private final TopicPartition tp10 = new TopicPartition(TOPIC1, 0);
+
+ private final UUID process0UUID = UUID.randomUUID();
+ private final UUID process1UUID = UUID.randomUUID();
+
+ private final Subtopology subtopology1 = new Subtopology(1, "topology1");
+
+ private final MockTime time = new MockTime();
+ private final StreamsConfig streamsConfig = new
StreamsConfig(configProps());
+ private final MockClientSupplier mockClientSupplier = new
MockClientSupplier();
+
+ private final MockInternalTopicManager mockInternalTopicManager = new
MockInternalTopicManager(
+ time,
+ streamsConfig,
+ mockClientSupplier.restoreConsumer,
+ false
+ );
+
+ private Map<String, Object> configProps() {
+ final Map<String, Object> configurationMap = new HashMap<>();
+ configurationMap.put(StreamsConfig.APPLICATION_ID_CONFIG,
APPLICATION_ID);
+ configurationMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
USER_END_POINT);
+
+ final ReferenceContainer referenceContainer = new ReferenceContainer();
+ /*
Review Comment:
Might need later after more implementation...
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/RackAwareTaskAssignorTest.java:
##########
@@ -0,0 +1,264 @@
+package org.apache.kafka.streams.processor.internals.assignment;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import org.apache.kafka.clients.admin.MockAdminClient;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.StreamsConfig.InternalConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.internals.RackAwareTaskAssignor;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.apache.kafka.test.MockClientSupplier;
+import org.apache.kafka.test.MockInternalTopicManager;
+import org.junit.Test;
+
+public class RackAwareTaskAssignorTest {
+ private final static String TOPIC0 = "topic0";
+ private final static String TOPIC1 = "topic1";
+ private static final String USER_END_POINT = "localhost:8080";
+ private static final String APPLICATION_ID =
"stream-partition-assignor-test";
+
+ private final Node node0 = new Node(0, "node0", 1, "rack1");
+ private final Node node1 = new Node(1, "node1", 1, "rack2");
+ private final Node node2 = new Node(2, "node2", 1, "rack3");
+ private final Node[] replicas = new Node[] {node0, node1, node2};
+
+ private final PartitionInfo partitionInfo00 = new PartitionInfo(TOPIC0, 0,
node0, replicas, replicas);
+ private final PartitionInfo partitionInfo01 = new PartitionInfo(TOPIC0, 1,
node0, replicas, replicas);
+
+ private final TopicPartition tp00 = new TopicPartition(TOPIC0, 0);
+ private final TopicPartition tp01 = new TopicPartition(TOPIC0, 1);
+ private final TopicPartition tp10 = new TopicPartition(TOPIC1, 0);
+
+ private final UUID process0UUID = UUID.randomUUID();
+ private final UUID process1UUID = UUID.randomUUID();
+
+ private final Subtopology subtopology1 = new Subtopology(1, "topology1");
+
+ private final MockTime time = new MockTime();
+ private final StreamsConfig streamsConfig = new
StreamsConfig(configProps());
+ private final MockClientSupplier mockClientSupplier = new
MockClientSupplier();
+
+ private final MockInternalTopicManager mockInternalTopicManager = new
MockInternalTopicManager(
+ time,
+ streamsConfig,
+ mockClientSupplier.restoreConsumer,
+ false
+ );
+
+ private Map<String, Object> configProps() {
+ final Map<String, Object> configurationMap = new HashMap<>();
+ configurationMap.put(StreamsConfig.APPLICATION_ID_CONFIG,
APPLICATION_ID);
+ configurationMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
USER_END_POINT);
+
+ final ReferenceContainer referenceContainer = new ReferenceContainer();
+ /*
+ referenceContainer.mainConsumer = consumer;
+ referenceContainer.adminClient = adminClient;
+ referenceContainer.taskManager = taskManager;
+ referenceContainer.streamsMetadataState = streamsMetadataState;
+ referenceContainer.time = time;
+ */
+
configurationMap.put(InternalConfig.REFERENCE_CONTAINER_PARTITION_ASSIGNOR,
referenceContainer);
+ return configurationMap;
+ }
+
+ @Test
+ public void disableActiveSinceMissingClusterInfo() {
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2)),
+ new HashSet<>(Arrays.asList(partitionInfo00, partitionInfo01)),
+ Collections.emptySet(),
+ Collections.emptySet()
+ );
+
+ final Map<UUID, Map<String, Optional<String>>> processRacks = new
HashMap<>();
+
+ processRacks.put(process0UUID , Collections.singletonMap("consumer1",
Optional.of("rack1")));
+
+ final RackAwareTaskAssignor assignor = new RackAwareTaskAssignor(
+ metadata,
+ Collections.singletonMap(new TaskId(1, 1), new
HashSet<>(Arrays.asList(tp00, tp10))),
Review Comment:
For simplicity, this is just 1 task.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/RackAwareTaskAssignorTest.java:
##########
@@ -0,0 +1,264 @@
+package org.apache.kafka.streams.processor.internals.assignment;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import org.apache.kafka.clients.admin.MockAdminClient;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.StreamsConfig.InternalConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.internals.RackAwareTaskAssignor;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.apache.kafka.test.MockClientSupplier;
+import org.apache.kafka.test.MockInternalTopicManager;
+import org.junit.Test;
+
+public class RackAwareTaskAssignorTest {
+ private final static String TOPIC0 = "topic0";
+ private final static String TOPIC1 = "topic1";
+ private static final String USER_END_POINT = "localhost:8080";
+ private static final String APPLICATION_ID =
"stream-partition-assignor-test";
+
+ private final Node node0 = new Node(0, "node0", 1, "rack1");
+ private final Node node1 = new Node(1, "node1", 1, "rack2");
+ private final Node node2 = new Node(2, "node2", 1, "rack3");
+ private final Node[] replicas = new Node[] {node0, node1, node2};
+
+ private final PartitionInfo partitionInfo00 = new PartitionInfo(TOPIC0, 0,
node0, replicas, replicas);
+ private final PartitionInfo partitionInfo01 = new PartitionInfo(TOPIC0, 1,
node0, replicas, replicas);
+
+ private final TopicPartition tp00 = new TopicPartition(TOPIC0, 0);
+ private final TopicPartition tp01 = new TopicPartition(TOPIC0, 1);
+ private final TopicPartition tp10 = new TopicPartition(TOPIC1, 0);
+
+ private final UUID process0UUID = UUID.randomUUID();
+ private final UUID process1UUID = UUID.randomUUID();
+
+ private final Subtopology subtopology1 = new Subtopology(1, "topology1");
+
+ private final MockTime time = new MockTime();
+ private final StreamsConfig streamsConfig = new
StreamsConfig(configProps());
+ private final MockClientSupplier mockClientSupplier = new
MockClientSupplier();
+
+ private final MockInternalTopicManager mockInternalTopicManager = new
MockInternalTopicManager(
+ time,
+ streamsConfig,
+ mockClientSupplier.restoreConsumer,
+ false
+ );
+
+ private Map<String, Object> configProps() {
+ final Map<String, Object> configurationMap = new HashMap<>();
+ configurationMap.put(StreamsConfig.APPLICATION_ID_CONFIG,
APPLICATION_ID);
+ configurationMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
USER_END_POINT);
+
+ final ReferenceContainer referenceContainer = new ReferenceContainer();
+ /*
+ referenceContainer.mainConsumer = consumer;
+ referenceContainer.adminClient = adminClient;
+ referenceContainer.taskManager = taskManager;
+ referenceContainer.streamsMetadataState = streamsMetadataState;
+ referenceContainer.time = time;
+ */
+
configurationMap.put(InternalConfig.REFERENCE_CONTAINER_PARTITION_ASSIGNOR,
referenceContainer);
+ return configurationMap;
+ }
+
+ @Test
+ public void disableActiveSinceMissingClusterInfo() {
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2)),
+ new HashSet<>(Arrays.asList(partitionInfo00, partitionInfo01)),
+ Collections.emptySet(),
+ Collections.emptySet()
+ );
+
+ final Map<UUID, Map<String, Optional<String>>> processRacks = new
HashMap<>();
+
+ processRacks.put(process0UUID , Collections.singletonMap("consumer1",
Optional.of("rack1")));
+
+ final RackAwareTaskAssignor assignor = new RackAwareTaskAssignor(
+ metadata,
+ Collections.singletonMap(new TaskId(1, 1), new
HashSet<>(Arrays.asList(tp00, tp10))),
+ Collections.singletonMap(subtopology1, Collections.singleton(new
TaskId(1, 1))),
+ processRacks,
+ mockInternalTopicManager,
+ new
AssignorConfiguration(streamsConfig.originals()).assignmentConfigs()
+ );
+
+ // False since tp10 is missing in cluster metadata
+ assertFalse(assignor.canEnableForActive());
+ }
+
+ @Test
+ public void disableActiveSinceRackMissingInClient() {
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2)),
+ new HashSet<>(Arrays.asList(partitionInfo00, partitionInfo01)),
+ Collections.emptySet(),
+ Collections.emptySet()
+ );
+
+ final Map<UUID, Map<String, Optional<String>>> processRacks = new
HashMap<>();
+
+ // Missing rackId config in client
+ processRacks.put(process0UUID , Collections.singletonMap("consumer1",
Optional.empty()));
+
+ final RackAwareTaskAssignor assignor = new RackAwareTaskAssignor(
+ metadata,
+ Collections.singletonMap(new TaskId(1, 1),
Collections.singleton(tp00)),
+ Collections.singletonMap(subtopology1, Collections.singleton(new
TaskId(1, 1))),
+ processRacks,
+ mockInternalTopicManager,
+ new
AssignorConfiguration(streamsConfig.originals()).assignmentConfigs()
+ );
+
+ // False since process1 doesn't have rackId
+ assertFalse(assignor.canEnableForActive());
+ }
+
+ @Test
+ public void disableActiveSinceRackDiffersInSameProcess() {
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2)),
+ new HashSet<>(Arrays.asList(partitionInfo00, partitionInfo01)),
+ Collections.emptySet(),
+ Collections.emptySet()
+ );
+
+ final Map<UUID, Map<String, Optional<String>>> processRacks = new
HashMap<>();
+
+ // Different rackId for same process
+ processRacks.computeIfAbsent(process0UUID , k -> new
HashMap<>()).put("consumer1", Optional.of("rack1"));
+ processRacks.computeIfAbsent(process0UUID , k -> new
HashMap<>()).put("consumer2", Optional.of("rack2"));
+
+ final RackAwareTaskAssignor assignor = new RackAwareTaskAssignor(
+ metadata,
+ Collections.singletonMap(new TaskId(1, 1),
Collections.singleton(tp00)),
+ Collections.singletonMap(subtopology1, Collections.singleton(new
TaskId(1, 1))),
+ processRacks,
+ mockInternalTopicManager,
+ new
AssignorConfiguration(streamsConfig.originals()).assignmentConfigs()
+ );
+
+ assertFalse(assignor.canEnableForActive());
+ }
+
+ @Test
+ public void enableRackAwareAssignorForActiveWithoutDescribingTopics() {
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2)),
+ new HashSet<>(Arrays.asList(partitionInfo00, partitionInfo01)),
+ Collections.emptySet(),
+ Collections.emptySet()
+ );
+
+ final Map<UUID, Map<String, Optional<String>>> processRacks = new
HashMap<>();
+
+ processRacks.computeIfAbsent(process0UUID , k -> new
HashMap<>()).put("consumer1", Optional.of("rack1"));
+
+ final RackAwareTaskAssignor assignor = new RackAwareTaskAssignor(
+ metadata,
+ Collections.singletonMap(new TaskId(1, 1),
Collections.singleton(tp00)),
+ Collections.singletonMap(subtopology1, Collections.singleton(new
TaskId(1, 1))),
+ processRacks,
+ mockInternalTopicManager,
+ new
AssignorConfiguration(streamsConfig.originals()).assignmentConfigs()
+ );
+
+ // tp00 has rackInfo in cluster metadata
+ assertTrue(assignor.canEnableForActive());
+ }
+
+ @Test
+ public void enableRackAwareAssignorForActiveWithDescribingTopics() {
+ final PartitionInfo noNodeInfo = new PartitionInfo(TOPIC0, 0, null,
new Node[0], new Node[0]);
+
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2, Node.noNode())),
// mockClientSupplier.setCluster requires noNode
Review Comment:
https://github.com/apache/kafka/blob/trunk/streams/src/test/java/org/apache/kafka/test/MockClientSupplier.java#L56
has `cluster.nodeById(-1)`
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/RackAwareTaskAssignorTest.java:
##########
@@ -0,0 +1,264 @@
+package org.apache.kafka.streams.processor.internals.assignment;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import org.apache.kafka.clients.admin.MockAdminClient;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.StreamsConfig.InternalConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.internals.RackAwareTaskAssignor;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.apache.kafka.test.MockClientSupplier;
+import org.apache.kafka.test.MockInternalTopicManager;
+import org.junit.Test;
+
+public class RackAwareTaskAssignorTest {
+ private final static String TOPIC0 = "topic0";
+ private final static String TOPIC1 = "topic1";
+ private static final String USER_END_POINT = "localhost:8080";
+ private static final String APPLICATION_ID =
"stream-partition-assignor-test";
+
+ private final Node node0 = new Node(0, "node0", 1, "rack1");
+ private final Node node1 = new Node(1, "node1", 1, "rack2");
+ private final Node node2 = new Node(2, "node2", 1, "rack3");
+ private final Node[] replicas = new Node[] {node0, node1, node2};
+
+ private final PartitionInfo partitionInfo00 = new PartitionInfo(TOPIC0, 0,
node0, replicas, replicas);
+ private final PartitionInfo partitionInfo01 = new PartitionInfo(TOPIC0, 1,
node0, replicas, replicas);
+
+ private final TopicPartition tp00 = new TopicPartition(TOPIC0, 0);
+ private final TopicPartition tp01 = new TopicPartition(TOPIC0, 1);
+ private final TopicPartition tp10 = new TopicPartition(TOPIC1, 0);
+
+ private final UUID process0UUID = UUID.randomUUID();
+ private final UUID process1UUID = UUID.randomUUID();
+
+ private final Subtopology subtopology1 = new Subtopology(1, "topology1");
+
+ private final MockTime time = new MockTime();
+ private final StreamsConfig streamsConfig = new
StreamsConfig(configProps());
+ private final MockClientSupplier mockClientSupplier = new
MockClientSupplier();
+
+ private final MockInternalTopicManager mockInternalTopicManager = new
MockInternalTopicManager(
+ time,
+ streamsConfig,
+ mockClientSupplier.restoreConsumer,
+ false
+ );
+
+ private Map<String, Object> configProps() {
+ final Map<String, Object> configurationMap = new HashMap<>();
+ configurationMap.put(StreamsConfig.APPLICATION_ID_CONFIG,
APPLICATION_ID);
+ configurationMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
USER_END_POINT);
+
+ final ReferenceContainer referenceContainer = new ReferenceContainer();
+ /*
+ referenceContainer.mainConsumer = consumer;
+ referenceContainer.adminClient = adminClient;
+ referenceContainer.taskManager = taskManager;
+ referenceContainer.streamsMetadataState = streamsMetadataState;
+ referenceContainer.time = time;
+ */
+
configurationMap.put(InternalConfig.REFERENCE_CONTAINER_PARTITION_ASSIGNOR,
referenceContainer);
+ return configurationMap;
+ }
+
+ @Test
+ public void disableActiveSinceMissingClusterInfo() {
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2)),
+ new HashSet<>(Arrays.asList(partitionInfo00, partitionInfo01)),
+ Collections.emptySet(),
+ Collections.emptySet()
+ );
+
+ final Map<UUID, Map<String, Optional<String>>> processRacks = new
HashMap<>();
+
+ processRacks.put(process0UUID , Collections.singletonMap("consumer1",
Optional.of("rack1")));
+
+ final RackAwareTaskAssignor assignor = new RackAwareTaskAssignor(
+ metadata,
+ Collections.singletonMap(new TaskId(1, 1), new
HashSet<>(Arrays.asList(tp00, tp10))),
+ Collections.singletonMap(subtopology1, Collections.singleton(new
TaskId(1, 1))),
+ processRacks,
+ mockInternalTopicManager,
+ new
AssignorConfiguration(streamsConfig.originals()).assignmentConfigs()
+ );
+
+ // False since tp10 is missing in cluster metadata
+ assertFalse(assignor.canEnableForActive());
+ }
+
+ @Test
+ public void disableActiveSinceRackMissingInClient() {
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2)),
+ new HashSet<>(Arrays.asList(partitionInfo00, partitionInfo01)),
Review Comment:
We don't need to. The test is mainly testing
`processRacks.put(process0UUID , Collections.singletonMap("consumer1",
Optional.empty()));`
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/RackAwareTaskAssignorTest.java:
##########
@@ -0,0 +1,264 @@
+package org.apache.kafka.streams.processor.internals.assignment;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import org.apache.kafka.clients.admin.MockAdminClient;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.StreamsConfig.InternalConfig;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.internals.RackAwareTaskAssignor;
+import
org.apache.kafka.streams.processor.internals.TopologyMetadata.Subtopology;
+import
org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
+import org.apache.kafka.test.MockClientSupplier;
+import org.apache.kafka.test.MockInternalTopicManager;
+import org.junit.Test;
+
+public class RackAwareTaskAssignorTest {
+ private final static String TOPIC0 = "topic0";
+ private final static String TOPIC1 = "topic1";
+ private static final String USER_END_POINT = "localhost:8080";
+ private static final String APPLICATION_ID =
"stream-partition-assignor-test";
+
+ private final Node node0 = new Node(0, "node0", 1, "rack1");
+ private final Node node1 = new Node(1, "node1", 1, "rack2");
+ private final Node node2 = new Node(2, "node2", 1, "rack3");
+ private final Node[] replicas = new Node[] {node0, node1, node2};
+
+ private final PartitionInfo partitionInfo00 = new PartitionInfo(TOPIC0, 0,
node0, replicas, replicas);
+ private final PartitionInfo partitionInfo01 = new PartitionInfo(TOPIC0, 1,
node0, replicas, replicas);
+
+ private final TopicPartition tp00 = new TopicPartition(TOPIC0, 0);
+ private final TopicPartition tp01 = new TopicPartition(TOPIC0, 1);
+ private final TopicPartition tp10 = new TopicPartition(TOPIC1, 0);
+
+ private final UUID process0UUID = UUID.randomUUID();
+ private final UUID process1UUID = UUID.randomUUID();
+
+ private final Subtopology subtopology1 = new Subtopology(1, "topology1");
+
+ private final MockTime time = new MockTime();
+ private final StreamsConfig streamsConfig = new
StreamsConfig(configProps());
+ private final MockClientSupplier mockClientSupplier = new
MockClientSupplier();
+
+ private final MockInternalTopicManager mockInternalTopicManager = new
MockInternalTopicManager(
+ time,
+ streamsConfig,
+ mockClientSupplier.restoreConsumer,
+ false
+ );
+
+ private Map<String, Object> configProps() {
+ final Map<String, Object> configurationMap = new HashMap<>();
+ configurationMap.put(StreamsConfig.APPLICATION_ID_CONFIG,
APPLICATION_ID);
+ configurationMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
USER_END_POINT);
+
+ final ReferenceContainer referenceContainer = new ReferenceContainer();
+ /*
+ referenceContainer.mainConsumer = consumer;
+ referenceContainer.adminClient = adminClient;
+ referenceContainer.taskManager = taskManager;
+ referenceContainer.streamsMetadataState = streamsMetadataState;
+ referenceContainer.time = time;
+ */
+
configurationMap.put(InternalConfig.REFERENCE_CONTAINER_PARTITION_ASSIGNOR,
referenceContainer);
+ return configurationMap;
+ }
+
+ @Test
+ public void disableActiveSinceMissingClusterInfo() {
+ final Cluster metadata = new Cluster(
+ "cluster",
+ new HashSet<>(Arrays.asList(node0, node1, node2)),
+ new HashSet<>(Arrays.asList(partitionInfo00, partitionInfo01)),
+ Collections.emptySet(),
+ Collections.emptySet()
+ );
+
+ final Map<UUID, Map<String, Optional<String>>> processRacks = new
HashMap<>();
+
+ processRacks.put(process0UUID , Collections.singletonMap("consumer1",
Optional.of("rack1")));
+
+ final RackAwareTaskAssignor assignor = new RackAwareTaskAssignor(
+ metadata,
+ Collections.singletonMap(new TaskId(1, 1), new
HashSet<>(Arrays.asList(tp00, tp10))),
+ Collections.singletonMap(subtopology1, Collections.singleton(new
TaskId(1, 1))),
+ processRacks,
+ mockInternalTopicManager,
+ new
AssignorConfiguration(streamsConfig.originals()).assignmentConfigs()
+ );
+
+ // False since tp10 is missing in cluster metadata
+ assertFalse(assignor.canEnableForActive());
Review Comment:
Ideally yes. But this might be hard to verify... May make
`validateTopicPartitionRack` public and verify it's false? Or we have to extract
```
final PartitionInfo partitionInfo =
fullMetadata.partition(topicPartition);
if (partitionInfo == null) {
log.error("TopicPartition {} doesn't exist in cluster",
topicPartition);
return false;
}
```
to a public function and verify that returns false?
--
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
