KafkaProducerFactoryImpl implements KafkaProducerFactory, HasSystemInfo { @Override public SharedKafkaProducer getSharedProducer(final DocRef kafkaConfigDocRef) { Objects.requireNonNull(kafkaConfigDocRef); Objects.requireNonNull(kafkaConfigDocRef.getUuid(), "No Kafka config UUID has been defined"); final SharedKafkaProducerImpl sharedKafkaProducer; final Optional<KafkaConfigDoc> optKafkaConfigDoc = kafkaConfigDocCache.get(kafkaConfigDocRef); if (optKafkaConfigDoc.isPresent()) { final KafkaConfigDoc kafkaConfigDoc = optKafkaConfigDoc.get(); final DocRef docRefFromDoc = DocRefUtil.create(kafkaConfigDoc); final SharedKafkaProducerIdentity desiredKey = new SharedKafkaProducerIdentity(kafkaConfigDoc); final SharedKafkaProducerImpl activeSharedProducer = currentSharedProducersMap.get(desiredKey.getConfigUuid()); if (activeSharedProducer != null && desiredKey.equals(activeSharedProducer.getSharedKafkaProducerIdentity())) { LOGGER.debug("Using latest SharedKafkaProducer"); sharedKafkaProducer = activeSharedProducer; } else { sharedKafkaProducer = currentSharedProducersMap.compute( kafkaConfigDoc.getUuid(), (k, existingValue) -> { final SharedKafkaProducerImpl sharedKafkaProducer2; if (existingValue == null) { sharedKafkaProducer2 = createSharedProducer(kafkaConfigDoc, docRefFromDoc, desiredKey); } else { if (desiredKey.equals(existingValue.getSharedKafkaProducerIdentity())) { LOGGER.debug("Existing SharedKafkaProducer is up to date"); sharedKafkaProducer2 = existingValue; } else { existingValue.markSuperseded(); LOGGER.debug("Superseding existing sharedKafkaProducer {}", existingValue); removeRedundantSharedKafkaProducers(); sharedKafkaProducer2 = createSharedProducer(kafkaConfigDoc, docRefFromDoc, desiredKey); } } return sharedKafkaProducer2; }); } sharedKafkaProducer.incrementUseCount(); } else { sharedKafkaProducer = SharedKafkaProducerImpl.empty(); } return sharedKafkaProducer; } @Inject  KafkaProducerFactoryImpl(final KafkaConfigDocCache kafkaConfigDocCache); @Override SharedKafkaProducer getSharedProducer(final DocRef kafkaConfigDocRef); @Override void returnSharedKafkaProducer(final SharedKafkaProducer sharedKafkaProducer); static Properties getProperties(KafkaConfigDoc doc); @Override SystemInfoResult getSystemInfo();  }
@Test public void getSupplier_empty() { final KafkaProducerFactory kafkaProducerFactory = new KafkaProducerFactoryImpl(kafkaConfigDocCache); final KafkaConfigDoc kafkaConfigDoc = createKafkaConfigDoc("Config1", "v1"); final DocRef docRef = DocRefUtil.create(kafkaConfigDoc); final SharedKafkaProducer sharedKafkaProducer = kafkaProducerFactory.getSharedProducer(docRef); assertThat(sharedKafkaProducer.getKafkaProducer()).isEmpty(); assertThat(sharedKafkaProducer.hasKafkaProducer()).isFalse(); sharedKafkaProducer.close(); }
@Test public void getSupplier_twoConfigs() { final KafkaProducerFactory kafkaProducerFactory = new KafkaProducerFactoryImpl(kafkaConfigDocCache); final KafkaConfigDoc kafkaConfigDoc1 = createKafkaConfigDoc("Config1", "v1"); final KafkaConfigDoc kafkaConfigDoc2 = createKafkaConfigDoc("Config2", "v1"); final DocRef docRef1 = DocRefUtil.create(kafkaConfigDoc1); final DocRef docRef2 = DocRefUtil.create(kafkaConfigDoc2); Mockito.when(kafkaConfigDocCache.get(Mockito.eq(docRef1))) .thenReturn(Optional.of(kafkaConfigDoc1)); Mockito.when(kafkaConfigDocCache.get(Mockito.eq(docRef2))) .thenReturn(Optional.of(kafkaConfigDoc2)); final SharedKafkaProducer sharedKafkaProducer1 = kafkaProducerFactory.getSharedProducer(docRef1); assertThat(sharedKafkaProducer1.getKafkaProducer()).isPresent(); assertThat(sharedKafkaProducer1.hasKafkaProducer()).isTrue(); assertThat(sharedKafkaProducer1.getConfigName()).isEqualTo(kafkaConfigDoc1.getName()); assertThat(sharedKafkaProducer1.getConfigUuid()).isEqualTo(kafkaConfigDoc1.getUuid()); assertThat(sharedKafkaProducer1.getConfigVersion()).isEqualTo(kafkaConfigDoc1.getVersion()); KafkaProducer<String, byte[]> kafkaProducer1 = sharedKafkaProducer1.getKafkaProducer().get(); assertThat(kafkaProducer1).isNotNull(); final SharedKafkaProducer sharedKafkaProducer2 = kafkaProducerFactory.getSharedProducer(docRef2); assertThat(sharedKafkaProducer2.getKafkaProducer()).isPresent(); assertThat(sharedKafkaProducer2.hasKafkaProducer()).isTrue(); assertThat(sharedKafkaProducer2.getConfigName()).isEqualTo(kafkaConfigDoc2.getName()); assertThat(sharedKafkaProducer2.getConfigUuid()).isEqualTo(kafkaConfigDoc2.getUuid()); assertThat(sharedKafkaProducer2.getConfigVersion()).isEqualTo(kafkaConfigDoc2.getVersion()); KafkaProducer<String, byte[]> kafkaProducer2 = sharedKafkaProducer2.getKafkaProducer().get(); assertThat(kafkaProducer2).isNotNull(); assertThat(kafkaProducer1).isNotSameAs(kafkaProducer2); final SharedKafkaProducer sharedKafkaProducer1B = kafkaProducerFactory.getSharedProducer(docRef1); assertThat(sharedKafkaProducer1B.getKafkaProducer()).isPresent(); assertThat(sharedKafkaProducer1B.hasKafkaProducer()).isTrue(); assertThat(sharedKafkaProducer1B.getConfigName()).isEqualTo(kafkaConfigDoc1.getName()); assertThat(sharedKafkaProducer1B.getConfigUuid()).isEqualTo(kafkaConfigDoc1.getUuid()); assertThat(sharedKafkaProducer1B.getConfigVersion()).isEqualTo(kafkaConfigDoc1.getVersion()); KafkaProducer<String, byte[]> kafkaProducer1b = sharedKafkaProducer1B.getKafkaProducer().get(); assertThat(kafkaProducer1b).isNotNull(); assertThat(kafkaProducer1b).isNotSameAs(kafkaProducer2); assertThat(kafkaProducer1b).isSameAs(kafkaProducer1); kafkaProducer1.close(); kafkaProducer2.close(); kafkaProducer1b.close(); }
@Test public void getSupplier_updatedConfig() { final KafkaProducerFactory kafkaProducerFactory = new KafkaProducerFactoryImpl(kafkaConfigDocCache); final KafkaConfigDoc kafkaConfigDoc1mk1 = createKafkaConfigDoc("Config1", "v1"); final DocRef docRef1 = DocRefUtil.create(kafkaConfigDoc1mk1); Mockito.when(kafkaConfigDocCache.get(Mockito.eq(docRef1))) .thenReturn(Optional.of(kafkaConfigDoc1mk1)); final SharedKafkaProducer sharedKafkaProducer1 = kafkaProducerFactory.getSharedProducer(docRef1); assertThat(sharedKafkaProducer1.getKafkaProducer()).isPresent(); assertThat(sharedKafkaProducer1.hasKafkaProducer()).isTrue(); assertThat(sharedKafkaProducer1.getConfigName()).isEqualTo(kafkaConfigDoc1mk1.getName()); assertThat(sharedKafkaProducer1.getConfigUuid()).isEqualTo(kafkaConfigDoc1mk1.getUuid()); assertThat(sharedKafkaProducer1.getConfigVersion()).isEqualTo(kafkaConfigDoc1mk1.getVersion()); final KafkaProducer<String, byte[]> kafkaProducer1 = sharedKafkaProducer1.getKafkaProducer().get(); assertThat(kafkaProducer1).isNotNull(); final KafkaConfigDoc kafkaConfigDoc1mk2 = createKafkaConfigDoc("Config1", "v2"); kafkaConfigDoc1mk2.setUuid(kafkaConfigDoc1mk1.getUuid()); Mockito.when(kafkaConfigDocCache.get(Mockito.eq(docRef1))) .thenReturn(Optional.of(kafkaConfigDoc1mk2)); final SharedKafkaProducer sharedKafkaProducer1B = kafkaProducerFactory.getSharedProducer(docRef1); assertThat(sharedKafkaProducer1B.getKafkaProducer()).isPresent(); assertThat(sharedKafkaProducer1B.hasKafkaProducer()).isTrue(); assertThat(sharedKafkaProducer1B.getConfigName()).isEqualTo(kafkaConfigDoc1mk1.getName()); assertThat(sharedKafkaProducer1B.getConfigUuid()).isEqualTo(kafkaConfigDoc1mk1.getUuid()); assertThat(sharedKafkaProducer1B.getConfigVersion()).isEqualTo(kafkaConfigDoc1mk2.getVersion()); final KafkaProducer<String, byte[]> kafkaProducer1b = sharedKafkaProducer1B.getKafkaProducer().get(); assertThat(kafkaProducer1b).isNotNull(); assertThat(kafkaProducer1b).isNotSameAs(kafkaProducer1); final SharedKafkaProducer sharedKafkaProducer1C = kafkaProducerFactory.getSharedProducer(docRef1); assertThat(sharedKafkaProducer1C.getKafkaProducer()).isPresent(); assertThat(sharedKafkaProducer1C.hasKafkaProducer()).isTrue(); assertThat(sharedKafkaProducer1C.getConfigName()).isEqualTo(kafkaConfigDoc1mk1.getName()); assertThat(sharedKafkaProducer1C.getConfigUuid()).isEqualTo(kafkaConfigDoc1mk1.getUuid()); assertThat(sharedKafkaProducer1C.getConfigVersion()).isEqualTo(kafkaConfigDoc1mk2.getVersion()); KafkaProducer<String, byte[]> kafkaProducer1c = sharedKafkaProducer1C.getKafkaProducer().get(); assertThat(kafkaProducer1c).isNotNull(); assertThat(kafkaProducer1c).isNotSameAs(kafkaProducer1); assertThat(kafkaProducer1c).isSameAs(kafkaProducer1b); kafkaProducer1.close(); kafkaProducer1b.close(); kafkaProducer1c.close(); }