@Override
  public ConnectResponse<List<SimpleConnectorPluginInfo>> connectorPlugins() {
    try {
      LOG.debug("Issuing request to Kafka Connect at URI {} to list connector plugins", connectUri);

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

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

      return connectResponse;
    } catch (final Exception e) {
      throw new KsqlServerException(e);
    }
  }
@Test
  public void testListPlugins() throws JsonProcessingException {
    // Given:

    MAPPER.writeValueAsString(ImmutableList.of(SAMPLE_PLUGIN));
    WireMock.stubFor(
        WireMock.get(WireMock.urlEqualTo(pathPrefix + "/connector-plugins"))
            .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(ImmutableList.of(SAMPLE_PLUGIN))))
    );

    // When:
    final ConnectResponse<List<SimpleConnectorPluginInfo>> response = client.connectorPlugins();

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