@SuppressWarnings("OptionalGetWithoutIsPresent")
  public static StatementExecutorResponse execute(
      final ConfiguredStatement<ListConnectors> configuredStatement,
      final SessionProperties sessionProperties,
      final KsqlExecutionContext ksqlExecutionContext,
      final ServiceContext serviceContext
  ) {
    final ConnectClient connectClient = serviceContext.getConnectClient();
    final ConnectResponse<List<String>> connectors = serviceContext.getConnectClient().connectors();
    if (connectors.error().isPresent()) {
      final String errorMsg = "Failed to list connectors: " + connectors.error().get();
      throw new KsqlRestException(EndpointResponse.create()
          .status(connectors.httpCode())
          .entity(new KsqlErrorMessage(Errors.toErrorCode(connectors.httpCode()), errorMsg))
          .build()
      );
    }

    final List<SimpleConnectorInfo> infos = new ArrayList<>();
    final List<KsqlWarning> warnings = new ArrayList<>();
    final Scope scope = configuredStatement.getStatement().getScope();

    for (final String name : connectors.datum().get()) {
      final ConnectResponse<ConnectorInfo> response = connectClient.describe(name);

      if (response.datum().filter(i -> inScope(i.type(), scope)).isPresent()) {
        final ConnectResponse<ConnectorStateInfo> status = connectClient.status(name);
        infos.add(fromConnectorInfoResponse(name, response, status));
      } else if (response.error().isPresent()) {
        if (scope == Scope.ALL) {
          infos.add(new SimpleConnectorInfo(name, ConnectorType.UNKNOWN, null, null));
        }

        warnings.add(
            new KsqlWarning(
                String.format(
                    "Could not describe connector %s: %s",
                    name,
                    response.error().get())));
      }
    }

    return StatementExecutorResponse.handled(Optional.of(
        new ConnectorList(
            configuredStatement.getMaskedStatementText(),
            warnings,
            infos)
    ));
  }
@Test
  public void shouldFilterNonMatchingConnectors() {
    // Given:
    when(connectClient.connectors())
        .thenReturn(ConnectResponse.success(ImmutableList.of("connector", "connector2"),
            HttpStatus.SC_OK));
    final ConfiguredStatement<ListConnectors> statement = ConfiguredStatement
        .of(PreparedStatement.of("", new ListConnectors(Optional.empty(), Scope.SINK)),
            SessionConfig.of(new KsqlConfig(ImmutableMap.of()), ImmutableMap.of())
        );

    // When:
    final Optional<KsqlEntity> entity = ListConnectorsExecutor
        .execute(statement, mock(SessionProperties.class), engine, serviceContext).getEntity();

    // Then:
    assertThat("expected response!", entity.isPresent());
    final ConnectorList connectorList = (ConnectorList) entity.get();

    assertThat(connectorList, is(new ConnectorList(
        "",
        ImmutableList.of(),
        ImmutableList.of()
    )));
  }