Repository: skiptests/astraea
Pull request #317 (2022-05-04 07:54:39)

Add Isolation and Compression to simplify builder
一些重構，包含：
1) 新增`Compression` and `Isolation` 來避免直接使用kafka參數
2) 重構一些較長的函示


Patch:
diff --git a/app/src/main/java/org/astraea/argument/CompressionField.java b/app/src/main/java/org/astraea/argument/CompressionField.java
index 4e937574..9f02ebbe 100644
--- a/app/src/main/java/org/astraea/argument/CompressionField.java
+++ b/app/src/main/java/org/astraea/argument/CompressionField.java
@@ -4,17 +4,18 @@ import com.beust.jcommander.ParameterException;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.kafka.common.record.CompressionType;
+import org.astraea.topic.Compression;
 
-public class CompressionField extends Field<CompressionType> {
+public class CompressionField extends Field<Compression> {
   /**
    * @param value Name of compression type. Accept lower-case name only ("none", "gzip", "snappy",
    *     "lz4", "zstd").
    */
   @Override
-  public CompressionType convert(String value) {
+  public Compression convert(String value) {
     try {
       // `CompressionType#forName` accept lower-case name only.
-      return CompressionType.forName(value);
+      return Compression.of(value);
     } catch (IllegalArgumentException e) {
       throw new ParameterException(
           "the "
diff --git a/app/src/main/java/org/astraea/consumer/Builder.java b/app/src/main/java/org/astraea/consumer/Builder.java
index c8a579a8..d62b1ca8 100644
--- a/app/src/main/java/org/astraea/consumer/Builder.java
+++ b/app/src/main/java/org/astraea/consumer/Builder.java
@@ -13,25 +13,18 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 
 public class Builder<Key, Value> {
-
-  public enum OffsetPolicy {
-    EARLIEST,
-    LATEST
-  }
-
-  private final Map<String, Object> configs = new HashMap<>();
+  private final Map<String, Object> configs =
+      new HashMap<>(
+          Map.of(ConsumerConfig.GROUP_ID_CONFIG, "groupId-" + System.currentTimeMillis()));
   private Deserializer<?> keyDeserializer = Deserializer.BYTE_ARRAY;
   private Deserializer<?> valueDeserializer = Deserializer.BYTE_ARRAY;
-  private OffsetPolicy offsetPolicy = OffsetPolicy.LATEST;
-  private String groupId = "groupId-" + System.currentTimeMillis();
   private final Set<String> topics = new HashSet<>();
   private ConsumerRebalanceListener listener = ignore -> {};
 
   Builder() {}
 
   public Builder<Key, Value> groupId(String groupId) {
-    this.groupId = Objects.requireNonNull(groupId);
-    return this;
+    return config(ConsumerConfig.GROUP_ID_CONFIG, Objects.requireNonNull(groupId));
   }
 
   /**
@@ -40,8 +33,7 @@ public class Builder<Key, Value> {
    * @return this builder
    */
   public Builder<Key, Value> fromBeginning() {
-    this.offsetPolicy = OffsetPolicy.EARLIEST;
-    return this;
+    return config(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
   }
 
   public Builder<Key, Value> topics(Set<String> topics) {
@@ -62,14 +54,18 @@ public class Builder<Key, Value> {
     return (Builder<Key, NewValue>) this;
   }
 
-  public Builder<Key, Value> configs(Map<String, Object> configs) {
+  public Builder<Key, Value> config(String key, String value) {
+    this.configs.put(key, value);
+    return this;
+  }
+
+  public Builder<Key, Value> configs(Map<String, String> configs) {
     this.configs.putAll(configs);
     return this;
   }
 
   public Builder<Key, Value> brokers(String brokers) {
-    this.configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Objects.requireNonNull(brokers));
-    return this;
+    return config(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Objects.requireNonNull(brokers));
   }
 
   public Builder<Key, Value> consumerRebalanceListener(ConsumerRebalanceListener listener) {
@@ -77,17 +73,12 @@ public class Builder<Key, Value> {
     return this;
   }
 
+  public Builder<Key, Value> isolation(Isolation isolation) {
+    return config(ConsumerConfig.ISOLATION_LEVEL_CONFIG, isolation.nameOfKafka());
+  }
+
   @SuppressWarnings("unchecked")
   public Consumer<Key, Value> build() {
-    configs.put(ConsumerConfig.GROUP_ID_CONFIG, Objects.requireNonNull(groupId));
-    switch (offsetPolicy) {
-      case EARLIEST:
-        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-        break;
-      case LATEST:
-        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
-        break;
-    }
     var kafkaConsumer =
         new KafkaConsumer<>(
             configs,
diff --git a/app/src/main/java/org/astraea/consumer/Consumer.java b/app/src/main/java/org/astraea/consumer/Consumer.java
index a4d2268e..3e936217 100644
--- a/app/src/main/java/org/astraea/consumer/Consumer.java
+++ b/app/src/main/java/org/astraea/consumer/Consumer.java
@@ -29,7 +29,7 @@ public interface Consumer<Key, Value> extends AutoCloseable {
     return builder().brokers(brokers).build();
   }
 
-  static Consumer<byte[], byte[]> of(Map<String, Object> configs) {
+  static Consumer<byte[], byte[]> of(Map<String, String> configs) {
     return builder().configs(configs).build();
   }
 }
diff --git a/app/src/main/java/org/astraea/consumer/Isolation.java b/app/src/main/java/org/astraea/consumer/Isolation.java
new file mode 100644
index 00000000..63b91501
--- /dev/null
+++ b/app/src/main/java/org/astraea/consumer/Isolation.java
@@ -0,0 +1,17 @@
+package org.astraea.consumer;
+
+import java.util.Locale;
+
+public enum Isolation {
+  READ_UNCOMMITTED,
+  READ_COMMITTED;
+
+  public static Isolation of(String name) {
+    return Isolation.valueOf(name.toUpperCase(Locale.ROOT));
+  }
+
+  /** @return the name parsed by kafka */
+  public String nameOfKafka() {
+    return name().toLowerCase(Locale.ROOT);
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index e99671f8..95df6ec3 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -15,12 +15,9 @@ import java.util.concurrent.ExecutionException;
 import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
-import org.apache.kafka.common.record.CompressionType;
 import org.astraea.Utils;
 import org.astraea.argument.CompressionField;
 import org.astraea.argument.NonEmptyStringField;
@@ -33,8 +30,10 @@ import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.State;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
+import org.astraea.consumer.Isolation;
 import org.astraea.producer.Producer;
 import org.astraea.producer.TransactionalProducer;
+import org.astraea.topic.Compression;
 import org.astraea.topic.TopicAdmin;
 import org.astraea.utils.DataSize;
 import org.astraea.utils.DataUnit;
@@ -115,7 +114,8 @@ public class Performance {
                                     .brokers(param.brokers)
                                     .topics(Set.of(param.topic))
                                     .groupId(groupId)
-                                    .configs(param.consumerProps())
+                                    .configs(param.configs)
+                                    .isolation(param.isolation())
                                     .consumerRebalanceListener(
                                         ignore -> manager.countDownGetAssignment())
                                     .build(),
@@ -128,11 +128,13 @@ public class Performance {
                         i ->
                             producerExecutor(
                                 Producer.builder()
-                                    .configs(param.producerProps())
+                                    .configs(param.configs)
+                                    .compression(param.compression)
                                     .partitionClassName(param.partitioner)
                                     .build(),
                                 Producer.builder()
-                                    .configs(param.producerProps())
+                                    .configs(param.configs)
+                                    .compression(param.compression)
                                     .partitionClassName(param.partitioner)
                                     .buildTransactional(),
                                 param,
@@ -192,57 +194,71 @@ public class Performance {
       List<Integer> partitions,
       Manager manager) {
     return new Executor() {
+
+      private State doTransaction() {
+        var rand = new Random();
+        var senders =
+            IntStream.range(0, param.transactionSize)
+                .mapToObj(i -> manager.payload())
+                .filter(Optional::isPresent)
+                .map(
+                    p ->
+                        producer
+                            .sender()
+                            .topic(param.topic)
+                            .partition(partitions.get(rand.nextInt(partitions.size())))
+                            .key(manager.getKey())
+                            .value(p.get())
+                            .timestamp(System.currentTimeMillis()))
+                .collect(Collectors.toList());
+
+        // No records to send
+        if (senders.isEmpty()) return State.DONE;
+        transactionalProducer
+            .transaction(senders)
+            .forEach(
+                future ->
+                    future.whenComplete(
+                        (m, e) ->
+                            observer.accept(
+                                System.currentTimeMillis() - m.timestamp(),
+                                m.serializedValueSize())));
+        return State.RUNNING;
+      }
+
+      private State doSend() {
+        var rand = new Random();
+        var payload = manager.payload();
+        if (payload.isEmpty()) return State.DONE;
+
+        long start = System.currentTimeMillis();
+        producer
+            .sender()
+            .topic(param.topic)
+            .partition(partitions.get(rand.nextInt(partitions.size())))
+            .key(manager.getKey())
+            .value(payload.get())
+            .timestamp(start)
+            .run()
+            .whenComplete(
+                (m, e) ->
+                    observer.accept(System.currentTimeMillis() - start, m.serializedValueSize()));
+        return State.RUNNING;
+      }
+
       @Override
       public State execute() throws InterruptedException {
         // Wait for all consumers get assignment.
         manager.awaitPartitionAssignment();
-        var rand = new Random();
         // Do transactional send.
-        if (param.transaction()) {
-          var senders =
-              IntStream.range(0, param.transactionSize)
-                  .mapToObj(i -> manager.payload())
-                  .filter(Optional::isPresent)
-                  .map(
-                      p ->
-                          producer
-                              .sender()
-                              .topic(param.topic)
-                              .partition(partitions.get(rand.nextInt(partitions.size())))
-                              .key(manager.getKey())
-                              .value(p.get())
-                              .timestamp(System.currentTimeMillis()))
-                  .collect(Collectors.toList());
-
-          // No records to send
-          if (senders.isEmpty()) return State.DONE;
-          transactionalProducer
-              .transaction(senders)
-              .forEach(
-                  future ->
-                      future.whenComplete(
-                          (m, e) ->
-                              observer.accept(
-                                  System.currentTimeMillis() - m.timestamp(),
-                                  m.serializedValueSize())));
-        } else {
-          var payload = manager.payload();
-          if (payload.isEmpty()) return State.DONE;
-
-          long start = System.currentTimeMillis();
-          producer
-              .sender()
-              .topic(param.topic)
-              .partition(partitions.get(rand.nextInt(partitions.size())))
-              .key(manager.getKey())
-              .value(payload.get())
-              .timestamp(start)
-              .run()
-              .whenComplete(
-                  (m, e) ->
-                      observer.accept(System.currentTimeMillis() - start, m.serializedValueSize()));
+        switch (param.isolation()) {
+          case READ_COMMITTED:
+            return doTransaction();
+          case READ_UNCOMMITTED:
+            return doSend();
+          default:
+            return State.RUNNING;
         }
-        return State.RUNNING;
       }
 
       @Override
@@ -336,25 +352,12 @@ public class Performance {
         validateWith = StringMapField.class)
     Map<String, String> configs = Map.of();
 
-    public Map<String, Object> producerProps() {
-      var props = props();
-      props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compression.name);
-      props.putAll(configs);
-      return props;
-    }
-
-    public Map<String, Object> consumerProps() {
-      var props = props();
-      if (transaction()) props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
-      return props;
-    }
-
     @Parameter(
         names = {"--compression"},
         description =
             "String: the compression algorithm used by producer. Available algorithm are none, gzip, snappy, lz4, and zstd",
         converter = CompressionField.class)
-    CompressionType compression = CompressionType.NONE;
+    Compression compression = Compression.NONE;
 
     @Parameter(
         names = {"--transaction.size"},
@@ -362,8 +365,8 @@ public class Performance {
         validateWith = PositiveLongField.class)
     int transactionSize = 1;
 
-    public boolean transaction() {
-      return transactionSize > 1;
+    Isolation isolation() {
+      return transactionSize > 1 ? Isolation.READ_COMMITTED : Isolation.READ_UNCOMMITTED;
     }
 
     @Parameter(
diff --git a/app/src/main/java/org/astraea/producer/Builder.java b/app/src/main/java/org/astraea/producer/Builder.java
index 9f157719..4d82ff6c 100644
--- a/app/src/main/java/org/astraea/producer/Builder.java
+++ b/app/src/main/java/org/astraea/producer/Builder.java
@@ -17,6 +17,7 @@ import org.apache.kafka.common.errors.AuthorizationException;
 import org.apache.kafka.common.errors.OutOfOrderSequenceException;
 import org.apache.kafka.common.errors.ProducerFencedException;
 import org.astraea.consumer.Header;
+import org.astraea.topic.Compression;
 
 public class Builder<Key, Value> {
   private final Map<String, Object> configs = new HashMap<>();
@@ -37,20 +38,27 @@ public class Builder<Key, Value> {
     return (Builder<Key, NewValue>) this;
   }
 
-  public Builder<Key, Value> configs(Map<String, Object> configs) {
+  public Builder<Key, Value> config(String key, String value) {
+    this.configs.put(key, value);
+    return this;
+  }
+
+  public Builder<Key, Value> configs(Map<String, String> configs) {
     this.configs.putAll(configs);
     return this;
   }
 
   public Builder<Key, Value> brokers(String brokers) {
-    this.configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Objects.requireNonNull(brokers));
-    return this;
+    return config(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Objects.requireNonNull(brokers));
   }
 
   public Builder<Key, Value> partitionClassName(String partitionClassName) {
-    this.configs.put(
+    return config(
         ProducerConfig.PARTITIONER_CLASS_CONFIG, Objects.requireNonNull(partitionClassName));
-    return this;
+  }
+
+  public Builder<Key, Value> compression(Compression compression) {
+    return config(ProducerConfig.COMPRESSION_TYPE_CONFIG, compression.nameOfKafka());
   }
 
   private static <Key, Value> CompletionStage<Metadata> doSend(
diff --git a/app/src/main/java/org/astraea/producer/Producer.java b/app/src/main/java/org/astraea/producer/Producer.java
index 2d49f8c5..11ddbbd0 100644
--- a/app/src/main/java/org/astraea/producer/Producer.java
+++ b/app/src/main/java/org/astraea/producer/Producer.java
@@ -21,7 +21,7 @@ public interface Producer<Key, Value> extends AutoCloseable {
     return builder().brokers(brokers).build();
   }
 
-  static Producer<byte[], byte[]> of(Map<String, Object> configs) {
+  static Producer<byte[], byte[]> of(Map<String, String> configs) {
     return builder().configs(configs).build();
   }
 }
diff --git a/app/src/main/java/org/astraea/topic/Builder.java b/app/src/main/java/org/astraea/topic/Builder.java
index e89c7e29..25946585 100644
--- a/app/src/main/java/org/astraea/topic/Builder.java
+++ b/app/src/main/java/org/astraea/topic/Builder.java
@@ -398,6 +398,12 @@ public class Builder {
       return this;
     }
 
+    @Override
+    public Creator config(String key, String value) {
+      this.configs.put(key, value);
+      return this;
+    }
+
     @Override
     public Creator configs(Map<String, String> configs) {
       this.configs.putAll(configs);
diff --git a/app/src/main/java/org/astraea/topic/Compression.java b/app/src/main/java/org/astraea/topic/Compression.java
new file mode 100644
index 00000000..30e19b3a
--- /dev/null
+++ b/app/src/main/java/org/astraea/topic/Compression.java
@@ -0,0 +1,20 @@
+package org.astraea.topic;
+
+import java.util.Locale;
+
+public enum Compression {
+  NONE,
+  GZIP,
+  SNAPPY,
+  LZ4,
+  ZSTD;
+
+  public static Compression of(String name) {
+    return Compression.valueOf(name.toUpperCase(Locale.ROOT));
+  }
+
+  /** @return the name parsed by kafka */
+  public String nameOfKafka() {
+    return name().toLowerCase(Locale.ROOT);
+  }
+}
diff --git a/app/src/main/java/org/astraea/topic/Creator.java b/app/src/main/java/org/astraea/topic/Creator.java
index 42d8fafe..1940c28f 100644
--- a/app/src/main/java/org/astraea/topic/Creator.java
+++ b/app/src/main/java/org/astraea/topic/Creator.java
@@ -17,7 +17,7 @@ public interface Creator {
    * @return this creator
    */
   default Creator compacted() {
-    return configs(Map.of(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT));
+    return config(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
   }
 
   /**
@@ -28,10 +28,15 @@ public interface Creator {
    */
   default Creator compactionMaxLag(Duration value) {
     return compacted()
-        .configs(
-            Map.of(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, String.valueOf(value.toMillis())));
+        .config(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, String.valueOf(value.toMillis()));
   }
 
+  default Creator compression(Compression compression) {
+    return config(TopicConfig.COMPRESSION_TYPE_CONFIG, compression.nameOfKafka());
+  }
+
+  Creator config(String key, String value);
+
   /**
    * @param configs used to create new topic
    * @return this creator


Test patch:
diff --git a/app/src/test/java/org/astraea/argument/CompressionFieldTest.java b/app/src/test/java/org/astraea/argument/CompressionFieldTest.java
index be6c69e4..dc28a507 100644
--- a/app/src/test/java/org/astraea/argument/CompressionFieldTest.java
+++ b/app/src/test/java/org/astraea/argument/CompressionFieldTest.java
@@ -3,7 +3,7 @@ package org.astraea.argument;
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.ParameterException;
 import java.util.stream.Stream;
-import org.apache.kafka.common.record.CompressionType;
+import org.astraea.topic.Compression;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -14,14 +14,14 @@ public class CompressionFieldTest {
         names = {"--field"},
         converter = CompressionField.class,
         validateWith = CompressionField.class)
-    public CompressionType value;
+    public Compression value;
   }
 
   @Test
   void testConversion() {
     var arg = new CompressionField();
-    Stream.of(CompressionType.values())
-        .forEach(type -> Assertions.assertEquals(type, arg.convert(type.name)));
+    Stream.of(Compression.values())
+        .forEach(type -> Assertions.assertEquals(type, arg.convert(type.name())));
     Assertions.assertThrows(ParameterException.class, () -> arg.convert("aaa"));
   }
 
@@ -29,6 +29,6 @@ public class CompressionFieldTest {
   void testParse() {
     var arg =
         org.astraea.argument.Argument.parse(new FakeParameter(), new String[] {"--field", "gzip"});
-    Assertions.assertEquals(CompressionType.GZIP, arg.value);
+    Assertions.assertEquals(Compression.GZIP, arg.value);
   }
 }
diff --git a/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java b/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
index c655bfa4..5c50861d 100644
--- a/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
+++ b/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
@@ -100,7 +100,8 @@ public class PartitionerTest extends RequireBrokerCluster {
             .keySerializer(Serializer.STRING)
             .configs(
                 initProConfig().entrySet().stream()
-                    .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue)))
+                    .collect(
+                        Collectors.toMap(e -> e.getKey().toString(), e -> e.getValue().toString())))
             .build()) {
       var i = 0;
       while (i < 300) {
@@ -166,7 +167,7 @@ public class PartitionerTest extends RequireBrokerCluster {
                                             .collect(
                                                 Collectors.toMap(
                                                     e -> e.getKey().toString(),
-                                                    Map.Entry::getValue)))
+                                                    e -> e.getValue().toString())))
                                     .build(),
                                 topicName,
                                 key,
@@ -265,7 +266,8 @@ public class PartitionerTest extends RequireBrokerCluster {
             .keySerializer(Serializer.STRING)
             .configs(
                 props.entrySet().stream()
-                    .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue)))
+                    .collect(
+                        Collectors.toMap(e -> e.getKey().toString(), e -> e.getValue().toString())))
             .build()) {
       var metadata =
           producer
diff --git a/app/src/test/java/org/astraea/performance/PerformanceTest.java b/app/src/test/java/org/astraea/performance/PerformanceTest.java
index e8e2c7a7..2aee96c2 100644
--- a/app/src/test/java/org/astraea/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/performance/PerformanceTest.java
@@ -12,6 +12,7 @@ import org.astraea.Utils;
 import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
+import org.astraea.consumer.Isolation;
 import org.astraea.producer.Producer;
 import org.astraea.service.RequireBrokerCluster;
 import org.astraea.topic.TopicAdmin;
@@ -160,9 +161,9 @@ public class PerformanceTest extends RequireBrokerCluster {
   @Test
   void testTransactionSet() {
     var argument = new Performance.Argument();
-    Assertions.assertFalse(argument.transaction());
+    Assertions.assertEquals(Isolation.READ_UNCOMMITTED, argument.isolation());
     argument.transactionSize = 3;
-    Assertions.assertTrue(argument.transaction());
+    Assertions.assertEquals(Isolation.READ_COMMITTED, argument.isolation());
   }
 
   @Test
@@ -197,7 +198,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     };
 
     var arg = org.astraea.argument.Argument.parse(new Performance.Argument(), arguments1);
-    Assertions.assertEquals("value", arg.producerProps().get("key").toString());
+    Assertions.assertEquals("value", arg.configs.get("key"));
 
     String[] arguments2 = {"--bootstrap.servers", "localhost:9092", "--topic", ""};
     Assertions.assertThrows(
diff --git a/app/src/test/java/org/astraea/producer/ProducerTest.java b/app/src/test/java/org/astraea/producer/ProducerTest.java
index f5cd9bf3..0a422a0e 100644
--- a/app/src/test/java/org/astraea/producer/ProducerTest.java
+++ b/app/src/test/java/org/astraea/producer/ProducerTest.java
@@ -3,13 +3,12 @@ package org.astraea.producer;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Deserializer;
 import org.astraea.consumer.Header;
+import org.astraea.consumer.Isolation;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -81,14 +80,13 @@ public class ProducerTest extends RequireBrokerCluster {
       producer.transaction(senders);
     }
 
-    Map<String, Object> prop = Map.of(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
     try (var consumer =
         Consumer.builder()
             .brokers(bootstrapServers())
             .fromBeginning()
             .topics(Set.of(topicName))
             .keyDeserializer(Deserializer.STRING)
-            .configs(prop)
+            .isolation(Isolation.READ_COMMITTED)
             .build()) {
       var records = consumer.poll(Duration.ofSeconds(10));
       Assertions.assertEquals(3, records.size());
diff --git a/app/src/test/java/org/astraea/topic/CompressionTest.java b/app/src/test/java/org/astraea/topic/CompressionTest.java
new file mode 100644
index 00000000..5c9c954e
--- /dev/null
+++ b/app/src/test/java/org/astraea/topic/CompressionTest.java
@@ -0,0 +1,19 @@
+package org.astraea.topic;
+
+import java.util.Arrays;
+import org.apache.kafka.common.record.CompressionType;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class CompressionTest {
+
+  @Test
+  void testNameOfKafka() {
+    Assertions.assertEquals(Compression.values().length, CompressionType.values().length);
+    Arrays.stream(Compression.values())
+        .forEach(
+            c ->
+                Assertions.assertEquals(
+                    c.nameOfKafka(), CompressionType.forName(c.nameOfKafka()).name));
+  }
+}