@Nonnull
    public <K, V> KafkaProducer<K, V> getProducer(@Nullable String transactionalId) {
        if (getConfig().isShared()) {
            if (transactionalId != null) {
                throw new IllegalArgumentException("Cannot use transactions with shared "
                        + "KafkaProducer for DataConnection" + getConfig().getName());
            }
            retain();
            //noinspection unchecked
            return (KafkaProducer<K, V>) producerSupplier.get();
        } else {
            if (transactionalId != null) {
                @SuppressWarnings({"rawtypes", "unchecked"})
                Map<String, Object> castProperties = (Map) getConfig().getProperties();
                Map<String, Object> copy = new HashMap<>(castProperties);
                copy.put("transactional.id", transactionalId);
                return new KafkaProducer<>(copy);
            } else {
                return new KafkaProducer<>(getConfig().getProperties());
            }
        }
    }
@Test
    public void releasing_data_connection_does_not_close_shared_producer() {
        kafkaDataConnection = createKafkaDataConnection(kafkaTestSupport);

        Producer<Object, Object> producer = kafkaDataConnection.getProducer(null);
        kafkaDataConnection.release();

        try {
            producer.partitionsFor("my-topic");
        } catch (Exception e) {
            fail("Should not throw exception", e);
        }
    }