@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 shouldFailOnMissingMigrationFile() throws Exception {
    // Given:
    final List<String> migratedVersions = ImmutableList.of("1", "2", "3");
    final List<String> migrationFiles = ImmutableList.of("1", "3");
    final List<String> checksums = givenExistingMigrationFiles(migrationFiles);
    givenAppliedMigrations(migratedVersions, ImmutableList.of(checksums.get(0), "missing", checksums.get(1)));

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

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

    // verification stops on failure, so version "1" is never queried
    verifyClientCallsForVersions(ImmutableList.of("2", "3"));
  }