AnomalyConsumer { public boolean process(KafkaConsumer kafkaConsumer, boolean continueProcessing) { try { ConsumerRecords<String, MappedMetricData> metricRecords = kafkaConsumer.poll(POLL_INTERVAL); log.trace("Read {} metric records from topic={}", metricRecords.count(), consumerConfig.getString(TOPIC)); anomaliesProcessor.processMetrics(metricRecords, executorService); } catch (WakeupException e) { kafkaConsumer.close(); continueProcessing = false; } catch (Exception e) { log.error(e.getLocalizedMessage()); e.printStackTrace(); } return continueProcessing; } AnomalyConsumer(); void listen(); boolean process(KafkaConsumer kafkaConsumer, boolean continueProcessing); @VisibleForTesting void setKafkaConsumer(KafkaConsumer<String, MappedMetricData> kafkaConsumer); @VisibleForTesting void setExecutorService(ExecutorService executorService); AnomaliesProcessor getAnomaliesProcessor(); public Config consumerConfig; public Properties metricProps; }
@Test public void testProcess() { ConsumerRecords<String, MappedMetricData> metricRecords = AnomaliesProcessorTest.buildMetricRecords(2, AnomalyLevel.STRONG); when(kafkaConsumer.poll(anyLong())).thenReturn(metricRecords); assertTrue(anomalyConsumer.process(kafkaConsumer, true)); }
@Test public void testProcessZeroMetrics() { ConsumerRecords<String, MappedMetricData> metricRecords = AnomaliesProcessorTest.buildMetricRecords(0, AnomalyLevel.STRONG); when(kafkaConsumer.poll(anyLong())).thenReturn(metricRecords); assertTrue(anomalyConsumer.process(kafkaConsumer, true)); }
@Test public void testProcessWakeException() { when(kafkaConsumer.poll(anyLong())).thenThrow(WakeupException.class); assertFalse(anomalyConsumer.process(kafkaConsumer, true)); verify(kafkaConsumer, times(1)).close(); }
@Test public void testProcessException() { when(kafkaConsumer.poll(anyLong())).thenThrow(IllegalStateException.class); assertTrue(anomalyConsumer.process(kafkaConsumer, true)); }