public static String getMaskedStatement(final String query) {
    try {
      final ParseTree tree = DefaultKsqlParser.getParseTree(query);
      return new Visitor().visit(tree);
    } catch (final Exception | StackOverflowError e) {
      return fallbackMasking(query);
    }
  }
@Test
  public void shouldMaskInvalidCreateStreamWithQuotes() {
    // Given:
    // Typo in "WITH" => "WIT"
    final String query = "CREATE STREAM `stream` (id varchar) WIT ('format' = 'avro', \"kafka_topic\" = 'test_topic', partitions=3);";

    // When
    final String maskedQuery = QueryMask.getMaskedStatement(query);

    // Then: we are replacing key which is not in ALLOWED_KEYS if the statement is invalid and we do regex matching
    final String expected  = "CREATE STREAM `stream` (id varchar) WIT ('format'='[string]', \"kafka_topic\"='[string]', partitions=3);";
    assertThat(maskedQuery, is(expected));
  }