@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(resolveUri(CONNECTORS + "/" + connector + STATUS))
          .setHeaders(requestHeaders)
          .responseTimeout(Timeout.ofMilliseconds(requestTimeoutMs))
          .connectTimeout(Timeout.ofMilliseconds(requestTimeoutMs))
          .execute(httpClient)
          .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);
    }
  }
@Test
  public void testStatus() throws JsonProcessingException {
    // Given:
    WireMock.stubFor(
        WireMock.get(WireMock.urlEqualTo(pathPrefix + "/connectors/foo/status"))
            .withHeader(AUTHORIZATION.toString(), new EqualToPattern(AUTH_HEADER))
            .withHeader(CUSTOM_HEADER_NAME, new EqualToPattern(CUSTOM_HEADER_VALUE))
            .willReturn(WireMock.aResponse()
                .withStatus(HttpStatus.SC_OK)
                .withBody(MAPPER.writeValueAsString(SAMPLE_STATUS)))
    );

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

    // Then:
    final ConnectorStateInfo connectorStateInfo = response.datum().orElseThrow(AssertionError::new);
    // equals is not implemented on ConnectorStateInfo
    assertThat(connectorStateInfo.name(), is(SAMPLE_STATUS.name()));
    assertThat(connectorStateInfo.type(), is(SAMPLE_STATUS.type()));
    assertThat(connectorStateInfo.connector().state(), is(SAMPLE_STATUS.connector().state()));
    assertThat(connectorStateInfo.connector().workerId(), is(SAMPLE_STATUS.connector().workerId()));
    assertThat(connectorStateInfo.connector().trace(), is(SAMPLE_STATUS.connector().trace()));
    assertThat(connectorStateInfo.tasks().size(), is(SAMPLE_STATUS.tasks().size()));
    assertThat(connectorStateInfo.tasks().get(0).id(), is(SAMPLE_STATUS.tasks().get(0).id()));
    assertThat("Expected no error!", !response.error().isPresent());
  }