private <T> HttpResponse<T> httpRequest(HttpClient client, String url, String method,
                                           HttpHeaders headers, Object requestBodyData,
                                           TypeReference<T> responseFormat, SecretKey sessionKey,
                                           String requestSignatureAlgorithm) {
        try {
            String serializedBody = requestBodyData == null ? null : JSON_SERDE.writeValueAsString(requestBodyData);
            log.trace("Sending {} with input {} to {}", method, serializedBody, url);

            Request req = client.newRequest(url);
            req.method(method);
            req.accept("application/json");
            req.agent("kafka-connect");
            addHeadersToRequest(headers, req);

            if (serializedBody != null) {
                req.content(new StringContentProvider(serializedBody, StandardCharsets.UTF_8), "application/json");
            }

            if (sessionKey != null && requestSignatureAlgorithm != null) {
                InternalRequestSignature.addToRequest(
                    Crypto.SYSTEM,
                    sessionKey,
                    serializedBody != null ? serializedBody.getBytes(StandardCharsets.UTF_8) : null,
                    requestSignatureAlgorithm,
                    req
                );
            }

            ContentResponse res = req.send();

            int responseCode = res.getStatus();
            log.debug("Request's response code: {}", responseCode);
            if (responseCode == HttpStatus.NO_CONTENT_204) {
                return new HttpResponse<>(responseCode, convertHttpFieldsToMap(res.getHeaders()), null);
            } else if (responseCode >= 400) {
                ErrorMessage errorMessage = JSON_SERDE.readValue(res.getContentAsString(), ErrorMessage.class);
                throw new ConnectRestException(responseCode, errorMessage.errorCode(), errorMessage.message());
            } else if (responseCode >= 200 && responseCode < 300) {
                T result = JSON_SERDE.readValue(res.getContentAsString(), responseFormat);
                return new HttpResponse<>(responseCode, convertHttpFieldsToMap(res.getHeaders()), result);
            } else {
                throw new ConnectRestException(Response.Status.INTERNAL_SERVER_ERROR,
                        Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(),
                        "Unexpected status code when handling forwarded request: " + responseCode);
            }
        } catch (IOException | TimeoutException | ExecutionException e) {
            log.error("IO error forwarding REST request to {} :", url, e);
            throw new ConnectRestException(Response.Status.INTERNAL_SERVER_ERROR, "IO Error trying to forward REST request: " + e.getMessage(), e);
        } catch (InterruptedException e) {
            log.error("Thread was interrupted forwarding REST request to {} :", url, e);
            Thread.currentThread().interrupt();
            throw new ConnectRestException(Response.Status.INTERNAL_SERVER_ERROR, "Thread was interrupted trying to forward REST request: " + e.getMessage(), e);
        } catch (ConnectRestException e) {
            // catching any explicitly thrown ConnectRestException-s to preserve its status code
            // and to avoid getting it overridden by the more generic catch (Throwable) clause down below
            log.error("Error forwarding REST request to {} :", url, e);
            throw e;
        } catch (Throwable t) {
            log.error("Error forwarding REST request to {} :", url, t);
            throw new ConnectRestException(Response.Status.INTERNAL_SERVER_ERROR, "Error trying to forward REST request: " + t.getMessage(), t);
        }
    }
Sends HTTP request to remote REST server

@param url                       HTTP connection will be established with this url, non-null.
@param method                    HTTP method ("GET", "POST", "PUT", etc.), non-null
@param headers                   HTTP headers from REST endpoint
@param requestBodyData           Object to serialize as JSON and send in the request body.
@param responseFormat            Expected format of the response to the HTTP request, non-null.
@param <T>                       The type of the deserialized response to the HTTP request.
@param sessionKey                The key to sign the request with (intended for internal requests only);
                                 may be null if the request doesn't need to be signed
@param requestSignatureAlgorithm The algorithm to sign the request with (intended for internal requests only);
                                 may be null if the request doesn't need to be signed
@return The deserialized response to the HTTP request, containing null if no data is expected or returned.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestClient.java