feat: new command to restore ksqlDB command topic backups

bin/ksql-restore-command-topic

@@ -0,0 +1,28 @@
+#!/bin/bash
+# (Copyright) [2020] Confluent, Inc.
+
+base_dir=$( cd "$( dirname "${BASH_SOURCE[0]}" )/.." && pwd )
+
+: "${KSQL_CONFIG_DIR:="$base_dir/config"}"
+
+# logj4 settings
+if [ -z "$KSQL_LOG4J_OPTS" ]; then
+  # Test for files from dev -> packages so this will work as expected in dev if you have packages
+  # installed
+  if [ -e "$base_dir/config/log4j.properties" ]; then # Dev environment
+    KSQL_CONFIG_DIR="$base_dir/config"
+  elif [ -e "$base_dir/etc/ksqldb/log4j.properties" ]; then # Simple zip file layout
+    KSQL_CONFIG_DIR="$base_dir/etc/ksqldb"
+  elif [ -e "/etc/ksqldb/log4j.properties" ]; then # Normal install layout
+    KSQL_CONFIG_DIR="/etc/ksqldb"
+  fi
+fi
+
+: "${KSQL_LOG4J_OPTS:=""}"
+
+# Use file logging by default
+if [ -z "$KSQL_LOG4J_OPTS" ]; then
+  export KSQL_LOG4J_OPTS="-Dlog4j.configuration=file:$KSQL_CONFIG_DIR/log4j-file.properties"
+fi
+
+exec "$base_dir"/bin/ksql-run-class io.confluent.ksql.rest.server.restore.KsqlRestoreCommandTopic "$@"

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/BackupReplayFile.java

@@ -44,9 +44,22 @@ public final class BackupReplayFile implements Closeable {
   private final File file;
   private final FileOutputStream writer;
 
-  public BackupReplayFile(final File file) {
+  public static BackupReplayFile readOnly(final File file) {
+    return new BackupReplayFile(file, false);
+  }
+
+  public static BackupReplayFile writable(final File file) {
+    return new BackupReplayFile(file, true);
+  }
+
+  private BackupReplayFile(final File file, final boolean write) {
     this.file = Objects.requireNonNull(file, "file");
-    this.writer = createWriter(file);
+
+    if (write) {
+      this.writer = createWriter(file);
+    } else {
+      this.writer = null;
+    }
   }
 
   private static FileOutputStream createWriter(final File file) {
@@ -58,11 +71,19 @@ private static FileOutputStream createWriter(final File file) {
     }
   }
 
+  public File getFile() {
+    return file;
+  }
+
   public String getPath() {
     return file.getAbsolutePath();
   }
 
   public void write(final ConsumerRecord<byte[], byte[]> record) throws IOException {
+    if (writer == null) {
+      throw new IOException("Write permission denied.");
+    }
+
     writer.write(record.key());
     writer.write(KEY_VALUE_SEPARATOR_BYTES);
     writer.write(record.value());
@@ -72,7 +93,7 @@ public void write(final ConsumerRecord<byte[], byte[]> record) throws IOExceptio
 
   public List<Pair<byte[], byte[]>> readRecords() throws IOException {
     final List<Pair<byte[], byte[]>> commands = new ArrayList<>();
-    for (final String line : Files.readAllLines(file.toPath(), StandardCharsets.UTF_8)) {
+    for (final String line : Files.readAllLines(getFile().toPath(), StandardCharsets.UTF_8)) {
       final String commandId = line.substring(0, line.indexOf(KEY_VALUE_SEPARATOR_STR));
       final String command = line.substring(line.indexOf(KEY_VALUE_SEPARATOR_STR) + 1);
 
@@ -87,6 +108,8 @@ public List<Pair<byte[], byte[]>> readRecords() throws IOException {
 
   @Override
   public void close() throws IOException {
-    writer.close();
+    if (writer != null) {
+      writer.close();
+    }
   }
 }

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/CommandTopicBackupImpl.java

@@ -202,7 +202,7 @@ BackupReplayFile openOrCreateReplayFile() {
   }
 
   private BackupReplayFile newReplayFile() {
-    return new BackupReplayFile(Paths.get(
+    return BackupReplayFile.writable(Paths.get(
         backupLocation.getAbsolutePath(),
         String.format("%s%s_%s", PREFIX, topicName, ticker.read())
     ).toFile());
@@ -236,7 +236,7 @@ private Optional<BackupReplayFile> latestReplayFile() {
     }
 
     return (latestBakFile != null)
-        ? Optional.of(new BackupReplayFile(latestBakFile))
+        ? Optional.of(BackupReplayFile.writable(latestBakFile))
         : Optional.empty();
   }