C3QueryUtils { static void extractJsonResults(String jsonStr, List<String> topics, Map<String, TopicWorkload> workloads) { try { JSONObject jsonObj = JSON.parseObject(jsonStr); if (jsonObj == null) { LOGGER.info("Failed to parse C3 result: " + jsonStr); return; } for (String topic : topics) { JSONArray arr = jsonObj.getJSONArray(topic); if (arr == null || arr.size() == 0) { continue; } JSONObject metrics = arr.getJSONObject(0); if (metrics == null) { LOGGER.info("Failed to parse C3 result for topic '" + topic + "'"); continue; } Long startTimeSec = metrics.getLong("startTimeSec"); Long endTimeSec = metrics.getLong("endTimeSec"); Long totalBytes = metrics.getLong("totalBytes"); Long totalCount = metrics.getLong("totalCount"); if (startTimeSec == null || endTimeSec == null || totalBytes == null || totalCount == null) { LOGGER.info("Failed to parse C3 result for topic '" + topic + "'"); continue; } double period = endTimeSec - startTimeSec; if (period <= 0) { LOGGER.info("Invalid C3 result for topic '" + topic + "': startTimeSec=" + startTimeSec + "," + " endTimeSec=" + endTimeSec); continue; } TopicWorkload tw = new TopicWorkload(totalBytes / period, totalCount / period); tw.setLastUpdate(endTimeSec * 1000); workloads.put(topic, tw); } } catch (Exception e) { LOGGER.error("Failed to extract workload information from JSON: " + jsonStr, e); } }  static Map<String, TopicWorkload> retrieveTopicInRate(long timeInMs, long windowInMs, String c3Host,
      int c3Port, String kafkaCluster, List<String> topics);  }
@Test public void testExtractJsonResults() { Map<String, TopicWorkload> workloads = new HashMap<>(); C3QueryUtils.extractJsonResults( "{\"topic1\":[{\"endTimeSec\":1485991200,\"startTimeSec\":1485988200,\"totalBytes\":30000,\"totalCount\":3000}]," + "\"topic2\":[{\"endTimeSec\":1485991200,\"invalidCount\":0,\"maxLatencyFromCreation\":3750,\"meanLatencyFromCreation\":1576.2152157017917,\"p99LatencyFromCreation\":3529,\"startTimeSec\":1485988200,\"totalBytes\":3000000,\"totalCount\":1500}]}", Arrays.asList("topic1", "topic2"), workloads); TopicWorkload tw1 = workloads.get("topic1"); Assert.assertEquals(tw1.getBytesPerSecond(), 10, 0.01); Assert.assertEquals(tw1.getMsgsPerSecond(), 1, 0.01); TopicWorkload tw2 = workloads.get("topic2"); Assert.assertEquals(tw2.getBytesPerSecond(), 1000, 0.01); Assert.assertEquals(tw2.getMsgsPerSecond(), 0.5, 0.01); workloads.clear(); C3QueryUtils.extractJsonResults("", Arrays.asList("topic1", "topic2"), workloads); Assert.assertTrue(workloads.isEmpty()); C3QueryUtils.extractJsonResults("{}", Arrays.asList("topic1", "topic2"), workloads); Assert.assertTrue(workloads.isEmpty()); C3QueryUtils.extractJsonResults("{\"topic1\":[]}", Arrays.asList("topic1", "topic2"), workloads); Assert.assertTrue(workloads.isEmpty()); C3QueryUtils.extractJsonResults( "{\"topic1\":[{,\"startTimeSec\":1485988200,\"totalBytes\":30000,\"totalCount\":3000}]," + "\"topic2\":[{\"endTimeSec\":1485991200,\"invalidCount\":0,\"maxLatencyFromCreation\":3750,\"meanLatencyFromCreation\":1576.2152157017917,\"p99LatencyFromCreation\":3529,\"startTimeSec\":1485988200,\"totalBytes\":3000000,\"totalCount\":1500}]}", Arrays.asList("topic1", "topic2"), workloads); tw1 = workloads.get("topic1"); Assert.assertNull(tw1); tw2 = workloads.get("topic2"); Assert.assertEquals(tw2.getBytesPerSecond(), 1000, 0.01); Assert.assertEquals(tw2.getMsgsPerSecond(), 0.5, 0.01); workloads.clear(); C3QueryUtils.extractJsonResults( "{\"topic1\":[{\"endTimeSec\":1485988200,\"startTimeSec\":1485988200,\"totalBytes\":30000,\"totalCount\":3000}]," + "\"topic2\":[{\"endTimeSec\":1485991200,\"invalidCount\":0,\"maxLatencyFromCreation\":3750,\"meanLatencyFromCreation\":1576.2152157017917,\"p99LatencyFromCreation\":3529,\"startTimeSec\":1485988200,\"totalBytes\":3000000,\"totalCount\":1500}]}", Arrays.asList("topic1", "topic2"), workloads); tw1 = workloads.get("topic1"); Assert.assertNull(tw1); tw2 = workloads.get("topic2"); Assert.assertEquals(tw2.getBytesPerSecond(), 1000, 0.01); Assert.assertEquals(tw2.getMsgsPerSecond(), 0.5, 0.01); workloads.clear(); }