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

    public  AnomalyConsumer();

}