Repository: skiptests/astraea
Pull request #409 (2022-06-14 16:22:25)

Implement consumer assign topic partition api
related to comment https://github.com/skiptests/astraea/pull/405#discussion_r897002521


Patch:
diff --git a/app/src/main/java/org/astraea/app/consumer/Builder.java b/app/src/main/java/org/astraea/app/consumer/Builder.java
index 3f7a67f7..428aa679 100644
--- a/app/src/main/java/org/astraea/app/consumer/Builder.java
+++ b/app/src/main/java/org/astraea/app/consumer/Builder.java
@@ -21,29 +21,19 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Set;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 
-public class Builder<Key, Value> {
-  private final Map<String, Object> configs =
-      new HashMap<>(
-          Map.of(ConsumerConfig.GROUP_ID_CONFIG, "groupId-" + System.currentTimeMillis()));
+abstract class Builder<Key, Value> {
+  private final Map<String, Object> configs = new HashMap<>();
   private Deserializer<?> keyDeserializer = Deserializer.BYTE_ARRAY;
   private Deserializer<?> valueDeserializer = Deserializer.BYTE_ARRAY;
-  private final Set<String> topics = new HashSet<>();
-  private ConsumerRebalanceListener listener = ignore -> {};
   private int distanceFromLatest = -1;
 
   Builder() {}
 
-  public Builder<Key, Value> groupId(String groupId) {
-    return config(ConsumerConfig.GROUP_ID_CONFIG, Objects.requireNonNull(groupId));
-  }
-
   /**
    * make the consumer read data from beginning. By default, it reads the latest data.
    *
@@ -74,11 +64,6 @@ public class Builder<Key, Value> {
     return this;
   }
 
-  public Builder<Key, Value> topics(Set<String> topics) {
-    this.topics.addAll(topics);
-    return this;
-  }
-
   @SuppressWarnings("unchecked")
   public <NewKey> Builder<NewKey, Value> keyDeserializer(Deserializer<NewKey> keyDeserializer) {
     this.keyDeserializer = Objects.requireNonNull(keyDeserializer);
@@ -107,15 +92,13 @@ public class Builder<Key, Value> {
         ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Objects.requireNonNull(bootstrapServers));
   }
 
-  public Builder<Key, Value> consumerRebalanceListener(ConsumerRebalanceListener listener) {
-    this.listener = Objects.requireNonNull(listener);
-    return this;
-  }
-
   public Builder<Key, Value> isolation(Isolation isolation) {
     return config(ConsumerConfig.ISOLATION_LEVEL_CONFIG, isolation.nameOfKafka());
   }
 
+  protected abstract void assignOrSubscribe(
+      org.apache.kafka.clients.consumer.Consumer<Key, Value> kafkaConsumer);
+
   @SuppressWarnings("unchecked")
   public Consumer<Key, Value> build() {
     var kafkaConsumer =
@@ -123,7 +106,8 @@ public class Builder<Key, Value> {
             configs,
             Deserializer.of((Deserializer<Key>) keyDeserializer),
             Deserializer.of((Deserializer<Value>) valueDeserializer));
-    kafkaConsumer.subscribe(topics, ConsumerRebalanceListener.of(listener));
+
+    assignOrSubscribe(kafkaConsumer);
 
     // this mode is not supported by kafka, so we have to calculate the offset first
     if (distanceFromLatest > 0) {
diff --git a/app/src/main/java/org/astraea/app/consumer/Consumer.java b/app/src/main/java/org/astraea/app/consumer/Consumer.java
index e893603c..593fe3e0 100644
--- a/app/src/main/java/org/astraea/app/consumer/Consumer.java
+++ b/app/src/main/java/org/astraea/app/consumer/Consumer.java
@@ -18,7 +18,8 @@ package org.astraea.app.consumer;
 
 import java.time.Duration;
 import java.util.Collection;
-import java.util.Map;
+import java.util.Set;
+import org.astraea.app.admin.TopicPartition;
 
 /** An interface for polling records. */
 public interface Consumer<Key, Value> extends AutoCloseable {
@@ -48,15 +49,23 @@ public interface Consumer<Key, Value> extends AutoCloseable {
   @Override
   void close();
 
-  static Builder<byte[], byte[]> builder() {
-    return new Builder<>();
-  }
-
-  static Consumer<byte[], byte[]> of(String bootstrapServers) {
-    return builder().bootstrapServers(bootstrapServers).build();
+  /**
+   * Create a consumer builder by setting specific topics
+   *
+   * @param topics set of topic names
+   * @return consumer builder for topics
+   */
+  static TopicsBuilder<byte[], byte[]> forTopics(Set<String> topics) {
+    return new TopicsBuilder<>(topics);
   }
 
-  static Consumer<byte[], byte[]> of(Map<String, String> configs) {
-    return builder().configs(configs).build();
+  /**
+   * Create a consumer builder by setting specific topic partitions
+   *
+   * @param partitions set of TopicPartition
+   * @return consumer builder for partitions
+   */
+  static PartitionsBuilder<byte[], byte[]> forPartitions(Set<TopicPartition> partitions) {
+    return new PartitionsBuilder<>(partitions);
   }
 }
diff --git a/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java b/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
new file mode 100644
index 00000000..75b8a4f5
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
@@ -0,0 +1,109 @@
+/*
+ * 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.astraea.app.consumer;
+
+import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.toList;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.astraea.app.admin.TopicPartition;
+
+public class PartitionsBuilder<Key, Value> extends Builder<Key, Value> {
+  private final Set<TopicPartition> partitions;
+
+  PartitionsBuilder(Set<TopicPartition> partitions) {
+    this.partitions = requireNonNull(partitions);
+  }
+
+  /**
+   * make the consumer read data from beginning. By default, it reads the latest data.
+   *
+   * @return this builder
+   */
+  @Override
+  public PartitionsBuilder<Key, Value> fromBeginning() {
+    super.fromBeginning();
+    return this;
+  }
+
+  /**
+   * make the consumer read data from latest. this is default setting.
+   *
+   * @return this builder
+   */
+  @Override
+  public PartitionsBuilder<Key, Value> fromLatest() {
+    super.fromLatest();
+    return this;
+  }
+
+  /**
+   * set the offset to read from the latest offset. For example, the end offset is 5, and you set
+   * distanceFromLatest to 2, then you will read data from offset: 3
+   *
+   * @param distanceFromLatest the distance from the latest offset
+   * @return this builder
+   */
+  @Override
+  public PartitionsBuilder<Key, Value> distanceFromLatest(int distanceFromLatest) {
+    super.distanceFromLatest(distanceFromLatest);
+    return this;
+  }
+
+  @Override
+  public <NewKey> PartitionsBuilder<NewKey, Value> keyDeserializer(
+      Deserializer<NewKey> keyDeserializer) {
+    return (PartitionsBuilder<NewKey, Value>) super.keyDeserializer(keyDeserializer);
+  }
+
+  @Override
+  public <NewValue> PartitionsBuilder<Key, NewValue> valueDeserializer(
+      Deserializer<NewValue> valueDeserializer) {
+    return (PartitionsBuilder<Key, NewValue>) super.valueDeserializer(valueDeserializer);
+  }
+
+  @Override
+  public PartitionsBuilder<Key, Value> config(String key, String value) {
+    super.config(key, value);
+    return this;
+  }
+
+  @Override
+  public PartitionsBuilder<Key, Value> configs(Map<String, String> configs) {
+    super.configs(configs);
+    return this;
+  }
+
+  @Override
+  public PartitionsBuilder<Key, Value> bootstrapServers(String bootstrapServers) {
+    super.bootstrapServers(bootstrapServers);
+    return this;
+  }
+
+  @Override
+  public PartitionsBuilder<Key, Value> isolation(Isolation isolation) {
+    super.isolation(isolation);
+    return this;
+  }
+
+  @Override
+  protected void assignOrSubscribe(Consumer<Key, Value> kafkaConsumer) {
+    kafkaConsumer.assign(partitions.stream().map(TopicPartition::to).collect(toList()));
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
new file mode 100644
index 00000000..712b2dd0
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
@@ -0,0 +1,121 @@
+/*
+ * 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.astraea.app.consumer;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+
+public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
+  private final Set<String> topics;
+  private ConsumerRebalanceListener listener = ignore -> {};
+
+  TopicsBuilder(Set<String> topics) {
+    this.topics = requireNonNull(topics);
+    config(ConsumerConfig.GROUP_ID_CONFIG, "groupId-" + System.currentTimeMillis());
+  }
+
+  public TopicsBuilder<Key, Value> groupId(String groupId) {
+    config(ConsumerConfig.GROUP_ID_CONFIG, requireNonNull(groupId));
+    return this;
+  }
+
+  public TopicsBuilder<Key, Value> consumerRebalanceListener(ConsumerRebalanceListener listener) {
+    this.listener = Objects.requireNonNull(listener);
+    return this;
+  }
+
+  /**
+   * make the consumer read data from beginning. By default, it reads the latest data.
+   *
+   * @return this builder
+   */
+  @Override
+  public TopicsBuilder<Key, Value> fromBeginning() {
+    super.fromBeginning();
+    return this;
+  }
+
+  /**
+   * make the consumer read data from latest. this is default setting.
+   *
+   * @return this builder
+   */
+  @Override
+  public TopicsBuilder<Key, Value> fromLatest() {
+    super.fromLatest();
+    return this;
+  }
+
+  /**
+   * set the offset to read from the latest offset. For example, the end offset is 5, and you set
+   * distanceFromLatest to 2, then you will read data from offset: 3
+   *
+   * @param distanceFromLatest the distance from the latest offset
+   * @return this builder
+   */
+  @Override
+  public TopicsBuilder<Key, Value> distanceFromLatest(int distanceFromLatest) {
+    super.distanceFromLatest(distanceFromLatest);
+    return this;
+  }
+
+  @Override
+  public <NewKey> TopicsBuilder<NewKey, Value> keyDeserializer(
+      Deserializer<NewKey> keyDeserializer) {
+    return (TopicsBuilder<NewKey, Value>) super.keyDeserializer(keyDeserializer);
+  }
+
+  @Override
+  public <NewValue> TopicsBuilder<Key, NewValue> valueDeserializer(
+      Deserializer<NewValue> valueDeserializer) {
+    return (TopicsBuilder<Key, NewValue>) super.valueDeserializer(valueDeserializer);
+  }
+
+  @Override
+  public TopicsBuilder<Key, Value> config(String key, String value) {
+    super.config(key, value);
+    return this;
+  }
+
+  @Override
+  public TopicsBuilder<Key, Value> configs(Map<String, String> configs) {
+    super.configs(configs);
+    return this;
+  }
+
+  @Override
+  public TopicsBuilder<Key, Value> bootstrapServers(String bootstrapServers) {
+    super.bootstrapServers(bootstrapServers);
+    return this;
+  }
+
+  @Override
+  public TopicsBuilder<Key, Value> isolation(Isolation isolation) {
+    super.isolation(isolation);
+    return this;
+  }
+
+  @Override
+  protected void assignOrSubscribe(Consumer<Key, Value> kafkaConsumer) {
+    kafkaConsumer.subscribe(topics, ConsumerRebalanceListener.of(listener));
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/performance/Performance.java b/app/src/main/java/org/astraea/app/performance/Performance.java
index 6021e900..7bdad597 100644
--- a/app/src/main/java/org/astraea/app/performance/Performance.java
+++ b/app/src/main/java/org/astraea/app/performance/Performance.java
@@ -179,9 +179,8 @@ public class Performance {
                     .mapToObj(
                         i ->
                             consumerExecutor(
-                                Consumer.builder()
+                                Consumer.forTopics(Set.of(param.topic))
                                     .bootstrapServers(param.bootstrapServers())
-                                    .topics(Set.of(param.topic))
                                     .groupId(groupId)
                                     .configs(param.configs())
                                     .isolation(param.isolation())


Test patch:
diff --git a/app/src/test/java/org/astraea/app/admin/AdminTest.java b/app/src/test/java/org/astraea/app/admin/AdminTest.java
index c12d0c52..1430fdd6 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -170,9 +170,8 @@ public class AdminTest extends RequireBrokerCluster {
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).numberOfPartitions(3).create();
       try (var c1 =
-          Consumer.builder()
+          Consumer.forTopics(Set.of(topicName))
               .bootstrapServers(bootstrapServers())
-              .topics(Set.of(topicName))
               .groupId(consumerGroup)
               .build()) {
         // wait for syncing topic creation
@@ -183,9 +182,8 @@ public class AdminTest extends RequireBrokerCluster {
         Assertions.assertEquals(consumerGroup, consumerGroupMap.get(consumerGroup).groupId());
 
         try (var c2 =
-            Consumer.builder()
+            Consumer.forTopics(Set.of(topicName))
                 .bootstrapServers(bootstrapServers())
-                .topics(Set.of(topicName))
                 .groupId("abc")
                 .build()) {
           var count =
@@ -319,12 +317,11 @@ public class AdminTest extends RequireBrokerCluster {
       TimeUnit.SECONDS.sleep(3);
 
       try (var consumer =
-          Consumer.builder()
+          Consumer.forTopics(Set.of(topicName))
               .keyDeserializer(Deserializer.STRING)
               .valueDeserializer(Deserializer.STRING)
               .fromBeginning()
               .bootstrapServers(bootstrapServers())
-              .topics(Set.of(topicName))
               .build()) {
 
         var records =
diff --git a/app/src/test/java/org/astraea/app/admin/TopicExplorerTest.java b/app/src/test/java/org/astraea/app/admin/TopicExplorerTest.java
index bdb48aeb..eb5255d7 100644
--- a/app/src/test/java/org/astraea/app/admin/TopicExplorerTest.java
+++ b/app/src/test/java/org/astraea/app/admin/TopicExplorerTest.java
@@ -72,9 +72,8 @@ public class TopicExplorerTest extends RequireBrokerCluster {
             new Thread(
                     () -> {
                       var consumer =
-                          Consumer.builder()
+                          Consumer.forTopics(topics)
                               .bootstrapServers(bootstrapServers())
-                              .topics(topics)
                               .groupId(groupName)
                               .configs(
                                   Map.of(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, instanceName))
diff --git a/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java b/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
index cb22ffe6..4cbd8e75 100644
--- a/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
+++ b/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
@@ -16,13 +16,20 @@
  */
 package org.astraea.app.consumer;
 
+import static java.util.Collections.nCopies;
+import static java.util.stream.Collectors.toList;
+
+import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Set;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.IntStream;
+import java.util.stream.Stream;
 import org.apache.kafka.common.errors.WakeupException;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.common.Utils;
 import org.astraea.app.producer.Producer;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -52,8 +59,7 @@ public class ConsumerTest extends RequireBrokerCluster {
     var topic = "testPoll";
     produceData(topic, recordCount);
     try (var consumer =
-        Consumer.builder()
-            .topics(Set.of(topic))
+        Consumer.forTopics(Set.of(topic))
             .bootstrapServers(bootstrapServers())
             .fromBeginning()
             .build()) {
@@ -68,8 +74,7 @@ public class ConsumerTest extends RequireBrokerCluster {
     var topic = "testFromLatest";
     produceData(topic, 1);
     try (var consumer =
-        Consumer.builder()
-            .topics(Set.of(topic))
+        Consumer.forTopics(Set.of(topic))
             .bootstrapServers(bootstrapServers())
             .fromLatest()
             .build()) {
@@ -83,8 +88,7 @@ public class ConsumerTest extends RequireBrokerCluster {
   void testWakeup() throws InterruptedException {
     var topic = "testWakeup";
     try (var consumer =
-        Consumer.builder()
-            .topics(Set.of(topic))
+        Consumer.forTopics(Set.of(topic))
             .bootstrapServers(bootstrapServers())
             .fromLatest()
             .build()) {
@@ -115,8 +119,7 @@ public class ConsumerTest extends RequireBrokerCluster {
     java.util.function.BiConsumer<String, Integer> testConsumer =
         (id, expectedSize) -> {
           try (var consumer =
-              Consumer.builder()
-                  .topics(Set.of(topic))
+              Consumer.forTopics(Set.of(topic))
                   .bootstrapServers(bootstrapServers())
                   .fromBeginning()
                   .groupId(id)
@@ -151,18 +154,16 @@ public class ConsumerTest extends RequireBrokerCluster {
       producer.flush();
     }
     try (var consumer =
-        Consumer.builder()
+        Consumer.forTopics(Set.of(topic))
             .bootstrapServers(bootstrapServers())
-            .topics(Set.of(topic))
             .distanceFromLatest(3)
             .build()) {
       Assertions.assertEquals(3, consumer.poll(4, Duration.ofSeconds(5)).size());
     }
 
     try (var consumer =
-        Consumer.builder()
+        Consumer.forTopics(Set.of(topic))
             .bootstrapServers(bootstrapServers())
-            .topics(Set.of(topic))
             .distanceFromLatest(1000)
             .build()) {
       Assertions.assertEquals(10, consumer.poll(11, Duration.ofSeconds(5)).size());
@@ -174,9 +175,8 @@ public class ConsumerTest extends RequireBrokerCluster {
     var count = 1;
     var topic = "testPollingTime";
     try (var consumer =
-        Consumer.builder()
+        Consumer.forTopics(Set.of(topic))
             .bootstrapServers(bootstrapServers())
-            .topics(Set.of(topic))
             .fromBeginning()
             .build()) {
 
@@ -186,4 +186,50 @@ public class ConsumerTest extends RequireBrokerCluster {
           Duration.ofSeconds(10), () -> consumer.poll(Duration.ofSeconds(Integer.MAX_VALUE)));
     }
   }
+
+  @Test
+  void testAssignment() throws InterruptedException {
+    var topic = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer = Producer.of(bootstrapServers())) {
+      var partitionNum = 2;
+      admin.creator().topic(topic).numberOfPartitions(partitionNum).create();
+      TimeUnit.SECONDS.sleep(2);
+
+      for (int partitionId = 0; partitionId < partitionNum; partitionId++) {
+        for (int recordIdx = 0; recordIdx < 10; recordIdx++) {
+          producer
+              .sender()
+              .topic(topic)
+              .partition(partitionId)
+              .value(ByteBuffer.allocate(4).putInt(recordIdx).array())
+              .run();
+        }
+      }
+      producer.flush();
+    }
+
+    try (var consumer =
+        Consumer.forPartitions(Set.of(TopicPartition.of(topic, "1")))
+            .bootstrapServers(bootstrapServers())
+            .distanceFromLatest(20)
+            .build()) {
+      var records = consumer.poll(20, Duration.ofSeconds(5));
+      Assertions.assertEquals(10, records.size());
+      Assertions.assertEquals(
+          nCopies(10, 1), records.stream().map(Record::partition).collect(toList()));
+    }
+
+    try (var consumer =
+        Consumer.forPartitions(Set.of(TopicPartition.of(topic, "0"), TopicPartition.of(topic, "1")))
+            .bootstrapServers(bootstrapServers())
+            .distanceFromLatest(20)
+            .build()) {
+      var records = consumer.poll(20, Duration.ofSeconds(5));
+      Assertions.assertEquals(20, records.size());
+      Assertions.assertEquals(
+          Stream.concat(nCopies(10, 0).stream(), nCopies(10, 1).stream()).collect(toList()),
+          records.stream().map(Record::partition).sorted().collect(toList()));
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/consumer/RebalanceListenerTest.java b/app/src/test/java/org/astraea/app/consumer/RebalanceListenerTest.java
index 342d14ad..7c0c1eb3 100644
--- a/app/src/test/java/org/astraea/app/consumer/RebalanceListenerTest.java
+++ b/app/src/test/java/org/astraea/app/consumer/RebalanceListenerTest.java
@@ -31,9 +31,8 @@ public class RebalanceListenerTest extends RequireBrokerCluster {
     var getAssignment = new AtomicInteger(0);
     var topicName = "testRebalanceListener-" + System.currentTimeMillis();
     try (var consumer =
-        Consumer.builder()
+        Consumer.forTopics(Set.of(topicName))
             .bootstrapServers(bootstrapServers())
-            .topics(Set.of(topicName))
             .consumerRebalanceListener(ignore -> getAssignment.incrementAndGet())
             .build()) {
       try (var threadPool =
diff --git a/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java b/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
index 565f3e78..eab97352 100644
--- a/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
@@ -120,10 +120,9 @@ public class SmoothWeightRoundRobinDispatchTest extends RequireBrokerCluster {
     }
     sleep(1);
     try (var consumer =
-        Consumer.builder()
+        Consumer.forTopics(Set.of(topicName))
             .bootstrapServers(bootstrapServers())
             .fromBeginning()
-            .topics(Set.of(topicName))
             .keyDeserializer(Deserializer.STRING)
             .build()) {
       var records = consumer.poll(Duration.ofSeconds(20));
@@ -174,10 +173,9 @@ public class SmoothWeightRoundRobinDispatchTest extends RequireBrokerCluster {
       threadPool.waitAll();
     }
     try (var consumer =
-        Consumer.builder()
+        Consumer.forTopics(Set.of(topicName))
             .bootstrapServers(bootstrapServers())
             .fromBeginning()
-            .topics(Set.of(topicName))
             .keyDeserializer(Deserializer.STRING)
             .build()) {
       var records = consumer.poll(Duration.ofSeconds(20));
diff --git a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
index 30222b51..1b2efb92 100644
--- a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
@@ -102,10 +102,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     param.sizeDistributionType = DistributionType.FIXED;
     try (Executor executor =
         Performance.consumerExecutor(
-            Consumer.builder()
-                .topics(Set.of(topicName))
-                .bootstrapServers(bootstrapServers())
-                .build(),
+            Consumer.forTopics(Set.of(topicName)).bootstrapServers(bootstrapServers()).build(),
             metrics,
             new Manager(param, List.of(), List.of()),
             () -> false)) {
diff --git a/app/src/test/java/org/astraea/app/producer/ProducerTest.java b/app/src/test/java/org/astraea/app/producer/ProducerTest.java
index 5c5d84a7..fe5566df 100644
--- a/app/src/test/java/org/astraea/app/producer/ProducerTest.java
+++ b/app/src/test/java/org/astraea/app/producer/ProducerTest.java
@@ -68,10 +68,9 @@ public class ProducerTest extends RequireBrokerCluster {
     }
 
     try (var consumer =
-        Consumer.builder()
+        Consumer.forTopics(Set.of(topicName))
             .bootstrapServers(bootstrapServers())
             .fromBeginning()
-            .topics(Set.of(topicName))
             .keyDeserializer(Deserializer.STRING)
             .build()) {
       var records = consumer.poll(Duration.ofSeconds(10));
@@ -112,10 +111,9 @@ public class ProducerTest extends RequireBrokerCluster {
     }
 
     try (var consumer =
-        Consumer.builder()
+        Consumer.forTopics(Set.of(topicName))
             .bootstrapServers(bootstrapServers())
             .fromBeginning()
-            .topics(Set.of(topicName))
             .keyDeserializer(Deserializer.STRING)
             .isolation(Isolation.READ_COMMITTED)
             .build()) {
@@ -144,10 +142,9 @@ public class ProducerTest extends RequireBrokerCluster {
     producer.sender().topic(topic).value(new byte[10]).run().toCompletableFuture().get();
 
     try (var consumer =
-        Consumer.builder()
+        Consumer.forTopics(Set.of(topic))
             .bootstrapServers(bootstrapServers())
             .fromBeginning()
-            .topics(Set.of(topic))
             .isolation(
                 producer.transactional() ? Isolation.READ_COMMITTED : Isolation.READ_UNCOMMITTED)
             .build()) {
@@ -171,10 +168,9 @@ public class ProducerTest extends RequireBrokerCluster {
     latch.await();
 
     try (var consumer =
-        Consumer.builder()
+        Consumer.forTopics(Set.of(topic))
             .bootstrapServers(bootstrapServers())
             .fromBeginning()
-            .topics(Set.of(topic))
             .isolation(
                 producer.transactional() ? Isolation.READ_COMMITTED : Isolation.READ_UNCOMMITTED)
             .build()) {
diff --git a/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java b/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java
index 70233b35..75333670 100644
--- a/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java
@@ -40,9 +40,8 @@ public class GroupHandlerTest extends RequireBrokerCluster {
       TimeUnit.SECONDS.sleep(3);
 
       try (var consumer =
-          Consumer.builder()
+          Consumer.forTopics(Set.of(topicName))
               .groupId(groupId)
-              .topics(Set.of(topicName))
               .bootstrapServers(bootstrapServers())
               .build()) {
         Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());
@@ -80,9 +79,8 @@ public class GroupHandlerTest extends RequireBrokerCluster {
       var handler = new GroupHandler(admin);
 
       try (var consumer =
-          Consumer.builder()
+          Consumer.forTopics(Set.of(topicName))
               .groupId(groupId)
-              .topics(Set.of(topicName))
               .bootstrapServers(bootstrapServers())
               .build()) {
         Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());
@@ -103,9 +101,8 @@ public class GroupHandlerTest extends RequireBrokerCluster {
       var handler = new GroupHandler(admin);
 
       try (var consumer =
-          Consumer.builder()
+          Consumer.forTopics(Set.of(topicName))
               .groupId(groupId)
-              .topics(Set.of(topicName))
               .bootstrapServers(bootstrapServers())
               .build()) {
         Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());