Repository: skiptests/astraea
Pull request #139 (2021-12-04 08:13:42)

Enable to configure compact topic
compact topic也是一個用的資料方式，這邊讓我們封裝的APIs可以建立compact topic，並且補上測試情境


Patch:
diff --git a/app/src/main/java/org/astraea/consumer/Builder.java b/app/src/main/java/org/astraea/consumer/Builder.java
index 07b47bbb..9e5692fb 100644
--- a/app/src/main/java/org/astraea/consumer/Builder.java
+++ b/app/src/main/java/org/astraea/consumer/Builder.java
@@ -29,8 +29,13 @@ public class Builder<Key, Value> {
     return this;
   }
 
-  public Builder<Key, Value> offsetPolicy(OffsetPolicy offsetPolicy) {
-    this.offsetPolicy = Objects.requireNonNull(offsetPolicy);
+  /**
+   * make the consumer read data from beginning. By default, it reads the latest data.
+   *
+   * @return this builder
+   */
+  public Builder<Key, Value> fromBeginning() {
+    this.offsetPolicy = OffsetPolicy.EARLIEST;
     return this;
   }
 
diff --git a/app/src/main/java/org/astraea/producer/Builder.java b/app/src/main/java/org/astraea/producer/Builder.java
index 501394df..df4fea0b 100644
--- a/app/src/main/java/org/astraea/producer/Builder.java
+++ b/app/src/main/java/org/astraea/producer/Builder.java
@@ -112,6 +112,11 @@ public class Builder<Key, Value> {
         };
       }
 
+      @Override
+      public void flush() {
+        kafkaProducer.flush();
+      }
+
       @Override
       public void close() {
         kafkaProducer.close();
diff --git a/app/src/main/java/org/astraea/producer/Producer.java b/app/src/main/java/org/astraea/producer/Producer.java
index 62292971..fc0813e0 100644
--- a/app/src/main/java/org/astraea/producer/Producer.java
+++ b/app/src/main/java/org/astraea/producer/Producer.java
@@ -6,6 +6,9 @@ import java.util.Map;
 public interface Producer<Key, Value> extends AutoCloseable {
   Sender<Key, Value> sender();
 
+  /** this method is blocked until all data in buffer are sent. */
+  void flush();
+
   void close();
 
   static Builder<byte[], byte[]> builder() {
diff --git a/app/src/main/java/org/astraea/topic/Creator.java b/app/src/main/java/org/astraea/topic/Creator.java
index 243a9a3a..42d8fafe 100644
--- a/app/src/main/java/org/astraea/topic/Creator.java
+++ b/app/src/main/java/org/astraea/topic/Creator.java
@@ -1,6 +1,8 @@
 package org.astraea.topic;
 
+import java.time.Duration;
 import java.util.Map;
+import org.apache.kafka.common.config.TopicConfig;
 
 public interface Creator {
   Creator topic(String topic);
@@ -9,6 +11,27 @@ public interface Creator {
 
   Creator numberOfReplicas(short numberOfReplicas);
 
+  /**
+   * set the cleanup policy to compact. By default, the policy is `delete`/
+   *
+   * @return this creator
+   */
+  default Creator compacted() {
+    return configs(Map.of(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT));
+  }
+
+  /**
+   * the max time to make segments be eligible for compaction.
+   *
+   * @param value the max lag
+   * @return this creator
+   */
+  default Creator compactionMaxLag(Duration value) {
+    return compacted()
+        .configs(
+            Map.of(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, String.valueOf(value.toMillis())));
+  }
+
   /**
    * @param configs used to create new topic
    * @return this creator
diff --git a/app/src/main/java/org/astraea/topic/ReplicaSyncingMonitor.java b/app/src/main/java/org/astraea/topic/ReplicaSyncingMonitor.java
index 659d8b57..784aeb75 100644
--- a/app/src/main/java/org/astraea/topic/ReplicaSyncingMonitor.java
+++ b/app/src/main/java/org/astraea/topic/ReplicaSyncingMonitor.java
@@ -1,7 +1,6 @@
 package org.astraea.topic;
 
 import com.beust.jcommander.Parameter;
-import java.io.IOException;
 import java.time.Duration;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
@@ -22,8 +21,6 @@ public class ReplicaSyncingMonitor {
     Argument argument = ArgumentUtil.parseArgument(new Argument(), args);
     try (TopicAdmin topicAdmin = TopicAdmin.of(argument.props())) {
       execute(topicAdmin, argument);
-    } catch (IOException ioException) {
-      ioException.printStackTrace();
     }
   }
 
diff --git a/app/src/main/java/org/astraea/topic/TopicAdmin.java b/app/src/main/java/org/astraea/topic/TopicAdmin.java
index 20eb906e..d11434a2 100644
--- a/app/src/main/java/org/astraea/topic/TopicAdmin.java
+++ b/app/src/main/java/org/astraea/topic/TopicAdmin.java
@@ -58,4 +58,7 @@ public interface TopicAdmin extends Closeable {
 
   /** @return a partition migrator used to move partitions to another broker or folder. */
   Migrator migrator();
+
+  @Override
+  void close();
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java
index 051532b8..b7062ea1 100644
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java
@@ -9,7 +9,6 @@ import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.astraea.consumer.Builder;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Deserializer;
 import org.astraea.consumer.Header;
@@ -71,7 +70,7 @@ public class PartitionerTest extends RequireBrokerCluster {
     try (var consumer =
         Consumer.builder()
             .brokers(bootstrapServers())
-            .offsetPolicy(Builder.OffsetPolicy.EARLIEST)
+            .fromBeginning()
             .topics(Set.of(topicName))
             .keyDeserializer(Deserializer.STRING)
             .build()) {
diff --git a/app/src/test/java/org/astraea/producer/ProducerTest.java b/app/src/test/java/org/astraea/producer/ProducerTest.java
index f2add21b..e8e123af 100644
--- a/app/src/test/java/org/astraea/producer/ProducerTest.java
+++ b/app/src/test/java/org/astraea/producer/ProducerTest.java
@@ -4,7 +4,6 @@ import java.time.Duration;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
-import org.astraea.consumer.Builder;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Deserializer;
 import org.astraea.consumer.Header;
@@ -39,7 +38,7 @@ public class ProducerTest extends RequireBrokerCluster {
     try (var consumer =
         Consumer.builder()
             .brokers(bootstrapServers())
-            .offsetPolicy(Builder.OffsetPolicy.EARLIEST)
+            .fromBeginning()
             .topics(Set.of(topicName))
             .keyDeserializer(Deserializer.STRING)
             .build()) {
diff --git a/app/src/test/java/org/astraea/service/Services.java b/app/src/test/java/org/astraea/service/Services.java
index c3a7938c..1031d118 100644
--- a/app/src/test/java/org/astraea/service/Services.java
+++ b/app/src/test/java/org/astraea/service/Services.java
@@ -39,6 +39,9 @@ public final class Services {
             .mapToObj(
                 index -> {
                   Properties config = new Properties();
+                  // reduce the backoff of compact thread to test it quickly
+                  config.setProperty(
+                      KafkaConfig$.MODULE$.LogCleanerBackoffMsProp(), String.valueOf(2000));
                   // reduce the number from partitions and replicas to speedup the mini cluster
                   config.setProperty(
                       KafkaConfig$.MODULE$.OffsetsTopicPartitionsProp(), String.valueOf(1));
diff --git a/app/src/test/java/org/astraea/topic/TopicAdminTest.java b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
index 70eb2567..ce80d02d 100644
--- a/app/src/test/java/org/astraea/topic/TopicAdminTest.java
+++ b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
@@ -2,10 +2,8 @@ package org.astraea.topic;
 
 import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
-import java.io.IOException;
-import java.util.Collection;
-import java.util.Map;
-import java.util.Set;
+import java.time.Duration;
+import java.util.*;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -14,7 +12,10 @@ import java.util.stream.StreamSupport;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.config.TopicConfig;
 import org.astraea.Utils;
+import org.astraea.consumer.Consumer;
+import org.astraea.consumer.Deserializer;
 import org.astraea.producer.Producer;
+import org.astraea.producer.Serializer;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -23,7 +24,7 @@ import org.junit.jupiter.api.condition.DisabledOnOs;
 public class TopicAdminTest extends RequireBrokerCluster {
 
   @Test
-  void testCreator() throws IOException {
+  void testCreator() {
     var topicName = "testCreator";
     try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
       topicAdmin
@@ -49,7 +50,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testCreateTopicRepeatedly() throws IOException {
+  void testCreateTopicRepeatedly() {
     var topicName = "testCreateTopicRepeatedly";
     try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
       Runnable createTopic =
@@ -89,7 +90,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testPartitions() throws IOException, InterruptedException {
+  void testPartitions() throws InterruptedException {
     var topicName = "testPartitions";
     try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
       topicAdmin.creator().topic(topicName).numberOfPartitions(3).create();
@@ -112,7 +113,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testGroups() throws IOException, InterruptedException {
+  void testGroups() throws InterruptedException {
     var topicName = "testGroups";
     try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
       topicAdmin.creator().topic(topicName).numberOfPartitions(3).create();
@@ -125,7 +126,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testOffsets() throws IOException, InterruptedException {
+  void testOffsets() throws InterruptedException {
     var topicName = "testOffsets";
     try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
       topicAdmin.creator().topic(topicName).numberOfPartitions(3).create();
@@ -145,7 +146,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
 
   @Test
   @DisabledOnOs(WINDOWS)
-  void testMigrateSinglePartition() throws IOException, InterruptedException {
+  void testMigrateSinglePartition() throws InterruptedException {
     var topicName = "testMigrateSinglePartition";
     try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
       topicAdmin.creator().topic(topicName).numberOfPartitions(1).create();
@@ -197,7 +198,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
 
   @Test
   @DisabledOnOs(WINDOWS)
-  void testMigrateAllPartitions() throws IOException, InterruptedException {
+  void testMigrateAllPartitions() throws InterruptedException {
     var topicName = "testMigrateAllPartitions";
     try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
       topicAdmin.creator().topic(topicName).numberOfPartitions(3).create();
@@ -217,7 +218,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testReplicaSize() throws IOException, ExecutionException, InterruptedException {
+  void testReplicaSize() throws ExecutionException, InterruptedException {
     var topicName = "testReplicaSize";
     try (var topicAdmin = TopicAdmin.of(bootstrapServers());
         var producer = Producer.builder().brokers(bootstrapServers()).build()) {
@@ -247,4 +248,59 @@ public class TopicAdminTest extends RequireBrokerCluster {
       Assertions.assertTrue(newSize > originSize);
     }
   }
+
+  @Test
+  void testCompact() throws InterruptedException {
+    var topicName = "testCompacted";
+    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+      topicAdmin.creator().topic(topicName).compactionMaxLag(Duration.ofSeconds(1)).create();
+
+      var key = "key";
+      var anotherKey = "anotherKey";
+      var value = "value";
+      try (var producer =
+          Producer.builder()
+              .keySerializer(Serializer.STRING)
+              .valueSerializer(Serializer.STRING)
+              .brokers(bootstrapServers())
+              .build()) {
+        IntStream.range(0, 10)
+            .forEach(i -> producer.sender().key(key).value(value).topic(topicName).run());
+        producer.flush();
+
+        // sleep and produce more data to generate the new segment
+        TimeUnit.SECONDS.sleep(2);
+        IntStream.range(0, 10)
+            .forEach(i -> producer.sender().key(anotherKey).value(value).topic(topicName).run());
+        producer.flush();
+      }
+
+      // sleep for compact (the backoff of compact thread is reduced to 2 seconds.
+      // see org.astraea.service.Services)
+      TimeUnit.SECONDS.sleep(3);
+
+      try (var consumer =
+          Consumer.builder()
+              .keyDeserializer(Deserializer.STRING)
+              .valueDeserializer(Deserializer.STRING)
+              .fromBeginning()
+              .brokers(bootstrapServers())
+              .topics(Set.of(topicName))
+              .build()) {
+
+        var records =
+            IntStream.range(0, 5)
+                .mapToObj(i -> consumer.poll(Duration.ofSeconds(1)))
+                .flatMap(Collection::stream)
+                .collect(Collectors.toList());
+
+        Assertions.assertEquals(
+            1, records.stream().filter(record -> record.key().equals(key)).count());
+
+        // those data are in active segment, so they can be compacted.
+        Assertions.assertEquals(
+            10, records.stream().filter(record -> record.key().equals(anotherKey)).count());
+      }
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/topic/cost/GetPartitionInfTest.java b/app/src/test/java/org/astraea/topic/cost/GetPartitionInfTest.java
index 52d70e24..f06c348c 100644
--- a/app/src/test/java/org/astraea/topic/cost/GetPartitionInfTest.java
+++ b/app/src/test/java/org/astraea/topic/cost/GetPartitionInfTest.java
@@ -2,7 +2,6 @@ package org.astraea.topic.cost;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
-import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -46,7 +45,7 @@ public class GetPartitionInfTest extends RequireBrokerCluster {
           .create();
       // wait for topic
       TimeUnit.SECONDS.sleep(5);
-    } catch (InterruptedException | IOException e) {
+    } catch (InterruptedException e) {
       e.printStackTrace();
     }
     var producer =
diff --git a/app/src/test/java/org/astraea/topic/cost/PartitionScoreTest.java b/app/src/test/java/org/astraea/topic/cost/PartitionScoreTest.java
index 8b7c5ad3..414018c7 100644
--- a/app/src/test/java/org/astraea/topic/cost/PartitionScoreTest.java
+++ b/app/src/test/java/org/astraea/topic/cost/PartitionScoreTest.java
@@ -2,7 +2,6 @@ package org.astraea.topic.cost;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
-import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -45,7 +44,7 @@ public class PartitionScoreTest extends RequireBrokerCluster {
           .create();
       // wait for topic
       TimeUnit.SECONDS.sleep(5);
-    } catch (InterruptedException | IOException e) {
+    } catch (InterruptedException e) {
       e.printStackTrace();
     }
     var producer =