kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject [kafka] branch 2.5 updated: KAFKA-9523: Migrate BranchedMultiLevelRepartitionConnectedTopologyTest into a unit test (#8081)
Date Mon, 10 Feb 2020 21:10:34 GMT
This is an automated email from the ASF dual-hosted git repository.

guozhang pushed a commit to branch 2.5
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/2.5 by this push:
     new 3865f63  KAFKA-9523: Migrate BranchedMultiLevelRepartitionConnectedTopologyTest into
a unit test (#8081)
3865f63 is described below

commit 3865f636c89ba9c063bf707e358223b5e7c3629f
Author: Boyang Chen <boyang@confluent.io>
AuthorDate: Mon Feb 10 13:09:26 2020 -0800

    KAFKA-9523: Migrate BranchedMultiLevelRepartitionConnectedTopologyTest into a unit test
(#8081)
    
    Relying on integration test to catch an algorithm bug introduces more flakiness, reduce
the test into a unit test to reduce the flakiness until we upgrade Java/Scala libs.
    
    Checked the test shall fail with older version of StreamsPartitionAssignor.
    
    Reviewers: Guozhang Wang <wangguoz@gmail.com>
---
 ...MultiLevelRepartitionConnectedTopologyTest.java | 151 ---------------------
 .../internals/StreamsPartitionAssignorTest.java    |  86 ++++++++++++
 2 files changed, 86 insertions(+), 151 deletions(-)

diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/BranchedMultiLevelRepartitionConnectedTopologyTest.java
b/streams/src/test/java/org/apache/kafka/streams/integration/BranchedMultiLevelRepartitionConnectedTopologyTest.java
deleted file mode 100644
index cfff070..0000000
--- a/streams/src/test/java/org/apache/kafka/streams/integration/BranchedMultiLevelRepartitionConnectedTopologyTest.java
+++ /dev/null
@@ -1,151 +0,0 @@
-/*
- * 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.integration;
-
-import kafka.utils.MockTime;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.apache.kafka.common.serialization.Serdes;
-import org.apache.kafka.streams.KafkaStreams;
-import org.apache.kafka.streams.KeyValue;
-import org.apache.kafka.streams.StreamsBuilder;
-import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.Topology;
-import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
-import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
-import org.apache.kafka.streams.kstream.KStream;
-import org.apache.kafka.streams.kstream.KTable;
-import org.apache.kafka.streams.kstream.Materialized;
-import org.apache.kafka.test.IntegrationTest;
-import org.apache.kafka.test.StreamsTestUtils;
-import org.apache.kafka.test.TestUtils;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Properties;
-import java.util.concurrent.ExecutionException;
-
-/**
- * Test out a topology with 3 level of sub-topology as:
- *           0
- *         /   \
- *        1    3
- *         \  /
- *          2
- * where each pair of the sub topology is connected by repartition topic.
- * The purpose of this test is to verify the robustness of the stream partition assignor
algorithm,
- * especially whether it could build the repartition topic counts (step zero) with a complex
topology.
- * The traversal path 0 -> 1 -> 2 -> 3 hits the case where sub-topology 2 will be
initialized while its
- * parent 3 hasn't been initialized yet.
- */
-@Category({IntegrationTest.class})
-public class BranchedMultiLevelRepartitionConnectedTopologyTest {
-
-    private static final Logger log = LoggerFactory.getLogger(BranchedMultiLevelRepartitionConnectedTopologyTest.class);
-
-    private static String inputStream;
-
-    private KafkaStreams kafkaStreams;
-
-    private Properties streamsConfiguration;
-
-    @ClassRule
-    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1);
-    private final MockTime mockTime = CLUSTER.time;
-
-    @Before
-    public void setUp() throws Exception {
-        final Properties props = new Properties();
-        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-        props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION, StreamsConfig.NO_OPTIMIZATION);
-        streamsConfiguration = StreamsTestUtils.getStreamsConfig(
-            "branched-repartition-topic-test",
-            CLUSTER.bootstrapServers(),
-            Serdes.ByteArray().getClass().getName(),
-            Serdes.ByteArray().getClass().getName(),
-            props);
-
-        inputStream = "input-stream";
-        CLUSTER.createTopic(inputStream, 3, 1);
-
-        IntegrationTestUtils.purgeLocalStreamsState(streamsConfiguration);
-    }
-
-    @Test
-    @SuppressWarnings("unchecked")
-    public void testTopologyBuild() throws InterruptedException, ExecutionException {
-
-        final StreamsBuilder builder = new StreamsBuilder();
-
-        final KStream<byte[], byte[]> input = builder.stream(inputStream);
-
-        final KStream<byte[], byte[]>[] branches = input
-            .flatMapValues(value -> Collections.singletonList(new byte[0]))
-            .branch((k, v) -> true, (k, v) -> false);
-
-        final KTable<byte[], byte[]> b1 = branches[0]
-            .map(KeyValue::new)
-            .groupByKey()
-            .reduce((k, v) -> v, Materialized.as("odd_store"))
-            .toStream()
-            .peek((k, v) -> { })
-            .map(KeyValue::new)
-            .groupByKey()
-            .reduce((k, v) -> v, Materialized.as("odd_store_2"));
-
-        final KTable<byte[], byte[]> b2 = branches[1]
-            .map(KeyValue::new)
-            .groupByKey()
-            .reduce((k, v) -> v, Materialized.as("even_store"))
-            .toStream()
-            .peek((k, v) -> { })
-            .map(KeyValue::new)
-            .groupByKey()
-            .reduce((k, v) -> v, Materialized.as("even_store_2"));
-
-        b1.join(b2, (v1, v2) -> v1, Materialized.as("joined_store"))
-            .toStream();
-
-        final Topology topology = builder.build(streamsConfiguration);
-        log.info("Built topology: {}", topology.describe());
-
-        final Properties producerConfig = TestUtils.producerConfig(
-            CLUSTER.bootstrapServers(), ByteArraySerializer.class, ByteArraySerializer.class);
-
-        final List<KeyValue<byte[], byte[]>> initialKeyValues = Collections.singletonList(
-            KeyValue.pair(new byte[1], new byte[1]));
-
-        IntegrationTestUtils.produceKeyValuesSynchronously(
-            inputStream, initialKeyValues, producerConfig, mockTime);
-
-        kafkaStreams = new KafkaStreams(topology, streamsConfiguration);
-
-        kafkaStreams.cleanUp();
-        kafkaStreams.start();
-
-        TestUtils.waitForCondition(() -> kafkaStreams.state() == KafkaStreams.State.RUNNING,
-                                   "Failed to observe stream transits to RUNNING");
-
-        kafkaStreams.close();
-    }
-}
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
index ffc422e..12fc9c7 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
@@ -1692,6 +1692,92 @@ public class StreamsPartitionAssignorTest {
         shouldThrowIfPreVersionProbingSubscriptionAndFutureSubscriptionIsMixed(2);
     }
 
+    @Test
+    public void shouldNotFailOnBranchedMultiLevelRepartitionConnectedTopology() {
+        // Test out a topology with 3 level of sub-topology as:
+        //            0
+        //          /   \
+        //         1    3
+        //          \  /
+        //           2
+        //  where each pair of the sub topology is connected by repartition topic.
+        //  The purpose of this test is to verify the robustness of the stream partition
assignor algorithm,
+        //  especially whether it could build the repartition topic counts (step zero) with
a complex topology.
+        //  The traversal path 0 -> 1 -> 2 -> 3 hits the case where sub-topology
2 will be initialized while its
+        //  parent 3 hasn't been initialized yet.
+        final String applicationId = "test";
+        builder.setApplicationId(applicationId);
+        builder.addSource(null, "KSTREAM-SOURCE-0000000000",  null, null, null, "input-stream");
+        builder.addProcessor("KSTREAM-FLATMAPVALUES-0000000001", new MockProcessorSupplier(),
"KSTREAM-SOURCE-0000000000");
+        builder.addProcessor("KSTREAM-BRANCH-0000000002", new MockProcessorSupplier(), "KSTREAM-FLATMAPVALUES-0000000001");
+        builder.addProcessor("KSTREAM-BRANCHCHILD-0000000003", new MockProcessorSupplier(),
"KSTREAM-BRANCH-0000000002");
+        builder.addProcessor("KSTREAM-BRANCHCHILD-0000000004", new MockProcessorSupplier(),
"KSTREAM-BRANCH-0000000002");
+        builder.addProcessor("KSTREAM-MAP-0000000005", new MockProcessorSupplier(), "KSTREAM-BRANCHCHILD-0000000003");
+
+        builder.addInternalTopic("odd_store-repartition");
+        builder.addProcessor("odd_store-repartition-filter", new MockProcessorSupplier(),
"KSTREAM-MAP-0000000005");
+        builder.addSink("odd_store-repartition-sink", "odd_store-repartition", null, null,
null, "odd_store-repartition-filter");
+        builder.addSource(null, "odd_store-repartition-source", null, null, null, "odd_store-repartition");
+        builder.addProcessor("KSTREAM-REDUCE-0000000006", new MockProcessorSupplier(), "odd_store-repartition-source");
+        builder.addProcessor("KTABLE-TOSTREAM-0000000010", new MockProcessorSupplier(), "KSTREAM-REDUCE-0000000006");
+        builder.addProcessor("KSTREAM-PEEK-0000000011", new MockProcessorSupplier(), "KTABLE-TOSTREAM-0000000010");
+        builder.addProcessor("KSTREAM-MAP-0000000012", new MockProcessorSupplier(), "KSTREAM-PEEK-0000000011");
+
+        builder.addInternalTopic("odd_store_2-repartition");
+        builder.addProcessor("odd_store_2-repartition-filter", new MockProcessorSupplier(),
"KSTREAM-MAP-0000000012");
+        builder.addSink("odd_store_2-repartition-sink", "odd_store_2-repartition", null,
null, null, "odd_store_2-repartition-filter");
+        builder.addSource(null, "odd_store_2-repartition-source", null, null, null, "odd_store_2-repartition");
+        builder.addProcessor("KSTREAM-REDUCE-0000000013", new MockProcessorSupplier(), "odd_store_2-repartition-source");
+        builder.addProcessor("KSTREAM-MAP-0000000017", new MockProcessorSupplier(), "KSTREAM-BRANCHCHILD-0000000004");
+
+        builder.addInternalTopic("even_store-repartition");
+        builder.addProcessor("even_store-repartition-filter", new MockProcessorSupplier(),
"KSTREAM-MAP-0000000017");
+        builder.addSink("even_store-repartition-sink", "even_store-repartition", null, null,
null, "even_store-repartition-filter");
+        builder.addSource(null, "even_store-repartition-source", null, null, null, "even_store-repartition");
+        builder.addProcessor("KSTREAM-REDUCE-0000000018", new MockProcessorSupplier(), "even_store-repartition-source");
+        builder.addProcessor("KTABLE-TOSTREAM-0000000022", new MockProcessorSupplier(), "KSTREAM-REDUCE-0000000018");
+        builder.addProcessor("KSTREAM-PEEK-0000000023", new MockProcessorSupplier(), "KTABLE-TOSTREAM-0000000022");
+        builder.addProcessor("KSTREAM-MAP-0000000024", new MockProcessorSupplier(), "KSTREAM-PEEK-0000000023");
+
+        builder.addInternalTopic("even_store_2-repartition");
+        builder.addProcessor("even_store_2-repartition-filter", new MockProcessorSupplier(),
"KSTREAM-MAP-0000000024");
+        builder.addSink("even_store_2-repartition-sink", "even_store_2-repartition", null,
null, null, "even_store_2-repartition-filter");
+        builder.addSource(null, "even_store_2-repartition-source", null, null, null, "even_store_2-repartition");
+        builder.addProcessor("KSTREAM-REDUCE-0000000025", new MockProcessorSupplier(), "even_store_2-repartition-source");
+        builder.addProcessor("KTABLE-JOINTHIS-0000000030", new MockProcessorSupplier(), "KSTREAM-REDUCE-0000000013");
+        builder.addProcessor("KTABLE-JOINOTHER-0000000031", new MockProcessorSupplier(),
"KSTREAM-REDUCE-0000000025");
+        builder.addProcessor("KTABLE-MERGE-0000000029", new MockProcessorSupplier(), "KTABLE-JOINTHIS-0000000030",
"KTABLE-JOINOTHER-0000000031");
+        builder.addProcessor("KTABLE-TOSTREAM-0000000032", new MockProcessorSupplier(), "KTABLE-MERGE-0000000029");
+
+        final List<String> topics = asList("input-stream", "test-even_store-repartition",
"test-even_store_2-repartition", "test-odd_store-repartition", "test-odd_store_2-repartition");
+
+        final UUID uuid1 = UUID.randomUUID();
+        createMockTaskManager(emptyTasks, emptyTasks, uuid1, builder);
+        EasyMock.replay(taskManager);
+
+        streamsMetadataState = EasyMock.createNiceMock(StreamsMetadataState.class);
+        configurePartitionAssignor(emptyMap());
+        final MockInternalTopicManager internalTopicManager =
+            new MockInternalTopicManager(streamsConfig, mockClientSupplier.restoreConsumer);
+        partitionAssignor.setInternalTopicManager(internalTopicManager);
+
+        subscriptions.put("consumer10",
+            new ConsumerPartitionAssignor.Subscription(
+                topics,
+                getInfo(uuid1, emptyTasks, emptyTasks).encode())
+        );
+
+        final Cluster metadata = new Cluster(
+            "cluster",
+            Collections.singletonList(Node.noNode()),
+            Collections.singletonList(new PartitionInfo("input-stream", 0, Node.noNode(),
new Node[0], new Node[0])),
+            Collections.emptySet(),
+            Collections.emptySet());
+
+        // This shall fail if we have bugs in the repartition topic creation due to the inconsistent
order of sub-topologies.
+        partitionAssignor.assign(metadata, new GroupSubscription(subscriptions));
+    }
+
     private static ByteBuffer encodeFutureSubscription() {
         final ByteBuffer buf = ByteBuffer.allocate(4 /* used version */ + 4 /* supported
version */);
         buf.putInt(LATEST_SUPPORTED_VERSION + 1);


Mime
View raw message