public static <T> HttpResponse<T> httpRequest(Logger logger, String url, String method,
            Object requestBodyData, TypeReference<T> responseFormat) throws IOException {
        HttpURLConnection connection = null;
        try {
            String serializedBody = requestBodyData == null ? null :
                JsonUtil.JSON_SERDE.writeValueAsString(requestBodyData);
            logger.debug("Sending {} with input {} to {}", method, serializedBody, url);
            connection = (HttpURLConnection) new URL(url).openConnection();
            connection.setRequestMethod(method);
            connection.setRequestProperty("User-Agent", "kafka");
            connection.setRequestProperty("Accept", "application/json");

            // connection.getResponseCode() implicitly calls getInputStream, so always set
            // this to true.
            connection.setDoInput(true);

            connection.setUseCaches(false);

            if (requestBodyData != null) {
                connection.setRequestProperty("Content-Type", "application/json");
                connection.setDoOutput(true);

                OutputStream os = connection.getOutputStream();
                os.write(serializedBody.getBytes(StandardCharsets.UTF_8));
                os.flush();
                os.close();
            }

            int responseCode = connection.getResponseCode();
            if (responseCode == HttpURLConnection.HTTP_NO_CONTENT) {
                return new HttpResponse<>(null, new ErrorResponse(responseCode, connection.getResponseMessage()));
            } else if ((responseCode >= 200) && (responseCode < 300)) {
                InputStream is = connection.getInputStream();
                T result = JsonUtil.JSON_SERDE.readValue(is, responseFormat);
                is.close();
                return new HttpResponse<>(result, null);
            } else {
                // If the resposne code was not in the 200s, we assume that this is an error
                // response.
                InputStream es = connection.getErrorStream();
                if (es == null) {
                    // Handle the case where HttpURLConnection#getErrorStream returns null.
                    return new HttpResponse<>(null, new ErrorResponse(responseCode, ""));
                }
                // Try to read the error response JSON.
                ErrorResponse error = JsonUtil.JSON_SERDE.readValue(es, ErrorResponse.class);
                es.close();
                return new HttpResponse<>(null, error);
            }
        } finally {
            if (connection != null) {
                connection.disconnect();
            }
        }
    }
Make an HTTP request.

@param logger            The logger to use.
@param url               HTTP connection will be established with this url.
@param method            HTTP method ("GET", "POST", "PUT", etc.)
@param requestBodyData   Object to serialize as JSON and send in the request body.
@param responseFormat    Expected format of the response to the HTTP request.
@param <T>               The type of the deserialized response to the HTTP request.
@return The deserialized response to the HTTP request, or null if no data is expected.
trogdor/src/main/java/org/apache/kafka/trogdor/rest/JsonRestServer.java