@Test(groups = "integration", enabled = false) public void testHandle() throws Exception { Assert.assertEquals(processor.getEventsReceived(), 0); Assert.assertEquals(kafkaEventHandler.getEventsSent(), 0); Assert.assertEquals(kafkaEventHandler.getEventsDiscarded(), 0); Assert.assertEquals(messagesReceived.size(), 0); processor.processEvent(event); Assert.assertEquals(processor.getEventsReceived(), 1); Assert.assertEquals(kafkaEventHandler.getEventsSent(), 1); Assert.assertEquals(kafkaEventHandler.getEventsDiscarded(), 0); Thread.sleep(10000); consumerService.shutdown(); Assert.assertEquals(messagesReceived.size(), 1); final Map<String, String> map = messagesReceived.get(0); Assert.assertEquals(map.get("sourceUUID"), event.getSourceUUID().toString()); Assert.assertEquals(map.get("eventType"), event.getEventType()); Assert.assertEquals(Long.valueOf(map.get("timestamp")), (Long) event.getTimestamp()); Assert.assertEquals(Double.valueOf(map.get("min_heapUsed")), ((MapEvent) event).getMap().get("min_heapUsed")); Assert.assertEquals(Double.valueOf(map.get("max_heapUsed")), ((MapEvent) event).getMap().get("max_heapUsed")); }
@Override public void handle(final Event event) { try { final ProducerData<String, String> data = new ProducerData<String, String>(event.getEventType(), eventToMessage(event)); producer.send(data); eventsSent.incrementAndGet(); } catch (IOException e) { log.warn(e); eventsDiscarded.getAndIncrement(); } }
KafkaEventHandler implements EventHandler { @Override public void handle(final Event event) { try { final ProducerData<String, String> data = new ProducerData<String, String>(event.getEventType(), eventToMessage(event)); producer.send(data); eventsSent.incrementAndGet(); } catch (IOException e) { log.warn(e); eventsDiscarded.getAndIncrement(); } } }
KafkaEventHandler implements EventHandler { @Override public void handle(final Event event) { try { final ProducerData<String, String> data = new ProducerData<String, String>(event.getEventType(), eventToMessage(event)); producer.send(data); eventsSent.incrementAndGet(); } catch (IOException e) { log.warn(e); eventsDiscarded.getAndIncrement(); } } @Inject  KafkaEventHandler(final CollectorConfig config); }
KafkaEventHandler implements EventHandler { @Override public void handle(final Event event) { try { final ProducerData<String, String> data = new ProducerData<String, String>(event.getEventType(), eventToMessage(event)); producer.send(data); eventsSent.incrementAndGet(); } catch (IOException e) { log.warn(e); eventsDiscarded.getAndIncrement(); } } @Inject  KafkaEventHandler(final CollectorConfig config); @Override void handle(final Event event); @MonitorableManaged(monitored = true, monitoringType = {MonitoringType.COUNTER, MonitoringType.RATE}) long getEventsSent(); @MonitorableManaged(monitored = true, monitoringType = {MonitoringType.COUNTER, MonitoringType.RATE}) long getEventsDiscarded(); }
KafkaEventHandler implements EventHandler { @Override public void handle(final Event event) { try { final ProducerData<String, String> data = new ProducerData<String, String>(event.getEventType(), eventToMessage(event)); producer.send(data); eventsSent.incrementAndGet(); } catch (IOException e) { log.warn(e); eventsDiscarded.getAndIncrement(); } } @Inject  KafkaEventHandler(final CollectorConfig config); @Override void handle(final Event event); @MonitorableManaged(monitored = true, monitoringType = {MonitoringType.COUNTER, MonitoringType.RATE}) long getEventsSent(); @MonitorableManaged(monitored = true, monitoringType = {MonitoringType.COUNTER, MonitoringType.RATE}) long getEventsDiscarded();  }