Repository: skiptests/astraea
Pull request #114 (2021-11-20 12:43:18)

Implement API for change log folder of partitions in TopicAdmin
#75
在TopicAdmin實作搬移partitions的API以及在embedded cluster配置多個folders

qoo332001: >Do you plan to update dockerfile完成 in this PR? Or you can enhance it in another one.

I will try to update the dockerfile in this PR tomorrow, after this PR merged I'll create another PR to complete #88

Patch:
diff --git a/app/build.gradle b/app/build.gradle
index f35bed23..e7602a13 100644
--- a/app/build.gradle
+++ b/app/build.gradle
@@ -1,11 +1,3 @@
-/*
- * This file was generated by the Gradle 'init' task.
- *
- * This generated file contains a sample Java application project to get you started.
- * For more details take a look at the 'Building Java & JVM projects' chapter in the Gradle
- * User Manual available at https://docs.gradle.org/7.2/userguide/building_java_projects.html
- */
-
 plugins {
     // Apply the application plugin to add support for building a CLI application in Java.
     id 'application'
@@ -18,9 +10,9 @@ repositories {
 }
 
 def versions = [
-    kafka: project.properties['kafka.version'] ?: "2.8.0",
+    kafka: project.properties['kafka.version'] ?: "2.8.1",
     junit: project.properties['junit.version'] ?: "5.8.1",
-    mockito: project.properties['mockito.version'] ?: "3.12.4",
+    mockito: project.properties['mockito.version'] ?: "4.1.0",
     jcommander: project.properties['jcommander.version'] ?: "1.81",
     slf4j: project.properties['slf4j.version'] ?: "1.7.32",
     zookeeper: project.properties['zookeeper.version'] ?: "3.5.9",
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 8142173d..563d2206 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -1,6 +1,8 @@
 package org.astraea.performance;
 
+import com.beust.jcommander.IStringConverter;
 import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
 import java.io.IOException;
 import java.time.Duration;
 import java.util.Map;
@@ -10,8 +12,11 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
 import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.record.CompressionType;
 import org.astraea.argument.ArgumentUtil;
 import org.astraea.argument.BasicArgumentWithPropFile;
 import org.astraea.concurrent.ThreadPool;
@@ -100,7 +105,7 @@ public class Performance {
                         i ->
                             producerExecutor(
                                 Producer.builder()
-                                    .configs(param.props())
+                                    .configs(param.producerProps())
                                     .partitionClassName(param.partitioner)
                                     .build(),
                                 param,
@@ -240,11 +245,36 @@ public class Performance {
         validateWith = ArgumentUtil.NotEmptyString.class)
     String partitioner = DefaultPartitioner.class.getName();
 
+    public Map<String, Object> producerProps() {
+      var props = props();
+      props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compression.name);
+      if (!this.jmxServers.isEmpty()) props.put("jmx_servers", this.jmxServers);
+      return props;
+    }
+
+    @Parameter(
+        names = {"--compression"},
+        description =
+            "String: the compression algorithm used by producer. Available algorithm are none, gzip, snappy, lz4, and zstd",
+        converter = CompressionArgument.class)
+    CompressionType compression = CompressionType.NONE;
+  }
+
+  static class CompressionArgument implements IStringConverter<CompressionType> {
+
     @Override
-    public Map<String, Object> props() {
-      Map<String, Object> prop = properties(propFile);
-      if (!this.jmxServers.isEmpty()) prop.put("jmx_servers", this.jmxServers);
-      return prop;
+    public CompressionType convert(String value) {
+      try {
+        return CompressionType.forName(value.toLowerCase());
+      } catch (IllegalArgumentException e) {
+        throw new ParameterException(
+            "the "
+                + value
+                + " is unsupported. The supported algorithms are "
+                + Stream.of(CompressionType.values())
+                    .map(CompressionType::name)
+                    .collect(Collectors.joining(",")));
+      }
     }
   }
 }
diff --git a/app/src/main/java/org/astraea/topic/Builder.java b/app/src/main/java/org/astraea/topic/Builder.java
index 236f8125..abc0f6f9 100644
--- a/app/src/main/java/org/astraea/topic/Builder.java
+++ b/app/src/main/java/org/astraea/topic/Builder.java
@@ -6,6 +6,7 @@ import java.util.stream.Collectors;
 import org.apache.kafka.clients.admin.*;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionReplica;
 import org.apache.kafka.common.config.ConfigResource;
 import org.astraea.Utils;
 
@@ -50,6 +51,14 @@ public class Builder {
                   .collect(Collectors.toSet()));
     }
 
+    @Override
+    public Map<Integer, Set<String>> brokerFolders(Set<Integer> brokers) {
+      return Utils.handleException(
+          () ->
+              admin.describeLogDirs(brokers).allDescriptions().get().entrySet().stream()
+                  .collect(Collectors.toMap(Map.Entry::getKey, map -> map.getValue().keySet())));
+    }
+
     @Override
     public Migrator migrator() {
       return new MigratorImpl(admin, this::partitions);
@@ -435,6 +444,22 @@ public class Builder {
       return this;
     }
 
+    @Override
+    public void moveTo(Map<Integer, String> brokerFolders) {
+      Utils.handleException(
+          () ->
+              admin.alterReplicaLogDirs(
+                  brokerFolders.entrySet().stream()
+                      .collect(
+                          Collectors.toMap(
+                              x ->
+                                  new TopicPartitionReplica(
+                                      partitions.iterator().next().topic(),
+                                      partitions.iterator().next().partition(),
+                                      x.getKey()),
+                              Map.Entry::getValue))));
+    }
+
     @Override
     public void moveTo(Set<Integer> brokers) {
       Utils.handleException(
diff --git a/app/src/main/java/org/astraea/topic/Migrator.java b/app/src/main/java/org/astraea/topic/Migrator.java
index e8463759..aeec5f19 100644
--- a/app/src/main/java/org/astraea/topic/Migrator.java
+++ b/app/src/main/java/org/astraea/topic/Migrator.java
@@ -1,5 +1,6 @@
 package org.astraea.topic;
 
+import java.util.Map;
 import java.util.Set;
 
 /** used to migrate partitions to another broker or broker folder. */
@@ -28,4 +29,11 @@ public interface Migrator {
    * @param brokers to host partitions
    */
   void moveTo(Set<Integer> brokers);
+
+  /**
+   * move partition to specify folder.
+   *
+   * @param brokerFolders contain
+   */
+  void moveTo(Map<Integer, String> brokerFolders);
 }
diff --git a/app/src/main/java/org/astraea/topic/TopicAdmin.java b/app/src/main/java/org/astraea/topic/TopicAdmin.java
index e6680509..20eb906e 100644
--- a/app/src/main/java/org/astraea/topic/TopicAdmin.java
+++ b/app/src/main/java/org/astraea/topic/TopicAdmin.java
@@ -50,6 +50,12 @@ public interface TopicAdmin extends Closeable {
   /** @return all brokers' ids */
   Set<Integer> brokerIds();
 
+  /**
+   * @param brokers a Set containing broker's ID
+   * @return all log directory
+   */
+  Map<Integer, Set<String>> brokerFolders(Set<Integer> brokers);
+
   /** @return a partition migrator used to move partitions to another broker or folder. */
   Migrator migrator();
 }
diff --git a/build.gradle b/build.gradle
index be119968..cb321605 100644
--- a/build.gradle
+++ b/build.gradle
@@ -1,11 +1,3 @@
-/*
- * This file was generated by the Gradle 'init' task.
- *
- * This generated file contains a sample Java application project to get you started.
- * For more details take a look at the 'Building Java & JVM projects' chapter in the Gradle
- * User Manual available at https://docs.gradle.org/7.2/userguide/building_java_projects.html
- */
-
 plugins {
     id "com.diffplug.spotless" version "5.14.3"
 }
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index 8477b31e..74ec68fa 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -108,7 +108,7 @@ if [[ "$(cat $config_file | grep log.dirs)" != "" ]]; then
   echo "you should not define log.dirs"
   exit 2
 else
-  echo "log.dirs=/tmp/kafka-logs" >> "$config_file"
+  echo "log.dirs=/tmp/kafka-logs1,/tmp/kafka-logs2,/tmp/kafka-logs3" >> "$config_file"
 fi
 
 # auto-generate broker id if it does not exist
diff --git a/docker/start_zookeeper.sh b/docker/start_zookeeper.sh
index 191ff457..6824b183 100755
--- a/docker/start_zookeeper.sh
+++ b/docker/start_zookeeper.sh
@@ -21,7 +21,7 @@ if [[ "$(which docker)" == "" ]]; then
 fi
 
 if [[ -z "$ZOOKEEPER_VERSION" ]]; then
-  ZOOKEEPER_VERSION=3.6.3
+  ZOOKEEPER_VERSION=3.7.0
 fi
 
 zookeeper_user=astraea
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
index ffed3a25..e750102e 100644
--- a/gradle/wrapper/gradle-wrapper.properties
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.2-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.3-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/settings.gradle b/settings.gradle
index 39f73a13..7b4aa626 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -1,11 +1,2 @@
-/*
- * This file was generated by the Gradle 'init' task.
- *
- * The settings file is used to specify which projects to include in your build.
- *
- * Detailed information about configuring a multi-project build in Gradle can be found
- * in the user manual at https://docs.gradle.org/7.2/userguide/multi_project_builds.html
- */
-
 rootProject.name = 'astraea'
 include('app')


Test patch:
diff --git a/app/src/test/java/org/astraea/performance/CompressionArgumentTest.java b/app/src/test/java/org/astraea/performance/CompressionArgumentTest.java
new file mode 100644
index 00000000..30d96a54
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/CompressionArgumentTest.java
@@ -0,0 +1,41 @@
+package org.astraea.performance;
+
+import com.beust.jcommander.ParameterException;
+import java.util.stream.Stream;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.record.CompressionType;
+import org.astraea.argument.ArgumentUtil;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class CompressionArgumentTest {
+
+  @Test
+  void testConversion() {
+    var arg = new Performance.CompressionArgument();
+    Stream.of(CompressionType.values())
+        .forEach(type -> Assertions.assertEquals(type, arg.convert(type.name)));
+    Assertions.assertThrows(ParameterException.class, () -> arg.convert("aaa"));
+  }
+
+  @Test
+  void testParse() {
+    var arg =
+        ArgumentUtil.parseArgument(
+            new Performance.Argument(),
+            new String[] {"--bootstrap.servers", "aa", "--compression", "gzip"});
+    Assertions.assertEquals(CompressionType.GZIP, arg.compression);
+    Assertions.assertEquals(
+        "gzip", arg.producerProps().get(ProducerConfig.COMPRESSION_TYPE_CONFIG));
+    Assertions.assertNull(arg.props().get(ProducerConfig.COMPRESSION_TYPE_CONFIG));
+  }
+
+  @Test
+  void testJmxServer() {
+    var arg =
+        ArgumentUtil.parseArgument(
+            new Performance.Argument(),
+            new String[] {"--bootstrap.servers", "aa", "--jmx.servers", "aaa"});
+    Assertions.assertEquals("aaa", arg.producerProps().get("jmx_servers"));
+  }
+}
diff --git a/app/src/test/java/org/astraea/service/Services.java b/app/src/test/java/org/astraea/service/Services.java
index f027ecba..c3a7938c 100644
--- a/app/src/test/java/org/astraea/service/Services.java
+++ b/app/src/test/java/org/astraea/service/Services.java
@@ -5,6 +5,7 @@ import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import kafka.server.KafkaConfig;
@@ -23,13 +24,20 @@ public final class Services {
   static BrokerCluster brokerCluster(ZookeeperCluster zk, int numberOfBrokers) {
     var tempFolders =
         IntStream.range(0, numberOfBrokers)
-            .mapToObj(i -> Utils.createTempDirectory("local_kafka"))
-            .collect(Collectors.toUnmodifiableList());
+            .boxed()
+            .collect(
+                Collectors.toMap(
+                    Function.identity(),
+                    brokerId ->
+                        Set.of(
+                            Utils.createTempDirectory("local_kafka").getAbsolutePath(),
+                            Utils.createTempDirectory("local_kafka").getAbsolutePath(),
+                            Utils.createTempDirectory("local_kafka").getAbsolutePath())));
+
     var brokers =
         IntStream.range(0, numberOfBrokers)
             .mapToObj(
                 index -> {
-                  File logDir = tempFolders.get(index);
                   Properties config = new Properties();
                   // reduce the number from partitions and replicas to speedup the mini cluster
                   config.setProperty(
@@ -40,7 +48,9 @@ public final class Services {
                   config.setProperty(KafkaConfig$.MODULE$.BrokerIdProp(), String.valueOf(index));
                   // bind broker on random port
                   config.setProperty(KafkaConfig$.MODULE$.ListenersProp(), "PLAINTEXT://:0");
-                  config.setProperty(KafkaConfig$.MODULE$.LogDirProp(), logDir.getAbsolutePath());
+                  config.setProperty(
+                      KafkaConfig$.MODULE$.LogDirsProp(), String.join(",", tempFolders.get(index)));
+
                   // increase the timeout in order to avoid ZkTimeoutException
                   config.setProperty(
                       KafkaConfig$.MODULE$.ZkSessionTimeoutMsProp(), String.valueOf(30 * 1000));
@@ -65,7 +75,7 @@ public final class Services {
               broker.shutdown();
               broker.awaitShutdown();
             });
-        tempFolders.forEach(Utils::delete);
+        tempFolders.values().forEach(fs -> fs.forEach(f -> Utils.delete(new File(f))));
       }
 
       @Override
@@ -76,8 +86,8 @@ public final class Services {
       @Override
       public Map<Integer, Set<String>> logFolders() {
         return IntStream.range(0, numberOfBrokers)
-            .mapToObj(brokerId -> Map.entry(brokerId, Set.of(tempFolders.get(brokerId).toString())))
-            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+            .boxed()
+            .collect(Collectors.toMap(Function.identity(), tempFolders::get));
       }
     };
   }
diff --git a/app/src/test/java/org/astraea/topic/TopicAdminTest.java b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
index ccad4dcb..70eb2567 100644
--- a/app/src/test/java/org/astraea/topic/TopicAdminTest.java
+++ b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
@@ -11,6 +11,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 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.producer.Producer;
@@ -160,6 +161,37 @@ public class TopicAdminTest extends RequireBrokerCluster {
                 && partitionReplicas.size() == 1
                 && partitionReplicas.get(0).broker() == broker;
           });
+
+      var currentBroker =
+          topicAdmin
+              .replicas(Set.of(topicName))
+              .get(new TopicPartition(topicName, 0))
+              .get(0)
+              .broker();
+      var allPath = topicAdmin.brokerFolders(Set.of(currentBroker));
+      var otherPath =
+          allPath.get(currentBroker).stream()
+              .filter(
+                  i ->
+                      !i.contains(
+                          topicAdmin
+                              .replicas(Set.of(topicName))
+                              .get(new TopicPartition(topicName, 0))
+                              .get(0)
+                              .path()))
+              .collect(Collectors.toSet());
+      topicAdmin
+          .migrator()
+          .partition(topicName, 0)
+          .moveTo(Map.of(currentBroker, otherPath.iterator().next()));
+      Utils.waitFor(
+          () -> {
+            var replicas = topicAdmin.replicas(Set.of(topicName));
+            var partitionReplicas = replicas.entrySet().iterator().next().getValue();
+            return replicas.size() == 1
+                && partitionReplicas.size() == 1
+                && partitionReplicas.get(0).path().equals(otherPath.iterator().next());
+          });
     }
   }