feat: new CLI parameter (-f,--file) to execute commands from a file a…

…nd exit (#6440)

ksqldb-cli/src/main/java/io/confluent/ksql/Ksql.java

@@ -23,6 +23,7 @@
 import io.confluent.ksql.rest.client.BasicCredentials;
 import io.confluent.ksql.rest.client.KsqlRestClient;
 import io.confluent.ksql.util.ErrorMessageUtil;
+import io.confluent.ksql.util.KsqlException;
 import io.confluent.ksql.version.metrics.KsqlVersionCheckerAgent;
 import io.confluent.ksql.version.metrics.collector.KsqlModuleType;
 import java.io.Console;
@@ -112,8 +113,15 @@ void run() {
           options.getOutputFormat(),
           restClient)
       ) {
-        if (options.getExecute() != null) {
-          cli.runCommand(options.getExecute());
+        if (options.getExecute().isPresent()) {
+          cli.runCommand(options.getExecute().get());
+        } else if (options.getScriptFile().isPresent()) {
+          final File scriptFile = new File(options.getScriptFile().get());
+          if (scriptFile.exists() && scriptFile.isFile()) {
+            cli.runScript(scriptFile.getPath());
+          } else {
+            throw new KsqlException("No such script file: " + scriptFile.getPath());
+          }
         } else {
           cli.runInteractively();
         }

ksqldb-cli/src/main/java/io/confluent/ksql/cli/Cli.java

@@ -21,6 +21,7 @@
 import io.confluent.ksql.cli.console.cmd.CliCommandRegisterUtil;
 import io.confluent.ksql.cli.console.cmd.RemoteServerSpecificCommand;
 import io.confluent.ksql.cli.console.cmd.RequestPipeliningCommand;
+import io.confluent.ksql.cli.console.cmd.RunScript;
 import io.confluent.ksql.parser.DefaultKsqlParser;
 import io.confluent.ksql.parser.KsqlParser;
 import io.confluent.ksql.parser.KsqlParser.ParsedStatement;
@@ -53,6 +54,7 @@
 import io.vertx.core.VertxException;
 import java.io.Closeable;
 import java.io.PrintWriter;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -182,6 +184,23 @@ private <R> RestResponse<R> makeKsqlRequest(
     throw new KsqlRestClientException("Failed to execute request " + ksql);
   }
 
+  public void runScript(final String scriptFile) {
+    RemoteServerSpecificCommand.validateClient(terminal.writer(), restClient);
+
+    try {
+      final RunScript runScriptCommand = RunScript.create(this);
+      runScriptCommand.execute(Collections.singletonList(scriptFile), terminal.writer());
+    } catch (final Exception exception) {
+      LOGGER.error("An error occurred while running a script file. Error = "
+          + exception.getMessage(), exception);
+
+      terminal.printError(ErrorMessageUtil.buildErrorMessage(exception),
+          exception.toString());
+    }
+
+    terminal.flush();
+  }
+
   public void runCommand(final String command) {
     RemoteServerSpecificCommand.validateClient(terminal.writer(), restClient);
     try {

ksqldb-cli/src/main/java/io/confluent/ksql/cli/Options.java

@@ -43,6 +43,8 @@ public class Options {
   private static final String OUTPUT_FORMAT_OPTION_NAME = "--output";
   private static final String EXECUTE_OPTION = "--execute";
   private static final String EXECUTE_SHORT_OPTION = "-e";
+  private static final String FILE_OPTION = "--file";
+  private static final String FILE_SHORT_OPTION = "-f";
 
   // Only here so that the help message generated by Help.help() is accurate
   @Inject
@@ -119,6 +121,12 @@ public class Options {
       description = "Execute one or more SQL statements and quit.")
   private String execute = null;
 
+  @SuppressWarnings("unused") // Accessed via reflection
+  @Option(
+      name = {FILE_OPTION, FILE_SHORT_OPTION},
+      description = "Execute commands from a file and exit.")
+  private String scriptFile = null;
+
   public static Options parse(final String...args) throws IOException {
     final SingleCommand<Options> optionsParser = SingleCommand.singleCommand(Options.class);
 
@@ -198,13 +206,17 @@ public Optional<BasicCredentials> getUserNameAndPassword() {
     return Optional.of(BasicCredentials.of(userName, password));
   }
 
-  public String getExecute() {
+  public Optional<String> getExecute() {
     if (execute == null || execute.isEmpty()) {
-      return execute;
+      return Optional.empty();
     }
 
     // Append a colon if not specified
     final char lastChar = execute.charAt(execute.length() - 1);
-    return (lastChar != ';') ? execute + ";" : execute;
+    return Optional.of((lastChar != ';') ? execute + ";" : execute);
+  }
+
+  public Optional<String> getScriptFile() {
+    return Optional.of(scriptFile);
   }
 }

ksqldb-cli/src/main/java/io/confluent/ksql/cli/console/cmd/RunScript.java

@@ -26,7 +26,7 @@
 import java.util.Objects;
 import java.util.stream.Collectors;
 
-final class RunScript implements CliSpecificCommand {
+public final class RunScript implements CliSpecificCommand {
 
   private static final String HELP = "run script <path_to_sql_file>:" + System.lineSeparator()
       + "\tLoad and run the statements in the supplied file." + System.lineSeparator()
@@ -38,7 +38,7 @@ private RunScript(final KsqlRequestExecutor requestExecutor) {
     this.requestExecutor = Objects.requireNonNull(requestExecutor, "requestExecutor");
   }
 
-  static RunScript create(final KsqlRequestExecutor requestExecutor) {
+  public static RunScript create(final KsqlRequestExecutor requestExecutor) {
     return new RunScript(requestExecutor);
   }
 

ksqldb-cli/src/test/java/io/confluent/ksql/KsqlTest.java

@@ -30,6 +30,7 @@
 import io.confluent.ksql.cli.console.OutputFormat;
 import io.confluent.ksql.rest.client.KsqlRestClient;
 import java.io.File;
+import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.util.Optional;
@@ -84,7 +85,7 @@ public void shouldRunInteractively() {
   @Test
   public void shouldRunNonInteractiveCommandWhenExecuteOptionIsUsed() {
     // Given:
-    when(options.getExecute()).thenReturn("this is a command");
+    when(options.getExecute()).thenReturn(Optional.of("this is a command"));
 
     // When:
     ksql.run();
@@ -93,6 +94,19 @@ public void shouldRunNonInteractiveCommandWhenExecuteOptionIsUsed() {
     verify(cli).runCommand("this is a command");
   }
 
+  @Test
+  public void shouldRunScriptFileWhenFileOptionIsUsed() throws IOException {
+    // Given:
+    final String sqlFile = TMP.newFile().getAbsolutePath();
+    when(options.getScriptFile()).thenReturn(Optional.of(sqlFile));
+
+    // When:
+    ksql.run();
+
+    // Then:
+    verify(cli).runScript(sqlFile);
+  }
+
   @Test
   public void shouldBuildClientWithCorrectServerAddress() {
     // Given:

ksqldb-cli/src/test/java/io/confluent/ksql/cli/CliTest.java

@@ -764,6 +764,20 @@ public void shouldPrintErrorIfCantConnectToRestServerOnRunCommand() throws Excep
         containsString("Please ensure that the URL provided is for an active KSQL server."));
   }
 
+  @Test
+  public void shouldPrintErrorIfCantConnectToRestServerOnRunScript() throws Exception {
+    // Given
+    final KsqlRestClient mockRestClient = givenMockRestClient();
+    when(mockRestClient.getServerInfo())
+        .thenThrow(new KsqlRestClientException("Boom", new IOException("")));
+
+    new Cli(1L, 1L, mockRestClient, console)
+        .runScript("script_file_ignored");
+
+    assertThat(terminal.getOutputString(),
+        containsString("Please ensure that the URL provided is for an active KSQL server."));
+  }
+
   @Test
   public void shouldRegisterRemoteCommand() {
     assertThat(console.getCliSpecificCommands().get("server"),
@@ -1000,6 +1014,22 @@ public void shouldRunScriptOnRunCommand() throws Exception {
         containsString("Created query with ID CSAS_SHOULDRUNSCRIPT"));
   }
 
+  @Test
+  public void shouldRunScriptOnRunScript() throws Exception {
+    // Given:
+    final File scriptFile = TMP.newFile("script.sql");
+    Files.write(scriptFile.toPath(), (""
+        + "CREATE STREAM shouldRunScript AS SELECT * FROM " + ORDER_DATA_PROVIDER.sourceName() + ";"
+        + "").getBytes(StandardCharsets.UTF_8));
+
+    // When:
+    localCli.runScript(scriptFile.getPath());
+
+    // Then:
+    assertThat(terminal.getOutputString(),
+        containsString("Created query with ID CSAS_SHOULDRUNSCRIPT"));
+  }
+
   @Test
   public void shouldUpdateCommandSequenceNumber() throws Exception {
     // Given: