@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_should_not_be_created_with_additional_props() {
        kafkaDataConnection = createKafkaDataConnection(kafkaTestSupport);
        Properties properties = new Properties();
        properties.setProperty("A", "B");

        assertThatThrownBy(() -> kafkaDataConnection.getProducer(null, properties))
                .isInstanceOf(HazelcastException.class)
                .hasMessageContaining("For shared Kafka producer, please provide all serialization options");

        kafkaDataConnection.release();
    }