@SuppressWarnings("OptionalGetWithoutIsPresent")
  public StatementExecutorResponse execute(
      final ConfiguredStatement<DescribeConnector> configuredStatement,
      final SessionProperties sessionProperties,
      final KsqlExecutionContext ksqlExecutionContext,
      final ServiceContext serviceContext
  ) {
    final String connectorName = configuredStatement
        .getStatement()
        .getConnectorName();

    final ConnectResponse<ConnectorStateInfo> statusResponse = serviceContext
        .getConnectClient()
        .status(connectorName);
    if (statusResponse.error().isPresent()) {
      final String errorMsg = "Failed to query connector status: " + statusResponse.error().get();
      throw new KsqlRestException(EndpointResponse.create()
          .status(statusResponse.httpCode())
          .entity(new KsqlErrorMessage(Errors.toErrorCode(statusResponse.httpCode()), errorMsg))
          .build()
      );
    }

    final ConnectResponse<ConnectorInfo> infoResponse = serviceContext
        .getConnectClient()
        .describe(connectorName);
    if (infoResponse.error().isPresent()) {
      final String errorMsg = "Failed to describe connector: " + infoResponse.error().get();
      throw new KsqlRestException(EndpointResponse.create()
          .status(infoResponse.httpCode())
          .entity(new KsqlErrorMessage(Errors.toErrorCode(infoResponse.httpCode()), errorMsg))
          .build()
      );
    }

    final ConnectorStateInfo status = statusResponse.datum().get();
    final ConnectorInfo info = infoResponse.datum().get();
    final Optional<Connector> connector = connectorFactory.apply(info);
    final List<KsqlWarning> warnings;
    final List<String> topics;
    if (connector.isPresent()) {
      // Small optimization. If a connector's info is not found in the response, don't query for
      // active topics with the given connectorName
      final ConnectResponse<Map<String, Map<String, List<String>>>> topicsResponse = serviceContext
          .getConnectClient()
          .topics(connectorName);
      // topics endpoint is relatively new (KAFKA-9422), so 404 here is expected behavior for older
      // Connect versions. Rather than showing a scary warning to the user, we just log it to the
      // server logs.
      if (topicsResponse.error().isPresent()
          && topicsResponse.httpCode() == HttpStatus.SC_NOT_FOUND) {
        topics = ImmutableList.of();
        warnings = ImmutableList.of();
        LOG.warn("Could not list related topics due to error: " + topicsResponse.error().get());
      } else if (topicsResponse.error().isPresent()) {
        topics = ImmutableList.of();
        warnings = ImmutableList.of(
            new KsqlWarning("Could not list related topics due to error: "
                + topicsResponse.error().get()));
      } else {
        topics = topicsResponse.datum()
            .get()
            .get(connectorName)
            .getOrDefault(TOPICS_KEY, ImmutableList.of());
        warnings = ImmutableList.of();
      }
    } else {
      topics = ImmutableList.of();
      warnings = ImmutableList.of();
    }

    final List<SourceDescription> sources;
    if (connector.isPresent()) {
      sources = ksqlExecutionContext
          .getMetaStore()
          .getAllDataSources()
          .values()
          .stream()
          .filter(source -> topics.contains(source.getKafkaTopicName()))
          .map(source -> SourceDescriptionFactory.create(
                  source,
                  false,
                  ImmutableList.of(),
                  ImmutableList.of(),
                  Optional.empty(),
                  ImmutableList.of(),
                  ImmutableList.of(),
                  ksqlExecutionContext.metricCollectors()
              )
          )
          .collect(Collectors.toList());
    } else {
      sources = ImmutableList.of();
    }

    final ConnectorDescription description = new ConnectorDescription(
        configuredStatement.getMaskedStatementText(),
        info.config().get(CONNECTOR_CLASS_CONFIG),
        status,
        sources,
        topics,
        warnings
    );

    return StatementExecutorResponse.handled(Optional.of(description));
  }
@Test
  public void shouldNotWarnClientOnMissingTopicsEndpoint() {
    // Given:
    when(connectClient.topics(any())).thenReturn(ConnectResponse.failure("not found",
        HttpStatus.SC_NOT_FOUND));

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

    // Then:
    verify(engine).getMetaStore();
    verify(metaStore).getAllDataSources();
    verify(connectClient).status("connector");
    verify(connectClient).describe("connector");
    verify(connectClient).topics("connector");
    assertThat("Expected a response", entity.isPresent());
    assertThat(entity.get(), instanceOf(ConnectorDescription.class));

    final ConnectorDescription description = (ConnectorDescription) entity.get();
    assertThat(description.getTopics(), empty());
    assertThat(description.getWarnings(), empty());
  }