feat(tool): add admin tool to help user start AutoMQ easily

bin/automq-kafka-admin.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+# 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.
+
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.automq.AutoMQKafkaAdminTool "$@"
+

bin/kafka-run-class.sh

@@ -335,6 +335,8 @@ if [ "x$GC_LOG_ENABLED" = "xtrue" ]; then
 fi
 
 KAFKA_JDK_COMPATIBILITY_OPTS=""
+# We need to override KAFKA_S3_ACCESS_KEY and KAFKA_S3_SECRET_KEY. There is no method called System.setEnv, so we set system environment variable by reflection. Add this --add-opens to enable reflection to set system env in class EnvUtil
+KAFKA_JDK_COMPATIBILITY_OPTS="${KAFKA_JDK_COMPATIBILITY_OPTS} --add-opens=java.base/java.util=ALL-UNNAMED "
 if [[ "$JAVA_MAJOR_VERSION" -ge "9" ]] ; then
   KAFKA_JDK_COMPATIBILITY_OPTS="${KAFKA_JDK_COMPATIBILITY_OPTS} --add-opens=java.base/jdk.internal.ref=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED -Dio.netty.tryReflectionSetAccessible=true"
 fi

build.gradle

@@ -925,7 +925,7 @@ project(':core') {
     implementation project(':metadata')
     implementation project(':raft')
     implementation project(':storage')
-
+    implementation project(':s3shell-kafka-sdk')
 
     implementation libs.argparse4j
     implementation libs.jacksonDatabind
@@ -1171,6 +1171,8 @@ project(':core') {
     from(project(':tools').configurations.runtimeClasspath) { into("libs/") }
     from(project(':trogdor').jar) { into("libs/") }
     from(project(':trogdor').configurations.runtimeClasspath) { into("libs/") }
+    from(project(':s3shell-kafka-sdk').jar) { into("libs/") }
+    from(project(':s3shell-kafka-sdk').configurations.runtimeClasspath) { into("libs/") }
     from(project(':shell').jar) { into("libs/") }
     from(project(':shell').configurations.runtimeClasspath) { into("libs/") }
     from(project(':connect:api').jar) { into("libs/") }
@@ -1842,27 +1844,59 @@ project(':storage') {
   }
 }
 
+project(':s3shell-kafka-sdk') {
+  archivesBaseName = "s3shell-kafka-sdk"
+
+  checkstyle {
+    configProperties = checkstyleConfigProperties("import-control-automq.xml")
+  }
+}
+
 project(':tools') {
   archivesBaseName = "kafka-tools"
 
   dependencies {
-    implementation project(':clients')
-    implementation project(':server-common')
-    implementation project(':log4j-appender')
+    implementation (project(':clients')){
+      exclude group: 'org.slf4j', module: '*'
+    }
+    implementation (project(':server-common')){
+      exclude group: 'org.slf4j', module: '*'
+    }
+    implementation (project(':log4j-appender')){
+      exclude group: 'org.slf4j', module: '*'
+    }
+    implementation project(':s3shell-kafka-sdk')
+
     implementation libs.argparse4j
     implementation libs.jacksonDatabind
     implementation libs.jacksonJDK8Datatypes
     implementation libs.slf4jApi
     implementation libs.log4j
-
-    implementation libs.jose4j                    // for SASL/OAUTHBEARER JWT validation
+    // for SASL/OAUTHBEARER JWT validation
+    implementation (libs.jose4j){
+      exclude group: 'org.slf4j', module: '*'
+    }
     implementation libs.jacksonJaxrsJsonProvider
+    implementation(libs.s3stream) {
+      exclude group: 'org.slf4j', module: '*'
+      exclude group: 'net.sourceforge.argparse4j', module: '*'
+    }
+    implementation libs.commonio
 
-    testImplementation project(':clients')
+
+    testImplementation (project(':clients')){
+      exclude group: 'org.slf4j', module: '*'
+    }
     testImplementation project(':clients').sourceSets.test.output
-    testImplementation project(':core')
+    testImplementation (project(':core')){
+      exclude group: 'org.slf4j', module: '*'
+    }
     testImplementation project(':core').sourceSets.test.output
-    testImplementation project(':server-common')
+    testImplementation (project(':server-common')){
+      exclude group: 'org.slf4j', module: '*'
+    }
+
+
     testImplementation project(':server-common').sourceSets.test.output
     testImplementation libs.junitJupiter
     testImplementation libs.mockitoInline // supports mocking static methods, final classes, etc.

checkstyle/import-control-automq.xml

@@ -0,0 +1,55 @@
+<!DOCTYPE import-control PUBLIC
+    "-//Puppy Crawl//DTD Import Control 1.1//EN"
+    "http://www.puppycrawl.com/dtds/import_control_1_1.dtd">
+<!--
+ 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.
+-->
+
+<import-control pkg="com.automq">
+
+  <!-- THINK HARD ABOUT THE LAYERING OF THE PROJECT BEFORE CHANGING THIS FILE -->
+
+  <!-- common library dependencies -->
+  <!-- common library dependencies -->
+  <allow pkg="java" />
+  <allow pkg="javax.management" />
+  <allow pkg="org.slf4j" />
+  <allow pkg="org.junit" />
+  <allow pkg="org.opentest4j" />
+  <allow pkg="org.hamcrest" />
+  <allow pkg="org.mockito" />
+  <allow pkg="org.easymock" />
+  <allow pkg="org.powermock" />
+  <allow pkg="java.security" />
+  <allow pkg="javax.net.ssl" />
+  <allow pkg="javax.security" />
+  <allow pkg="org.ietf.jgss" />
+  <allow pkg="net.jqwik.api" />
+  <allow pkg="io.netty" />
+  <allow pkg="software.amazon.awssdk" />
+  <allow pkg="com.automq.stream" />
+  <allow pkg="org.apache.kafka.tools.automq"/>
+  <allow pkg="com.automq.s3shell"/>
+  <allow pkg="org.apache.commons.io"/>
+
+  <allow pkg="com.github.luben.zstd" />
+  <allow pkg="com.google.common.cache" />
+
+
+
+
+
+</import-control>

checkstyle/import-control-core.xml

@@ -146,4 +146,12 @@
     <allow pkg="org.apache.directory" />
     <allow pkg="org.mockito" />
   </subpackage>
+
+  <subpackage name="s3shell">
+    <allow pkg="joptsimple" />
+    <allow pkg="kafka.log.stream.s3" />
+    <allow pkg="kafka.s3shell" />
+    <allow pkg="kafka.tools" />
+    <allow pkg="com.automq.s3shell.sdk" />
+  </subpackage>
 </import-control>

checkstyle/import-control.xml

@@ -40,6 +40,10 @@
   <allow pkg="io.netty" />
   <allow pkg="software.amazon.awssdk" />
   <allow pkg="com.automq.stream" />
+  <allow pkg="org.apache.kafka.tools.automq"/>
+  <allow pkg="com.automq.s3shell"/>
+  <allow pkg="org.apache.commons.io"/>
+
   <allow pkg="com.github.luben.zstd" />
   <allow pkg="com.google.common.cache" />
 

core/src/main/java/kafka/s3shell/util/EnvUtil.java

@@ -0,0 +1,34 @@
+/*
+ * 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 kafka.s3shell.util;
+
+import java.lang.reflect.Field;
+import java.util.Map;
+
+public class EnvUtil {
+    public static void setEnv(String key, String value) {
+        try {
+            Map<String, String> env = System.getenv();
+            Field field = env.getClass().getDeclaredField("m");
+            field.setAccessible(true);
+            Map<String, String> writableEnv = (Map<String, String>) field.get(env);
+            writableEnv.put(key, value);
+        } catch (Exception e) {
+            throw new IllegalStateException("Failed to set environment variable", e);
+        }
+    }
+}

core/src/main/java/kafka/s3shell/util/KafkaFormatUtil.java

@@ -0,0 +1,44 @@
+/*
+ * 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 kafka.s3shell.util;
+
+import com.automq.s3shell.sdk.util.S3PropUtil;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Properties;
+import kafka.tools.StorageTool;
+
+public class KafkaFormatUtil {
+    public static void formatStorage(String clusterId, Properties props) throws IOException {
+        String propFileName = String.format("automq-%s.properties", clusterId);
+        String propFilePath = "generated/" + propFileName;
+        String logDirPath = props.getProperty("log.dirs");
+
+        Path propPath = Paths.get(propFilePath);
+        if (Files.exists(propPath)) {
+            //delete if exists
+            Files.delete(propPath);
+        }
+
+        S3PropUtil.persist(props, propFileName);
+        if (!Files.isDirectory(Paths.get(logDirPath)) || !Files.exists(Paths.get(logDirPath, "meta.properties"))) {
+            StorageTool.main(new String[] {"auto-format", "-t", clusterId, "-c=" + propFilePath});
+        }
+    }
+}

core/src/main/java/kafka/s3shell/util/S3ShellPropUtil.java

@@ -0,0 +1,116 @@
+/*
+ * 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 kafka.s3shell.util;
+
+import com.automq.s3shell.sdk.constant.ServerConfigKey;
+import com.automq.s3shell.sdk.model.S3Url;
+import com.automq.s3shell.sdk.util.S3PropUtil;
+import java.io.IOException;
+import java.util.List;
+import java.util.Properties;
+import joptsimple.OptionParser;
+import joptsimple.OptionSet;
+import org.apache.kafka.common.internals.FatalExitError;
+
+import static kafka.log.stream.s3.ConfigUtils.ACCESS_KEY_NAME;
+import static kafka.log.stream.s3.ConfigUtils.SECRET_KEY_NAME;
+
+public class S3ShellPropUtil {
+
+    private static OptionParser acceptOption() {
+        OptionParser optionParser = new OptionParser();
+        optionParser.accepts("override", "Optional property that should override values set in server.properties file")
+            .withRequiredArg()
+            .ofType(String.class);
+        optionParser.accepts("config", "Path to server.properties file")
+            .withRequiredArg()
+            .ofType(String.class);
+        optionParser.accepts("s3-url", "URL for S3 storage")
+            .withRequiredArg()
+            .ofType(String.class);
+        optionParser.accepts("version", "Print version information and exit.");
+        return optionParser;
+    }
+
+    public static Properties autoGenPropsByCmd(String[] args, String processRole) throws IOException {
+        if (args.length < 1) {
+            throw new FatalExitError(1);
+        }
+
+        Properties props = new Properties();
+        switch (processRole) {
+            case "broker":
+                props.putAll(S3PropUtil.loadTemplateProps(S3PropUtil.BROKER_PROPS_PATH));
+                break;
+            case "controller":
+                props.putAll(S3PropUtil.loadTemplateProps(S3PropUtil.CONTROLLER_PROPS_PATH));
+                break;
+            case "broker,controller":
+            case "controller,broker":
+                props.putAll(S3PropUtil.loadTemplateProps(S3PropUtil.SERVER_PROPS_PATH));
+                break;
+            default:
+                throw new IllegalArgumentException("Invalid process role:" + processRole);
+        }
+
+        // Handle --override options
+        OptionParser optionParser = acceptOption();
+        OptionSet options = optionParser.parse(args);
+        handleOption(options, props);
+
+        return props;
+    }
+
+    private static void handleOption(OptionSet options, Properties props) {
+        S3Url s3Url = null;
+        if (options.has("s3-url")) {
+            String s3UrlStr = (String) options.valueOf("s3-url");
+            s3Url = S3Url.parse(s3UrlStr);
+            props.put(ServerConfigKey.S3_ENDPOINT.getKeyName(), s3Url.getEndpointProtocol().getName() + "://" + s3Url.getS3Endpoint());
+            props.put(ServerConfigKey.S3_REGION.getKeyName(), s3Url.getS3Region());
+            props.put(ServerConfigKey.S3_BUCKET.getKeyName(), s3Url.getS3DataBucket());
+            props.put(ServerConfigKey.S3_PATH_STYLE.getKeyName(), String.valueOf(s3Url.isS3PathStyle()));
+
+            // override system env
+            EnvUtil.setEnv(ACCESS_KEY_NAME, s3Url.getS3AccessKey());
+            EnvUtil.setEnv(SECRET_KEY_NAME, s3Url.getS3SecretKey());
+        }
+
+        if (options.has("override")) {
+            List<?> overrideOptions = options.valuesOf("override");
+            for (Object o : overrideOptions) {
+                String option = (String) o;
+                String[] keyValue = option.split("=", 2);
+                if (keyValue.length == 2) {
+                    props.setProperty(keyValue[0], keyValue[1]);
+                } else {
+                    throw new IllegalArgumentException("Invalid override option format: " + option);
+                }
+            }
+        }
+
+        //format storage
+        if (s3Url != null) {
+            try {
+                KafkaFormatUtil.formatStorage(s3Url.getClusterId(), props);
+            } catch (IOException e) {
+                throw new RuntimeException(String.format("Format storage failed for cluster:%s", s3Url.getClusterId()), e);
+            }
+        }
+    }
+}
+