@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);

            // Optimistically assume the map will have the latest SharedKafkaProducer
            final SharedKafkaProducerImpl activeSharedProducer = currentSharedProducersMap.get(
                    desiredKey.getConfigUuid());
            if (activeSharedProducer != null
                    && desiredKey.equals(activeSharedProducer.getSharedKafkaProducerIdentity())) {
                // This is the latest SharedKafkaProducer so use it
                LOGGER.debug("Using latest SharedKafkaProducer");
                sharedKafkaProducer = activeSharedProducer;
            } else {
                // Not there or not latest so compute a new/updated one atomically
                sharedKafkaProducer = currentSharedProducersMap.compute(
                        kafkaConfigDoc.getUuid(),
                        (k, existingValue) -> {
                            final SharedKafkaProducerImpl sharedKafkaProducer2;
                            if (existingValue == null) {
                                // Don't have one so create a new one
                                sharedKafkaProducer2 = createSharedProducer(kafkaConfigDoc, docRefFromDoc, desiredKey);
                            } else {
                                // we already have one so check if it should be current
                                if (desiredKey.equals(existingValue.getSharedKafkaProducerIdentity())) {
                                    // is up to date
                                    LOGGER.debug("Existing SharedKafkaProducer is up to date");
                                    sharedKafkaProducer2 = existingValue;
                                } else {
                                    // needs to be superseded
                                    existingValue.markSuperseded();
                                    LOGGER.debug("Superseding existing sharedKafkaProducer {}", existingValue);

                                    // If nobody is using the superseded one then we can bin it now
                                    removeRedundantSharedKafkaProducers();

                                    // swap out the existing current SharedKafkaProducer for our new one
                                    sharedKafkaProducer2 = createSharedProducer(kafkaConfigDoc,
                                            docRefFromDoc,
                                            desiredKey);
                                }
                            }
                            return sharedKafkaProducer2;
                        });
            }
            // Increment the use count so we can track when objects are unused and can be closed.
            sharedKafkaProducer.incrementUseCount();
        } else {
            // No doc for this docref so return an empty SharedKafkaProducer
            sharedKafkaProducer = SharedKafkaProducerImpl.empty();
        }
        return sharedKafkaProducer;
    }
Class to provide shared {@link KafkaProducer} instances wrapped inside a {@link SharedKafkaProducer}.
We want one {@link KafkaProducer} per {@link KafkaConfigDoc}, however the config in a {@link KafkaConfigDoc}
can change at runtime so on each request for a {@link SharedKafkaProducer} we need to check if the
{@link KafkaConfigDoc} has changed since the time we created it. New calls to
{@link KafkaProducerFactory#getSharedProducer(DocRef)} will always get a {@link SharedKafkaProducer} for the
latest config. {@link KafkaProducer}
stroom-kafka/stroom-kafka-impl/src/main/java/stroom/kafka/impl/KafkaProducerFactoryImpl.java