@Override
  public ConnectResponse<Map<String, Map<String, List<String>>>> topics(final String connector) {
    try {
      LOG.debug("Issuing request to Kafka Connect at URI {} to get active topics for {}",
          connectUri, connector);

      final ConnectResponse<Map<String, Map<String, List<String>>>> connectResponse = withRetries(
          () -> Request.get(resolveUri(CONNECTORS + "/" + connector + TOPICS))
              .setHeaders(requestHeaders)
              .responseTimeout(Timeout.ofMilliseconds(requestTimeoutMs))
              .connectTimeout(Timeout.ofMilliseconds(requestTimeoutMs))
              .execute(httpClient)
              .handleResponse(
                  createHandler(HttpStatus.SC_OK,
                      new TypeReference<Map<String, Map<String, List<String>>>>() {},
                      Function.identity())));

      connectResponse.error()
          .ifPresent(
              error -> LOG.warn("Could not query topics of connector {}: {}", connector, error));

      return connectResponse;
    } catch (final Exception e) {
      throw new KsqlServerException(e);
    }
  }
@Test
  public void testTopics() throws JsonProcessingException {
    // Given:
    WireMock.stubFor(
        WireMock.get(WireMock.urlEqualTo(pathPrefix + "/connectors/foo/topics"))
            .withHeader(AUTHORIZATION.toString(), new EqualToPattern(AUTH_HEADER))
            .withHeader(CUSTOM_HEADER_NAME, new EqualToPattern(CUSTOM_HEADER_VALUE))
            .willReturn(WireMock.aResponse()
                .withStatus(HttpStatus.SC_OK)
                .withBody(MAPPER.writeValueAsString(SAMPLE_TOPICS)))
    );

    // When:
    final ConnectResponse<Map<String, Map<String, List<String>>>> response = client.topics("foo");

    // Then:
    final Map<String, Map<String, List<String>>> activeTopics = response.datum()
        .orElseThrow(AssertionError::new);
    // equals is not implemented on ConnectorStateInfo
    assertThat(activeTopics.keySet(), is(ImmutableSet.of("foo")));
    assertThat(activeTopics.get("foo").get("topics"), is(ImmutableList.of("topicA", "topicB")));
    assertThat("Expected no error!", !response.error().isPresent());
  }