@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),
            Clock.systemDefaultZone()
    );
  }
@Test
  public void shouldApplyAssertTopicCommands() throws Exception {
    command = PARSER.parse("-v", "3");
    createMigrationFile(1, NAME, migrationsDir, COMMAND);
    createMigrationFile(3, NAME, migrationsDir, ASSERT_TOPIC_COMMANDS);
    givenCurrentMigrationVersion("1");
    givenAppliedMigration(1, NAME, MigrationState.MIGRATED);

    // When:
    final int result = command.command(config, (cfg, headers) -> ksqlClient, migrationsDir, Clock.fixed(
        Instant.ofEpochMilli(1000), ZoneId.systemDefault()));

    // Then:
    assertThat(result, is(0));
    final InOrder inOrder = inOrder(ksqlClient);

    verifyMigratedVersion(inOrder, 3, "1", MigrationState.MIGRATED,
        () -> {
          inOrder.verify(ksqlClient).assertTopic("abc", ImmutableMap.of(), true);
          inOrder.verify(ksqlClient).assertTopic("abc", ImmutableMap.of(), false, Duration.ofSeconds(10));
          inOrder.verify(ksqlClient).assertTopic("abc", ImmutableMap.of("FOO", 4, "BAR", 6), false);
          inOrder.verify(ksqlClient).assertTopic("abc", ImmutableMap.of("FOO", 4, "BAR", 6), true, Duration.ofSeconds(10));
        });
    inOrder.verify(ksqlClient).close();
    inOrder.verifyNoMoreInteractions();
  }