@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 shared_producer_is_allowed_to_be_created_with_empty_props() {
        kafkaDataConnection = createKafkaDataConnection(kafkaTestSupport);

        Producer<Object, Object> kafkaProducer = kafkaDataConnection.getProducer(null, new Properties());
        assertThat(kafkaProducer).isNotNull();

        kafkaProducer.close();
        kafkaDataConnection.release();
    }