private void returnAction(final SharedKafkaProducer sharedKafkaProducer) {

        LOGGER.debug("returnAction called for {}", sharedKafkaProducer);
        if (sharedKafkaProducer != null && sharedKafkaProducer.hasKafkaProducer()) {
            if (sharedKafkaProducer instanceof SharedKafkaProducerImpl) {
                SharedKafkaProducerImpl sharedKafkaProducerImpl = (SharedKafkaProducerImpl) sharedKafkaProducer;
                sharedKafkaProducerImpl.decrementUseCount();
                LOGGER.debug("superseded is {} and useCount is {} after decrement",
                        sharedKafkaProducerImpl.isSuperseded(), sharedKafkaProducerImpl.getUseCount());
                if (sharedKafkaProducerImpl.isSuperseded() && sharedKafkaProducerImpl.getUseCount() <= 0) {
                    allSharedProducersMap.remove(sharedKafkaProducerImpl.getSharedKafkaProducerIdentity());
                    final KafkaProducer<String, byte[]> kafkaProducer = sharedKafkaProducer.getKafkaProducer().get();
                    kafkaProducer.close();
                }
            } else {
                throw new RuntimeException("Unexpected type " + sharedKafkaProducer.getClass().getName());
            }
        }
    }
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