@SuppressFBWarnings(value = "DMI_RANDOM_USED_ONLY_ONCE")
  public static LocalCommands open(
      final KsqlEngine ksqlEngine,
      final File directory
  ) {
    if (!directory.exists()) {
      if (!directory.mkdirs()) {
        throw new KsqlServerException("Couldn't create the local commands directory: "
            + directory.getPath()
            + "\n Make sure the directory exists and is readable/writable for KSQL server "
            + "\n or its parent directory is readable/writable by KSQL server"
            + "\n or change it to a readable/writable directory by setting '"
            + KsqlRestConfig.KSQL_LOCAL_COMMANDS_LOCATION_CONFIG
            + "' config in the properties file."
        );
      }

      try {
        Files.setPosixFilePermissions(directory.toPath(),
            PosixFilePermissions.fromString("rwx------"));
      } catch (final IOException e) {
        throw new KsqlServerException(String.format(
            "Couldn't set POSIX permissions on the backups directory: %s. Error = %s",
            directory.getPath(), e.getMessage()));
      }
    }

    if (!directory.isDirectory()) {
      throw new KsqlServerException(directory.getPath()
          + " is not a directory."
          + "\n Make sure the directory exists and is readable/writable for KSQL server "
          + "\n or its parent directory is readable/writable by KSQL server"
          + "\n or change it to a readable/writable directory by setting '"
          + KsqlRestConfig.KSQL_LOCAL_COMMANDS_LOCATION_CONFIG
          + "' config in the properties file."
      );
    }

    if (!directory.canWrite() || !directory.canRead() || !directory.canExecute()) {
      throw new KsqlServerException("The local commands directory is not readable/writable "
          + "for KSQL server: "
          + directory.getPath()
          + "\n Make sure the directory exists and is readable/writable for KSQL server "
          + "\n or change it to a readable/writable directory by setting '"
          + KsqlRestConfig.KSQL_LOCAL_COMMANDS_LOCATION_CONFIG
          + "' config in the properties file."
      );
    }
    final File file = new File(directory, String.format("local_commands_%d_%s%s",
        System.currentTimeMillis(), Integer.toHexString(RANDOM.nextInt()),
        LOCAL_COMMANDS_FILE_SUFFIX));
    return new LocalCommands(directory, ksqlEngine, LocalCommandsFile.createWriteable(file));
  }
@Test
  public void shouldThrowWhenCommandLocationIsNotDirectory() throws IOException {
    // Given
    File file = commandsDir.newFile();

    // When
    final Exception e = assertThrows(
        KsqlServerException.class,
        () -> LocalCommands.open(ksqlEngine, file)
    );

    // Then
    assertThat(e.getMessage(), containsString(String.format(
        "%s is not a directory.",
        file.getAbsolutePath()
    )));
  }