@Test public void testProcessWakeException() { when(kafkaConsumer.poll(anyLong())).thenThrow(WakeupException.class); assertFalse(anomalyConsumer.process(kafkaConsumer, true)); verify(kafkaConsumer, times(1)).close(); }
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 { 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 { 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(); }
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(); }
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; }