public List<PrometheusQueryResult> queryMetric(String queryString,
                                                   long startTimeMs,
                                                   long endTimeMs) throws IOException {
        URI queryUri = URI.create(_prometheusEndpoint.toURI() + QUERY_RANGE_API_PATH);
        HttpPost httpPost = new HttpPost(queryUri);

        List<NameValuePair> data = new ArrayList<>();
        data.add(new BasicNameValuePair(QUERY, queryString));
        /* "start" and "end" are expected to be unix timestamp in seconds (number of seconds since the Unix epoch).
         They accept values with a decimal point (up to 64 bits). The samples returned are inclusive of the "end"
         timestamp provided.
         */
        data.add(new BasicNameValuePair(START, String.valueOf((double) startTimeMs / SEC_TO_MS)));
        data.add(new BasicNameValuePair(END, String.valueOf((double) endTimeMs / SEC_TO_MS)));
        // step is expected to be in seconds, and accept values with a decimal point (up to 64 bits).
        data.add(new BasicNameValuePair(STEP, String.valueOf((double) _samplingIntervalMs / SEC_TO_MS)));

        httpPost.setEntity(new UrlEncodedFormEntity(data));
        try (CloseableHttpResponse response = _httpClient.execute(httpPost)) {
            int responseCode = response.getStatusLine().getStatusCode();
            HttpEntity entity = response.getEntity();
            InputStream content = entity.getContent();
            String responseString = IOUtils.toString(content, StandardCharsets.UTF_8);
            if (responseCode != HttpServletResponse.SC_OK) {
                throw new IOException(String.format("Received non-success response code on Prometheus API HTTP call,"
                                                    + " response code = %d, response body = %s",
                                                    responseCode, responseString));
            }
            PrometheusResponse prometheusResponse = GSON.fromJson(responseString, PrometheusResponse.class);
            if (prometheusResponse == null) {
                throw new IOException(String.format(
                    "No response received from Prometheus API query, response body = %s", responseString));
            }

            if (!SUCCESS.equals(prometheusResponse.status())) {
                throw new IOException(String.format(
                    "Prometheus API query was not successful, response body = %s", responseString));
            }
            if (prometheusResponse.data() == null
                || prometheusResponse.data().result() == null) {
                throw new IOException(String.format(
                    "Response from Prometheus HTTP API is malformed, response body = %s", responseString));
            }
            EntityUtils.consume(entity);
            return prometheusResponse.data().result();
        }
    }
@Test(expected = IOException.class)
    public void testEmptyStatus() throws Exception {
        this.serverBootstrap.registerHandler("/api/v1/query_range", new HttpRequestHandler() {
            @Override
            public void handle(HttpRequest request, HttpResponse response, HttpContext context) {
                response.setStatusCode(HttpServletResponse.SC_OK);
                response.setEntity(new StringEntity(
                    "{\"data\":{\"result\": []}}", StandardCharsets.UTF_8));
            }
        });

        HttpHost httpHost = this.start();
        PrometheusAdapter prometheusAdapter
            = new PrometheusAdapter(this.httpclient, httpHost, SAMPLING_INTERVAL_MS);

        prometheusAdapter.queryMetric(
            "kafka_server_BrokerTopicMetrics_OneMinuteRate{name=\"BytesOutPerSec\",topic=\"\"}",
            START_TIME_MS, END_TIME_MS);
    }