KafkaMonitorJob implements Job { Stats collect(MBeanServerConnection connection) throws JMException, IOException { var stats = new Stats(); CompositeData heap = (CompositeData) connection.getAttribute(MEMORY_BEAN, "HeapMemoryUsage"); double heapUsed = (Long) heap.get("used"); stats.put("kafka_heap_used", heapUsed); double heapMax = (Long) heap.get("max"); stats.put("kafka_heap_max", heapMax); stats.checkHighUsage(heapUsed / heapMax, highHeapUsageThreshold, "heap"); CompositeData nonHeap = (CompositeData) connection.getAttribute(MEMORY_BEAN, "NonHeapMemoryUsage"); double nonHeapUsed = (Long) nonHeap.get("used"); stats.put("kafka_non_heap_used", nonHeapUsed); collectGCStats(stats, connection, youngGCStats, YOUNG_GC_BEAN); collectGCStats(stats, connection, oldGCStats, OLD_GC_BEAN); stats.put("kafka_bytes_out_rate", (Double) connection.getAttribute(BYTES_OUT_RATE_BEAN, "OneMinuteRate")); stats.put("kafka_bytes_in_rate", (Double) connection.getAttribute(BYTES_IN_RATE_BEAN, "OneMinuteRate")); return stats; } KafkaMonitorJob(JMXClient jmxClient, String app, String host, MessagePublisher<StatMessage> publisher); @Override void execute(JobContext context); public double highHeapUsageThreshold; }
@Test void collect() throws JMException, IOException { CompositeData heap = mock(CompositeData.class); when(heap.get("used")).thenReturn(1000L); when(heap.get("max")).thenReturn(2000L); when(connection.getAttribute(KafkaMonitorJob.MEMORY_BEAN, "HeapMemoryUsage")).thenReturn(heap); when(connection.getAttribute(KafkaMonitorJob.MEMORY_BEAN, "NonHeapMemoryUsage")).thenReturn(heap); when(connection.getAttribute(KafkaMonitorJob.OLD_GC_BEAN, "CollectionCount")).thenReturn(0L); when(connection.getAttribute(KafkaMonitorJob.OLD_GC_BEAN, "CollectionTime")).thenReturn(0L); when(connection.getAttribute(KafkaMonitorJob.YOUNG_GC_BEAN, "CollectionCount")).thenReturn(1L); when(connection.getAttribute(KafkaMonitorJob.YOUNG_GC_BEAN, "CollectionTime")).thenReturn(100L); when(connection.getAttribute(KafkaMonitorJob.BYTES_OUT_RATE_BEAN, "OneMinuteRate")).thenReturn(10D); when(connection.getAttribute(KafkaMonitorJob.BYTES_IN_RATE_BEAN, "OneMinuteRate")).thenReturn(10D); Stats stats = job.collect(connection); assertThat(stats.stats).containsKeys("kafka_heap_used", "kafka_heap_max", "kafka_non_heap_used", "kafka_gc_young_count", "kafka_gc_young_elapsed", "kafka_gc_old_count", "kafka_gc_old_elapsed", "kafka_bytes_out_rate", "kafka_bytes_in_rate"); }