KAFKA-16518; Adding standalone argument for storage

core/src/test/scala/unit/kafka/utils/TestUtils.scala

@@ -1178,7 +1178,7 @@ object TestUtils extends Logging {
     try {
       out = new PrintStream(stream)
       val bootstrapMetadata = StorageTool.buildBootstrapMetadata(metadataVersion, optionalMetadataRecords, "format command")
-      if (StorageTool.formatCommand(out, directories, metaProperties, bootstrapMetadata, metadataVersion, ignoreFormatted = false) != 0) {
+      if (StorageTool.formatCommand(out, directories, metaProperties, bootstrapMetadata, metadataVersion, ignoreFormatted = false, null, null) != 0) {
         throw new RuntimeException(stream.toString())
       }
       debug(s"Formatted storage directory(ies) ${directories}")

raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java

@@ -30,6 +30,8 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import static org.apache.kafka.common.config.ConfigDef.Importance.HIGH;
@@ -271,6 +273,14 @@ public static List<Node> voterConnectionsToNodes(Map<Integer, InetSocketAddress>
             .collect(Collectors.toList());
     }
 
+    //  to match pattern <replica-id>[-<replica-directory-id>]@<host>:<port>   replica-directory-id is optional
+    public static boolean validateControllerQuorumVoters(String controllerQuorumVoters) {
+        String regex = "(\\d+)(-\\d+)?@([a-zA-Z0-9.-]+):(\\d{1,5})";
+        Pattern pattern = Pattern.compile(regex);
+        Matcher matcher = pattern.matcher(controllerQuorumVoters);
+        return matcher.matches();
+    }
+
     public static class ControllerQuorumVotersValidator implements ConfigDef.Validator {
         @Override
         public void ensureValid(String name, Object value) {

raft/src/main/java/org/apache/kafka/raft/internals/VoterSet.java

@@ -414,4 +414,19 @@ public static VoterSet fromInetSocketAddresses(ListenerName listener, Map<Intege
 
         return new VoterSet(voterNodes);
     }
+
+    /**
+     * Creates a voter set from endpoints, node id and directory id.
+     *
+     * @param endpoints endpoints
+     * @param nodeId node id
+     * @param directoryId directory id
+     * @return the voter set
+     */
+    public static VoterSet fromMap(Endpoints endpoints, int nodeId, Uuid directoryId) {
+        Map<Integer, VoterSet.VoterNode> voters = new HashMap<>();
+        voters.put(nodeId, new VoterSet.VoterNode(ReplicaKey.of(nodeId, directoryId), endpoints,
+                new SupportedVersionRange((short) 0, (short) 1)));
+        return new VoterSet(voters);
+    }
 }

raft/src/main/java/org/apache/kafka/snapshot/Snapshots.java

@@ -37,6 +37,8 @@ public final class Snapshots {
     private static final Logger log = LoggerFactory.getLogger(Snapshots.class);
     public static final String SUFFIX = ".checkpoint";
     private static final String PARTIAL_SUFFIX = String.format("%s.part", SUFFIX);
+
+    public static final OffsetAndEpoch BOOTSTRAP_SNAPSHOT_ID = new OffsetAndEpoch(0, 0);
     public static final String DELETE_SUFFIX = String.format("%s.deleted", SUFFIX);
 
     private static final NumberFormat OFFSET_FORMATTER = NumberFormat.getInstance();

raft/src/test/java/org/apache/kafka/raft/QuorumConfigTest.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 org.apache.kafka.raft;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class QuorumConfigTest {
+
+    @Test
+    public void testValidControllerQuorumVoters() {
+        assertTrue(QuorumConfig.validateControllerQuorumVoters("1@hostname:12345"));
+        assertTrue(QuorumConfig.validateControllerQuorumVoters("1-2@hostname:12345")); // with replica directory id
+        assertTrue(QuorumConfig.validateControllerQuorumVoters("[email protected]:54321"));
+        assertTrue(QuorumConfig.validateControllerQuorumVoters("[email protected]:8080")); // with replica directory id
+    }
+
+    @Test
+    public void testInvalidControllerQuorumVoters() {
+        assertFalse(QuorumConfig.validateControllerQuorumVoters("1@hostname:123456")); // Port number too long
+        assertFalse(QuorumConfig.validateControllerQuorumVoters("1@hostname:")); // Missing port number
+        assertFalse(QuorumConfig.validateControllerQuorumVoters("1@hostname")); // Missing port number
+        assertFalse(QuorumConfig.validateControllerQuorumVoters("hostname:12345")); // Missing replica id, @
+        assertFalse(QuorumConfig.validateControllerQuorumVoters("1-2-3@hostname:12345")); // Invalid replica id range
+        assertFalse(QuorumConfig.validateControllerQuorumVoters("1@host name:12345")); // Invalid hostname
+        assertFalse(QuorumConfig.validateControllerQuorumVoters("@hostname:12345")); // Missing replica id
+    }
+}