@SuppressWarnings("unchecked")
  @Override
  public ConnectResponse<List<String>> connectors() {
    try {
      LOG.debug("Issuing request to Kafka Connect at URI {} to list connectors", connectUri);

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


      connectResponse.error()
          .ifPresent(error -> LOG.warn("Could not list connectors: {}.", error));

      return connectResponse;
    } catch (final Exception e) {
      throw new KsqlServerException(e);
    }
  }
@Test
  public void testListShouldRetryOnFailure() throws JsonProcessingException {
    // Given:
    WireMock.stubFor(
        WireMock.get(WireMock.urlEqualTo(pathPrefix + "/connectors"))
            .withHeader(AUTHORIZATION.toString(), new EqualToPattern(AUTH_HEADER))
            .withHeader(CUSTOM_HEADER_NAME, new EqualToPattern(CUSTOM_HEADER_VALUE))
            .willReturn(WireMock.aResponse()
                .withStatus(HttpStatus.SC_INTERNAL_SERVER_ERROR)
                .withBody("Encountered an error!"))
            .willReturn(WireMock.aResponse()
                .withStatus(HttpStatus.SC_OK)
                .withBody(MAPPER.writeValueAsString(ImmutableList.of("one", "two"))))
    );

    // When:
    final ConnectResponse<List<String>> response = client.connectors();

    // Then:
    assertThat(response.datum(), OptionalMatchers.of(is(ImmutableList.of("one", "two"))));
    assertThat("Expected no error!", !response.error().isPresent());
  }