diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/assignment/TaskTopicPartition.java b/streams/src/main/java/org/apache/kafka/streams/processor/assignment/TaskTopicPartition.java new file mode 100644 index 00000000000..c2422a4d875 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/processor/assignment/TaskTopicPartition.java @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.processor.assignment; + +import java.util.Optional; +import java.util.Set; +import org.apache.kafka.common.TopicPartition; + +/** + * This is a simple container class used during the assignment process to distinguish + * TopicPartitions type. Since the assignment logic can depend on the type of topic we're + * looking at, and the rack information of the partition, this container class should have + * everything necessary to make informed task assignment decisions. + */ +public interface TaskTopicPartition { + /** + * + * @return the {@code TopicPartition} for this task. + */ + TopicPartition topicPartition(); + + /** + * + * @return whether the underlying topic is a source topic or not. Source changelog topics + * are both source topics and changelog topics. + */ + boolean isSource(); + + /** + * + * @return whether the underlying topic is a changelog topic or not. Source changelog topics + * are both source topics and changelog topics. + */ + boolean isChangelog(); + + /** + * + * @return the broker rack ids on which this topic partition resides. If no information could + * be found, this will return an empty optional value. + */ + Optional> rackIds(); +} diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/DefaultTaskTopicPartition.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/DefaultTaskTopicPartition.java new file mode 100644 index 00000000000..815aa1ff64c --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/DefaultTaskTopicPartition.java @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.processor.internals.assignment; + +import java.util.Optional; +import java.util.Set; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.streams.processor.assignment.TaskTopicPartition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * This is a simple container class used during the assignment process to distinguish + * TopicPartitions type. Since the assignment logic can depend on the type of topic we're + * looking at, and the rack information of the partition, this container class should have + * everything necessary to make informed task assignment decisions. + */ +public class DefaultTaskTopicPartition implements TaskTopicPartition { + private static final Logger LOG = LoggerFactory.getLogger(DefaultTaskTopicPartition.class); + + private final TopicPartition topicPartition; + private final boolean isSourceTopic; + private final boolean isChangelogTopic; + private final Optional> rackIds; + + public DefaultTaskTopicPartition(final TopicPartition topicPartition, + final boolean isSourceTopic, + final boolean isChangelogTopic, + final Set rackIds) { + this.topicPartition = topicPartition; + this.isSourceTopic = isSourceTopic; + this.isChangelogTopic = isChangelogTopic; + this.rackIds = Optional.ofNullable(rackIds); + } + + @Override + public TopicPartition topicPartition() { + return topicPartition; + } + + @Override + public boolean isSource() { + return isSourceTopic; + } + + @Override + public boolean isChangelog() { + return isChangelogTopic; + } + + @Override + public Optional> rackIds() { + return rackIds; + } +}