class DefaultConnectClient implements ConnectClient {

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

      final ConnectResponse<ConnectorStateInfo> connectResponse = withRetries(() -> Request
          .get(connectUri.resolve(CONNECTORS + "/" + connector + STATUS))
          .setHeaders(headers())
          .responseTimeout(Timeout.ofMilliseconds(DEFAULT_TIMEOUT_MS))
          .connectTimeout(Timeout.ofMilliseconds(DEFAULT_TIMEOUT_MS))
          .execute()
          .handleResponse(
              createHandler(HttpStatus.SC_OK, new TypeReference<ConnectorStateInfo>() {},
                  Function.identity())));

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

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

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

}