@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,
        getMigrationsDir(getConfigFile(), config)
    );
  }
@Test
  public void shouldPrintInfo() throws Exception {
    // Given:
    givenMigrations(
        ImmutableList.of("1", "3"),
        ImmutableList.of(MIGRATED, ERROR),
        ImmutableList.of("N/A", "error reason"),
        ImmutableList.of("4"));

    // When:
    final int result = command.command(config, cfg -> ksqlClient, migrationsDir);

    // Then:
    assertThat(result, is(0));

    verify(logAppender, atLeastOnce()).doAppend(logCaptor.capture());
    final List<String> logMessages = logCaptor.getAllValues().stream()
        .map(LoggingEvent::getRenderedMessage)
        .collect(Collectors.toList());
    assertThat(logMessages, hasItem(containsString("Current migration version: 3")));
    assertThat(logMessages, hasItem(containsString(
        " Version | Name        | State    | Previous Version | Started On | Completed On | Error Reason \n" +
            "------------------------------------------------------------------------------------------------\n" +
            " 1       | some_name_1 | MIGRATED | <none>           | N/A        | N/A          | N/A          \n" +
            " 3       | some_name_3 | ERROR    | 1                | N/A        | N/A          | error reason \n" +
            " 4       | some name 4 | PENDING  | N/A              | N/A        | N/A          | N/A          \n" +
            "------------------------------------------------------------------------------------------------"
    )));
  }