kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: KAFKA-4427: Skip topic groups with no tasks
Date Tue, 29 Nov 2016 19:07:43 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 7d3aa01ce -> 7d0f3e75a


KAFKA-4427: Skip topic groups with no tasks

Author: Eno Thereska <eno.thereska@gmail.com>

Reviewers: Damian Guy <damian.guy@gmail.com>, Guozhang Wang <wangguoz@gmail.com>

Closes #2171 from enothereska/KAFKA-4427-topicgroups-with-no-tasks


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/7d0f3e75
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/7d0f3e75
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/7d0f3e75

Branch: refs/heads/trunk
Commit: 7d0f3e75ad2934b8ad5b84f81707b01285022523
Parents: 7d3aa01
Author: Eno Thereska <eno.thereska@gmail.com>
Authored: Tue Nov 29 11:07:39 2016 -0800
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Tue Nov 29 11:07:39 2016 -0800

----------------------------------------------------------------------
 .../internals/StreamPartitionAssignor.java      | 19 ++++----
 .../SingleGroupPartitionGrouperStub.java        | 47 ++++++++++++++++++++
 .../internals/StreamPartitionAssignorTest.java  | 40 +++++++++++++++++
 3 files changed, 98 insertions(+), 8 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/7d0f3e75/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java
index b06d7f7..84f78dc 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java
@@ -430,15 +430,18 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable
             for (InternalTopicConfig topicConfig : stateChangelogTopics.values()) {
                 // the expected number of partitions is the max value of TaskId.partition
+ 1
                 int numPartitions = -1;
-                for (TaskId task : tasksByTopicGroup.get(topicGroupId)) {
-                    if (numPartitions < task.partition + 1)
-                        numPartitions = task.partition + 1;
-                }
-
-                InternalTopicMetadata topicMetadata = new InternalTopicMetadata(topicConfig);
-                topicMetadata.numPartitions = numPartitions;
+                if (tasksByTopicGroup.get(topicGroupId) != null) {
+                    for (TaskId task : tasksByTopicGroup.get(topicGroupId)) {
+                        if (numPartitions < task.partition + 1)
+                            numPartitions = task.partition + 1;
+                    }
+                    InternalTopicMetadata topicMetadata = new InternalTopicMetadata(topicConfig);
+                    topicMetadata.numPartitions = numPartitions;
 
-                changelogTopicMetadata.put(topicConfig.name(), topicMetadata);
+                    changelogTopicMetadata.put(topicConfig.name(), topicMetadata);
+                } else {
+                    log.debug("stream-thread [{}] No tasks found for topic group {}", streamThread.getName(),
topicGroupId);
+                }
             }
         }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/7d0f3e75/streams/src/test/java/org/apache/kafka/streams/processor/internals/SingleGroupPartitionGrouperStub.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/SingleGroupPartitionGrouperStub.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/SingleGroupPartitionGrouperStub.java
new file mode 100644
index 0000000..0da5974
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/SingleGroupPartitionGrouperStub.java
@@ -0,0 +1,47 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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;
+
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.streams.processor.DefaultPartitionGrouper;
+import org.apache.kafka.streams.processor.PartitionGrouper;
+import org.apache.kafka.streams.processor.TaskId;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Used for testing the assignment of a subset of a topology group, not the entire topology
+ */
+public class SingleGroupPartitionGrouperStub implements PartitionGrouper {
+    private PartitionGrouper defaultPartitionGrouper = new DefaultPartitionGrouper();
+
+    @Override
+    public Map<TaskId, Set<TopicPartition>> partitionGroups(Map<Integer, Set<String>>
topicGroups, Cluster metadata) {
+        Map<Integer, Set<String>> includedTopicGroups = new HashMap<>();
+
+        for (Map.Entry<Integer, Set<String>> entry : topicGroups.entrySet())
{
+            includedTopicGroups.put(entry.getKey(), entry.getValue());
+            break; // arbitrarily use the first entry only
+        }
+        Map<TaskId, Set<TopicPartition>> result = defaultPartitionGrouper.partitionGroups(includedTopicGroups,
metadata);
+        return result;
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/7d0f3e75/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java
index 65cc628..8e50356 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java
@@ -215,6 +215,46 @@ public class StreamPartitionAssignorTest {
     }
 
     @Test
+    public void testAssignWithPartialTopology() throws Exception {
+        Properties props = configProps();
+        props.put(StreamsConfig.PARTITION_GROUPER_CLASS_CONFIG, SingleGroupPartitionGrouperStub.class);
+        StreamsConfig config = new StreamsConfig(props);
+
+        TopologyBuilder builder = new TopologyBuilder();
+        builder.addSource("source1", "topic1");
+        builder.addProcessor("processor1", new MockProcessorSupplier(), "source1");
+        builder.addStateStore(new MockStateStoreSupplier("store1", false), "processor1");
+        builder.addSource("source2", "topic2");
+        builder.addProcessor("processor2", new MockProcessorSupplier(), "source2");
+        builder.addStateStore(new MockStateStoreSupplier("store2", false), "processor2");
+        List<String> topics = Utils.mkList("topic1", "topic2");
+        Set<TaskId> allTasks = Utils.mkSet(task0, task1, task2);
+
+        UUID uuid1 = UUID.randomUUID();
+        String client1 = "client1";
+
+        StreamThread thread10 = new StreamThread(builder, config, new MockClientSupplier(),
"test", client1, uuid1, new Metrics(), new SystemTime(), new StreamsMetadataState(builder));
+
+        StreamPartitionAssignor partitionAssignor = new StreamPartitionAssignor();
+        partitionAssignor.configure(config.getConsumerConfigs(thread10, "test", client1));
+        Map<String, PartitionAssignor.Subscription> subscriptions = new HashMap<>();
+        subscriptions.put("consumer10",
+            new PartitionAssignor.Subscription(topics, new SubscriptionInfo(uuid1, Collections.<TaskId>emptySet(),
Collections.<TaskId>emptySet(), userEndPoint).encode()));
+
+        // will throw exception if it fails
+        Map<String, PartitionAssignor.Assignment> assignments = partitionAssignor.assign(metadata,
subscriptions);
+
+        // check assignment info
+        Set<TaskId> allActiveTasks = new HashSet<>();
+        AssignmentInfo info10 = checkAssignment(Utils.mkSet("topic1"), assignments.get("consumer10"));
+        allActiveTasks.addAll(info10.activeTasks);
+
+        assertEquals(3, allActiveTasks.size());
+        assertEquals(allTasks, new HashSet<>(allActiveTasks));
+    }
+
+
+    @Test
     public void testAssignEmptyMetadata() throws Exception {
         StreamsConfig config = new StreamsConfig(configProps());
 


Mime
View raw message