private SharedKafkaProducerImpl createSharedProducer(
            final KafkaConfigDoc kafkaConfigDoc,
            final DocRef kafkaConfigDocRef,
            final SharedKafkaProducerIdentity key) {

        final Properties producerProperties = getProperties(kafkaConfigDoc);
        final KafkaProducer<String, byte[]> kafkaProducer;
        try {
            // For flexibility we always use a byte[] as the msg value.  This means the same producer
            // can cope with different flavours of data and it is up to the consumer to know
            // what data is on what topic. It is also up to the use of the KP to serialise down to a byte[]
            LOGGER.debug("Creating KafkaProducer for {}", key);
            kafkaProducer = new KafkaProducer<>(producerProperties, new StringSerializer(), new ByteArraySerializer());
        } catch (Exception e) {
            throw new RuntimeException(LogUtil.message("Error creating KafkaProducer for {} - {}: {}",
                    kafkaConfigDoc.getName(),
                    kafkaConfigDoc.getUuid(),
                    e.getMessage()),
                    e);
        }
        final SharedKafkaProducerImpl sharedKafkaProducer = new SharedKafkaProducerImpl(
                kafkaProducer, this::returnAction, key, kafkaConfigDocRef);
        // Hold on to a reference to each one we create
        allSharedProducersMap.put(key, sharedKafkaProducer);
        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