public static ParsedCommand parse(
          // CHECKSTYLE_RULES.ON: CyclomaticComplexity
          final String sql, final Map<String, String> variables) {

    validateSupportedStatementType(sql);
    final String substituted;
    try {
      substituted = VariableSubstitutor.substitute(KSQL_PARSER.parse(sql).get(0), variables);
    } catch (ParseFailedException e) {
      throw new MigrationException(String.format(
              "Failed to parse the statement. Statement: %s. Reason: %s",
              sql, e.getMessage()));
    }
    final SqlBaseParser.SingleStatementContext statementContext = KSQL_PARSER.parse(substituted)
            .get(0).getStatement();
    final boolean isStatement = StatementType.get(statementContext.statement().getClass())
            == StatementType.STATEMENT;
    return new ParsedCommand(substituted,
            isStatement ? Optional.empty() : Optional.of(new AstBuilder(TypeRegistry.EMPTY)
            .buildStatement(statementContext)));
  }
@Test
  public void shouldParseCreateConnectorStatementWithVariables() {
    // Given:
    final String createConnector = "CREATE SOURCE CONNECTOR ${name} WITH(\n"
        + "    \"connector.class\"='io.confluent.connect.jdbc.JdbcSourceConnector',\n"
        + "    \"connection.url\"=${url},\n"
        + "    \"mode\"='bulk',\n"
        + "    \"topic.prefix\"='jdbc-',\n"
        + "    \"table.whitelist\"='users',\n"
        + "    \"key\"='username');";


            // When:
    List<CommandParser.ParsedCommand> commands = parse(createConnector, ImmutableMap.of("url", "'jdbc:postgresql://localhost:5432/my.db'", "name", "`jdbc_connector`"));

    // Then:
    assertThat(commands.size(), is(1));
    assertThat(commands.get(0).getStatement().isPresent(), is (true));
    assertThat(commands.get(0).getStatement().get(), instanceOf(CreateConnector.class));
    assertThat(commands.get(0).getCommand(), is("CREATE SOURCE CONNECTOR `jdbc_connector` WITH(\n"
            + "    \"connector.class\"='io.confluent.connect.jdbc.JdbcSourceConnector',\n"
            + "    \"connection.url\"='jdbc:postgresql://localhost:5432/my.db',\n"
            + "    \"mode\"='bulk',\n"
            + "    \"topic.prefix\"='jdbc-',\n"
            + "    \"table.whitelist\"='users',\n"
            + "    \"key\"='username');"));
    assertThat(((CreateConnector) commands.get(0).getStatement().get()).getName(), is("jdbc_connector"));
    assertThat(((CreateConnector) commands.get(0).getStatement().get()).getType() == CreateConnector.Type.SOURCE, is(true));
    assertThat(((CreateConnector) commands.get(0).getStatement().get()).getConfig().size(), is(6));
    assertThat(((CreateConnector) commands.get(0).getStatement().get()).getConfig().get("connector.class").getValue(), is("io.confluent.connect.jdbc.JdbcSourceConnector"));
    assertThat(((CreateConnector) commands.get(0).getStatement().get()).getConfig().get("connection.url").getValue(), is("jdbc:postgresql://localhost:5432/my.db"));
    assertThat(((CreateConnector) commands.get(0).getStatement().get()).getConfig().get("mode").getValue(), is("bulk"));
    assertThat(((CreateConnector) commands.get(0).getStatement().get()).getConfig().get("topic.prefix").getValue(), is("jdbc-"));
    assertThat(((CreateConnector) commands.get(0).getStatement().get()).getConfig().get("table.whitelist").getValue(), is("users"));
    assertThat(((CreateConnector) commands.get(0).getStatement().get()).getConfig().get("key").getValue(), is("username"));
    assertThat(((CreateConnector) commands.get(0).getStatement().get()).ifNotExists(), is(false));
  }