@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 shouldFallBackToSingleKeyPullQueriesIfNeeded() throws Exception {
    // Given:
    givenMigrations(
        ImmutableList.of("1", "3"),
        ImmutableList.of(MIGRATED, MIGRATED),
        ImmutableList.of("N/A", "N/A"),
        ImmutableList.of(),
        false);

    when(serverInfo.getServerVersion()).thenReturn("v0.13.0");

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

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

    verify(ksqlClient).executeQuery(
        "SELECT version, checksum, previous, state, name, started_on, completed_on, error_reason "
            + "FROM " + MIGRATIONS_TABLE + " WHERE version_key = '1';");
    verify(ksqlClient).executeQuery(
        "SELECT version, checksum, previous, state, name, started_on, completed_on, error_reason "
            + "FROM " + MIGRATIONS_TABLE + " WHERE version_key = '3';");
  }