class 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;
    }

}

class AnomalyConsumerTest {

    private KafkaConsumer<String, MappedMetricData> kafkaConsumer;
    private ExecutorService executorService;
    private AnomalyConsumer anomalyConsumer;

    @Test
    public void testProcessZeroMetrics() {

        ConsumerRecords<String, MappedMetricData> metricRecords = AnomaliesProcessorTest.buildMetricRecords(0,
                AnomalyLevel.STRONG);
        when(kafkaConsumer.poll(anyLong())).thenReturn(metricRecords);
        assertTrue(anomalyConsumer.process(kafkaConsumer, true));

    }

}