@Override
  protected int command() {
    if (!validateConfigFilePresent()) {
      return 1;
    }

    final MigrationConfig config;
    try {
      config = MigrationConfig.load(getConfigFile());
    } catch (KsqlException | MigrationException e) {
      LOGGER.error(e.getMessage());
      return 1;
    }

    return command(config, MigrationsUtil::getKsqlClient);
  }
@Test
  public void shouldNotCreateTableIfFailedToCreateStream() throws Exception {
    // Given:
    when(executeStatementCf.get()).thenThrow(new ExecutionException(new RuntimeException("failed")));

    // When:
    final int status = command.command(config, cfg -> client);

    // Then:
    assertThat(status, is(1));

    verify(client, never()).executeStatement(EXPECTED_CTAS_STATEMENT);
  }