@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()
    );
  }
@SuppressWarnings("unchecked")
  @Test
  public void shouldApplyDefineUndefineCommands() throws Exception {
    // Given:
    final Map<String, Object> variables = ImmutableMap.of("pre", "a", "str", "abc");
    command = PARSER.parse("-n");
    createMigrationFile(1, NAME, migrationsDir, DEFINE_COMMANDS);
    when(versionQueryResult.get()).thenReturn(ImmutableList.of());
    when(ksqlClient.getVariables()).thenReturn(
        ImmutableMap.of(), ImmutableMap.of(), variables, variables, variables, variables, variables,
        variables, variables, variables, variables, variables, variables, variables, variables,
        variables, variables, ImmutableMap.of()
    );

    // 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, 1, "<none>", MigrationState.MIGRATED, () -> {
      inOrder.verify(ksqlClient).executeStatement(COMMAND, new HashMap<>());
      inOrder.verify(ksqlClient).define("pre", "a");
      inOrder.verify(ksqlClient).define("str", "abc");
      inOrder.verify(ksqlClient).executeStatement(eq("CREATE STREAM abc AS SELECT * FROM FOO;"), propCaptor.capture());
      assertThat(propCaptor.getValue().size(), is(1));
      assertThat(propCaptor.getValue().get("abc"), is("yay"));
      inOrder.verify(ksqlClient).insertInto("`FOO`", new KsqlObject(ImmutableMap.of("`A`", "abc")));
      inOrder.verify(ksqlClient).undefine("str");
      inOrder.verify(ksqlClient).insertInto("`FOO`", new KsqlObject(ImmutableMap.of("`A`", "${str}")));
    });
    inOrder.verify(ksqlClient).close();
    inOrder.verifyNoMoreInteractions();
  }