class DefaultConnectClient implements ConnectClient {

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

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

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

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

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

}