public static StatementExecutorResponse execute(
      final ConfiguredStatement<ListProperties> statement,
      final SessionProperties sessionProperties,
      final KsqlExecutionContext executionContext,
      final ServiceContext serviceContext
  ) {
    final KsqlConfigResolver resolver = new KsqlConfigResolver();

    final Map<String, String> engineProperties = statement
        .getSessionConfig()
        .getConfig(false)
        .getAllConfigPropsWithSecretsObfuscated();

    final List<Property> mergedProperties = mergedProperties(statement);

    final List<String> overwritten = mergedProperties
        .stream()
        .filter(property -> !Objects.equals(
            engineProperties.get(property.getName()), property.getValue()))
        .map(Property::getName)
        .collect(Collectors.toList());

    final List<String> defaultProps = mergedProperties.stream()
        .filter(property -> resolver.resolve(property.getName(), false)
            .map(resolved -> resolved.isDefaultValue(property.getValue()))
            .orElse(false))
        .map(Property::getName)
        .collect(Collectors.toList());

    return StatementExecutorResponse.handled(Optional.of(new PropertiesList(
        statement.getMaskedStatementText(), mergedProperties, overwritten, defaultProps)));
  }
@Test
  public void shouldNotListUnrecognizedConnectProps() throws Exception {
    // Given:
    givenConnectWorkerProperties(
        "group.id=list_properties_unit_test\n"
            + "key.converter=io.confluent.connect.avro.AvroConverter\n"
            + "value.converter=io.confluent.connect.avro.AvroConverter\n"
            + "offset.storage.topic=topic1\n"
            + "config.storage.topic=topic2\n"
            + "status.storage.topic=topic3\n"
            + "other.config=<potentially sensitive data that should not be shown>\n"
            + "sasl.jaas.config=<potentially sensitive data that should not be shown even though it's a recognized config>\n"
    );

    // When:
    final PropertiesList properties = (PropertiesList) CustomExecutors.LIST_PROPERTIES.execute(
        engine.configure("LIST PROPERTIES;")
            .withConfig(new KsqlConfig(ImmutableMap.of(
                "ksql.connect.worker.config", connectPropsFile))),
        mock(SessionProperties.class),
        engine.getEngine(),
        engine.getServiceContext()
    ).getEntity().orElseThrow(IllegalStateException::new);

    // Then:
    assertThat(
        properties.getProperties(),
        hasItem(new Property("ksql.connect.worker.config", "KSQL", connectPropsFile)));
    assertThat(
        properties.getProperties(),
        hasItem(new Property("value.converter", "EMBEDDED CONNECT WORKER", "io.confluent.connect.avro.AvroConverter")));
    assertThat(toMap(properties), not(hasKey("other.config")));
    assertThat(toMap(properties), not(hasKey("sasl.jaas.config")));
  }