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

      final ConnectResponse<String> connectResponse = withRetries(() -> Request
          .delete(resolveUri(String.format("%s/%s", CONNECTORS, connector)))
          .setHeaders(requestHeaders)
          .responseTimeout(Timeout.ofMilliseconds(requestTimeoutMs))
          .connectTimeout(Timeout.ofMilliseconds(requestTimeoutMs))
          .execute(httpClient)
          .handleResponse(
              createHandler(
                  ImmutableList.of(HttpStatus.SC_NO_CONTENT, HttpStatus.SC_OK),
                  new TypeReference<Object>() {},
                  foo -> connector)));


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

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

    // When:
    final ConnectResponse<String> response = client.delete("foo");

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