KAFKA-18947 Remove unused raftManager in metadataShell (apache#19169)

* Remove unused `raftManager` in `metadataShell`
* Enhance error message when no snapshot provided.
* Since `raftManager` is removed, make `snapshot` a required argument.

Result when no snapshot is given
```
$ ./bin/kafka-metadata-shell.sh
usage: kafka-metadata-shell [-h] --snapshot SNAPSHOT [command [command ...]]
kafka-metadata-shell: error: argument --snapshot/-s is required
```
```
$ ./bin/kafka-metadata-shell.sh --help
usage: kafka-metadata-shell [-h] --snapshot SNAPSHOT [command [command ...]]

The Apache Kafka metadata shell

positional arguments:
  command                The command to run.

optional arguments:
  -h, --help             show this help message and exit
  --snapshot SNAPSHOT, -s SNAPSHOT
                         The metadata snapshot file to read.
```

Reviewers: Jhen-Yung Hsu <[email protected]>, TengYao Chi <[email protected]>, Chia-Ping Tsai <[email protected]>

Author: mingyen066

shell/src/main/java/org/apache/kafka/shell/MetadataShell.java

@@ -17,14 +17,12 @@
 
 package org.apache.kafka.shell;
 
-import kafka.raft.KafkaRaftManager;
 import kafka.tools.TerseFailure;
 
 import org.apache.kafka.common.utils.Exit;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.image.loader.MetadataLoader;
 import org.apache.kafka.metadata.util.SnapshotFileReader;
-import org.apache.kafka.server.common.ApiMessageAndVersion;
 import org.apache.kafka.server.fault.FaultHandler;
 import org.apache.kafka.server.fault.LoggingFaultHandler;
 import org.apache.kafka.server.util.FileLock;
@@ -60,17 +58,9 @@ public final class MetadataShell {
     private static final Logger log = LoggerFactory.getLogger(MetadataShell.class);
 
     public static class Builder {
-        private KafkaRaftManager<ApiMessageAndVersion> raftManager = null;
         private String snapshotPath = null;
         private FaultHandler faultHandler = new LoggingFaultHandler("shell", () -> { });
 
-        // Note: we assume that we have already taken the lock on the log directory before calling
-        // this method.
-        public Builder setRaftManager(KafkaRaftManager<ApiMessageAndVersion> raftManager) {
-            this.raftManager = raftManager;
-            return this;
-        }
-
         public Builder setSnapshotPath(String snapshotPath) {
             this.snapshotPath = snapshotPath;
             return this;
@@ -82,9 +72,7 @@ public Builder setFaultHandler(FaultHandler faultHandler) {
         }
 
         public MetadataShell build() {
-            return new MetadataShell(raftManager,
-                snapshotPath,
-                faultHandler);
+            return new MetadataShell(snapshotPath, faultHandler);
         }
     }
 
@@ -96,7 +84,7 @@ static File parent(File file) {
         File parent = file.getParentFile();
         return parent == null ? file : parent;
     }
-    
+
     static File parentParent(File file) {
         return parent(parent(file));
     }
@@ -136,8 +124,6 @@ static FileLock takeDirectoryLock(File directory) throws IOException {
 
     private final MetadataShellState state;
 
-    private final KafkaRaftManager<ApiMessageAndVersion> raftManager;
-
     private final String snapshotPath;
 
     private final FaultHandler faultHandler;
@@ -151,29 +137,17 @@ static FileLock takeDirectoryLock(File directory) throws IOException {
     private MetadataLoader loader;
 
     public MetadataShell(
-        KafkaRaftManager<ApiMessageAndVersion> raftManager,
         String snapshotPath,
         FaultHandler faultHandler
     ) {
         this.state = new MetadataShellState();
-        this.raftManager = raftManager;
         this.snapshotPath = snapshotPath;
         this.faultHandler = faultHandler;
         this.publisher = new MetadataShellPublisher(state);
         this.fileLock = null;
         this.snapshotFileReader = null;
     }
 
-    private void initializeWithRaftManager() {
-        raftManager.startup();
-        this.loader = new MetadataLoader.Builder().
-                setFaultHandler(faultHandler).
-                setNodeId(-1).
-                setHighWaterMarkAccessor(() -> raftManager.client().highWatermark()).
-                build();
-        raftManager.register(loader);
-    }
-
     private void initializeWithSnapshotFileReader() throws Exception {
         this.fileLock = takeDirectoryLockIfExists(parentParent(new File(snapshotPath)));
         this.loader = new MetadataLoader.Builder().
@@ -186,18 +160,7 @@ private void initializeWithSnapshotFileReader() throws Exception {
     }
 
     public void run(List<String> args) throws Exception {
-        if (raftManager != null) {
-            if (snapshotPath != null) {
-                throw new RuntimeException("Can't specify both a raft manager and " +
-                        "snapshot file reader.");
-            }
-            initializeWithRaftManager();
-        } else if (snapshotPath != null) {
-            initializeWithSnapshotFileReader();
-        } else {
-            throw new RuntimeException("You must specify either a raft manager or a " +
-                    "snapshot file reader.");
-        }
+        initializeWithSnapshotFileReader();
         loader.installPublishers(Collections.singletonList(publisher)).get(15, TimeUnit.MINUTES);
         if (args == null || args.isEmpty()) {
             // Interactive mode.
@@ -222,14 +185,7 @@ public void run(List<String> args) throws Exception {
 
     public void close() {
         Utils.closeQuietly(loader, "loader");
-        if (raftManager != null) {
-            try {
-                raftManager.shutdown();
-            } catch (Exception e) {
-                log.error("Error shutting down RaftManager", e);
-            }
-        }
-        Utils.closeQuietly(snapshotFileReader, "raftManager");
+        Utils.closeQuietly(snapshotFileReader, "snapshotFileReader");
         if (fileLock != null) {
             try {
                 fileLock.destroy();
@@ -248,7 +204,8 @@ public static void main(String[] args) {
             .description("The Apache Kafka metadata shell");
         parser.addArgument("--snapshot", "-s")
             .type(String.class)
-            .help("The snapshot file to read.");
+            .required(true)
+            .help("The metadata snapshot file to read.");
         parser.addArgument("command")
             .nargs("*")
             .help("The command to run.");

shell/src/test/java/org/apache/kafka/shell/MetadataShellIntegrationTest.java

@@ -84,7 +84,7 @@ public void close() {
     @ValueSource(booleans = {false, true})
     public void testLock(boolean canLock) throws Exception {
         try (IntegrationEnv env = new IntegrationEnv()) {
-            env.shell = new MetadataShell(null,
+            env.shell = new MetadataShell(
                 new File(new File(env.tempDir, "__cluster_metadata-0"), "00000000000122906351-0000000226.checkpoint").getAbsolutePath(),
                     env.faultHandler);