class DefaultConnectClient implements ConnectClient {

  @Override
  public ConnectResponse<ConnectorInfo> create(
      final String connector,
      final Map<String, String> config
  ) {
    try {
      LOG.debug("Issuing create request to Kafka Connect at URI {} with name {} and config {}",
          connectUri,
          connector,
          config);

      final ConnectResponse<ConnectorInfo> connectResponse = withRetries(() -> Request
          .post(connectUri.resolve(CONNECTORS))
          .setHeaders(headers())
          .responseTimeout(Timeout.ofMilliseconds(DEFAULT_TIMEOUT_MS))
          .connectTimeout(Timeout.ofMilliseconds(DEFAULT_TIMEOUT_MS))
          .bodyString(
              MAPPER.writeValueAsString(
                  ImmutableMap.of(
                      "name", connector,
                      "config", config)),
              ContentType.APPLICATION_JSON
          )
          .execute()
          .handleResponse(
              createHandler(HttpStatus.SC_CREATED, new TypeReference<ConnectorInfo>() {},
                  Function.identity())));

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

      return connectResponse;
    } catch (final Exception e) {
      throw new KsqlServerException(e);
    }
  }

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

}