KAFKA-16647 Remove setMetadataDirectory from BrokerNode/ControllerNode (

#15833)

Reviewers: Chia-Ping Tsai <[email protected]>

core/src/test/java/kafka/testkit/BrokerNode.java

@@ -44,7 +44,6 @@ public static class Builder {
         private String baseDirectory;
         private Uuid clusterId;
         private int numLogDirectories = 1;
-        private String metadataDirectory;
         private Map<String, String> propertyOverrides = Collections.emptyMap();
         private boolean combined;
 
@@ -59,11 +58,6 @@ public Builder setId(int id) {
             return this;
         }
 
-        public Builder setMetadataDirectory(String metadataDirectory) {
-            this.metadataDirectory = metadataDirectory;
-            return this;
-        }
-
         public Builder setNumLogDirectories(int numLogDirectories) {
             this.numLogDirectories = numLogDirectories;
             return this;
@@ -95,6 +89,9 @@ public BrokerNode build() {
             if (id == -1) {
                 throw new RuntimeException("You must set the node id.");
             }
+            if (numLogDirectories < 1) {
+                throw new RuntimeException("The value of numLogDirectories should be at least 1.");
+            }
             List<String> logDataDirectories = IntStream
                 .range(0, numLogDirectories)
                 .mapToObj(i -> {
@@ -110,14 +107,9 @@ public BrokerNode build() {
                     return new File(baseDirectory, logDir).getAbsolutePath();
                 })
                 .collect(Collectors.toList());
-            if (metadataDirectory == null) {
-                metadataDirectory = logDataDirectories.get(0);
-            } else if (!Paths.get(metadataDirectory).isAbsolute()) {
-                metadataDirectory = new File(baseDirectory, metadataDirectory).getAbsolutePath();
-            }
             MetaPropertiesEnsemble.Copier copier =
                     new MetaPropertiesEnsemble.Copier(MetaPropertiesEnsemble.EMPTY);
-            copier.setMetaLogDir(Optional.of(metadataDirectory));
+            copier.setMetaLogDir(Optional.of(logDataDirectories.get(0)));
             for (String logDir : logDataDirectories) {
                 copier.setLogDirProps(logDir, new MetaProperties.Builder().
                     setVersion(MetaPropertiesVersion.V1).
@@ -126,7 +118,6 @@ public BrokerNode build() {
                     setDirectoryId(copier.generateValidDirectoryId()).
                     build());
             }
-            copier.setMetaLogDir(Optional.of(metadataDirectory));
             return new BrokerNode(
                 copier.copy(),
                 combined,

core/src/test/java/kafka/testkit/BrokerNodeTest.java

@@ -0,0 +1,42 @@
+/*
+ * 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.testkit;
+
+import org.apache.kafka.common.Uuid;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class BrokerNodeTest {
+
+    @Test
+    public void testInvalidBuilder() {
+        Assertions.assertEquals("You must set the node id.",
+                Assertions.assertThrows(RuntimeException.class, () -> BrokerNode.builder()
+                        .setBaseDirectory("foo")
+                        .setClusterId(Uuid.randomUuid())
+                        .build()).getMessage());
+
+        Assertions.assertEquals("The value of numLogDirectories should be at least 1.",
+                Assertions.assertThrows(RuntimeException.class, () -> BrokerNode.builder()
+                        .setBaseDirectory("foo")
+                        .setClusterId(Uuid.randomUuid())
+                        .setId(0)
+                        .setNumLogDirectories(0)
+                        .build()).getMessage());
+    }
+}

core/src/test/java/kafka/testkit/ControllerNode.java

@@ -23,7 +23,6 @@
 import org.apache.kafka.metadata.properties.MetaPropertiesVersion;
 
 import java.io.File;
-import java.nio.file.Paths;
 import java.util.Objects;
 import java.util.Optional;
 
@@ -35,7 +34,6 @@ public static Builder builder() {
     public static class Builder {
         private int id = -1;
         private String baseDirectory;
-        private String metadataDirectory;
         private Uuid clusterId;
         private boolean combined;
 
@@ -50,11 +48,6 @@ public Builder setId(int id) {
             return this;
         }
 
-        public Builder setMetadataDirectory(String metadataDirectory) {
-            this.metadataDirectory = metadataDirectory;
-            return this;
-        }
-
         public Builder setClusterId(Uuid clusterId) {
             this.clusterId = clusterId;
             return this;
@@ -77,16 +70,8 @@ public ControllerNode build() {
             if (baseDirectory == null) {
                 throw new RuntimeException("You must set the base directory.");
             }
-            if (metadataDirectory == null) {
-                if (combined) {
-                    metadataDirectory = String.format("combined_%d_0", id);
-                } else {
-                    metadataDirectory = String.format("controller_%d", id);
-                }
-            }
-            if (!Paths.get(metadataDirectory).isAbsolute()) {
-                metadataDirectory = new File(baseDirectory, metadataDirectory).getAbsolutePath();
-            }
+            String metadataDirectory = new File(baseDirectory,
+                combined ? String.format("combined_%d_0", id) : String.format("controller_%d", id)).getAbsolutePath();
             MetaPropertiesEnsemble.Copier copier =
                 new MetaPropertiesEnsemble.Copier(MetaPropertiesEnsemble.EMPTY);
             copier.setMetaLogDir(Optional.of(metadataDirectory));