class DefaultConnectClient implements ConnectClient {

  @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(connectUri.resolve(String.format("%s/%s", CONNECTORS, connector)))
          .setHeaders(headers())
          .responseTimeout(Timeout.ofMilliseconds(DEFAULT_TIMEOUT_MS))
          .connectTimeout(Timeout.ofMilliseconds(DEFAULT_TIMEOUT_MS))
          .execute()
          .handleResponse(
              createHandler(HttpStatus.SC_NO_CONTENT, 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);
    }
  }

  public  DefaultConnectClient(
      final String connectUri,
      final Optional<String> authHeader
  );

}