StroomStatsInternalStatisticsService implements InternalStatisticsService { @Override public void putEvents(final Map<DocRef, List<InternalStatisticEvent>> eventsMap) { final DocRef kafkaConfigDocRef = new DocRef(KafkaConfigDoc.DOCUMENT_TYPE, internalStatisticsConfig.getKafkaConfigUuid()); try(SharedKafkaProducer sharedKafkaProducer = stroomKafkaProducerFactory.getSharedProducer(kafkaConfigDocRef)) { sharedKafkaProducer.getKafkaProducer().ifPresentOrElse( kafkaProducer -> sendMessages(eventsMap, kafkaProducer), () -> { throw new RuntimeException("The Kafka producer isn't initialised, unable to send any events"); } ); } } @Inject  StroomStatsInternalStatisticsService(final KafkaProducerFactory stroomKafkaProducerFactory,
                                         final HBaseStatisticsConfig internalStatisticsConfig); @Override void putEvents(final Map<DocRef, List<InternalStatisticEvent>> eventsMap); @Override String getDocRefType();  }
@Test void putEvents_multipleEvents() { final HBaseStatisticsConfig hBaseStatisticsConfig = new HBaseStatisticsConfig(); hBaseStatisticsConfig.setDocRefType(DOC_REF_TYPE_1); hBaseStatisticsConfig.getKafkaTopicsConfig().setCount("MyTopic"); initMocks(); StroomStatsInternalStatisticsService stroomStatsInternalStatisticsService = new StroomStatsInternalStatisticsService( mockStroomKafkaProducerFactory, hBaseStatisticsConfig); InternalStatisticEvent event1 = InternalStatisticEvent.createPlusOneCountStat( InternalStatisticKey.MEMORY, 0, Collections.emptyMap()); InternalStatisticEvent event2 = InternalStatisticEvent.createPlusOneCountStat( InternalStatisticKey.MEMORY, 1, Collections.emptyMap()); InternalStatisticEvent event3 = InternalStatisticEvent.createPlusOneCountStat( InternalStatisticKey.MEMORY, 1, Collections.emptyMap()); DocRef docRefA = new DocRef(DOC_REF_TYPE_1, UUID.randomUUID().toString(), "myStat1"); DocRef docRefB = new DocRef(DOC_REF_TYPE_2, UUID.randomUUID().toString(), "myStat2"); Map<DocRef, List<InternalStatisticEvent>> map = Map.of( docRefA, Arrays.asList(event1, event2), docRefB, Collections.singletonList(event3)); stroomStatsInternalStatisticsService.putEvents(map); Mockito.verify(mockKafkaProducer, Mockito.times(2)) .send(Mockito.any(ProducerRecord.class)); }
@Test void putEvents_largeBatch() { final HBaseStatisticsConfig hBaseStatisticsConfig = new HBaseStatisticsConfig(); hBaseStatisticsConfig.setDocRefType(DOC_REF_TYPE_1); hBaseStatisticsConfig.getKafkaTopicsConfig().setCount("MyTopic"); hBaseStatisticsConfig.setEventsPerMessage(10); initMocks(); StroomStatsInternalStatisticsService stroomStatsInternalStatisticsService = new StroomStatsInternalStatisticsService( mockStroomKafkaProducerFactory, hBaseStatisticsConfig ); DocRef docRefA = new DocRef(DOC_REF_TYPE_1, UUID.randomUUID().toString(), "myStat1"); DocRef docRefB = new DocRef(DOC_REF_TYPE_2, UUID.randomUUID().toString(), "myStat2"); Map<DocRef, List<InternalStatisticEvent>> map = Map.of( docRefA, createNEvents(InternalStatisticKey.MEMORY, 100), docRefB, createNEvents(InternalStatisticKey.CPU, 15)); stroomStatsInternalStatisticsService.putEvents(map); Mockito.verify(mockKafkaProducer, Mockito.times(12)) .send(Mockito.any(ProducerRecord.class)); }