@SuppressWarnings("WeakerAccess")
    public Map<String, Object> getProducerConfigs(final String clientId) {
        final Map<String, Object> clientProvidedProps = getClientPropsWithPrefix(PRODUCER_PREFIX, ProducerConfig.configNames());

        checkIfUnexpectedUserSpecifiedConsumerConfig(clientProvidedProps, NON_CONFIGURABLE_PRODUCER_EOS_CONFIGS);

        // generate producer configs from original properties and overridden maps
        final Map<String, Object> props = new HashMap<>(eosEnabled ? PRODUCER_EOS_OVERRIDES : PRODUCER_DEFAULT_OVERRIDES);
        props.putAll(getClientCustomProps());
        props.putAll(clientProvidedProps);

        // When using EOS alpha, stream should auto-downgrade the transactional commit protocol to be compatible with older brokers.
        if (StreamsConfigUtils.processingMode(this) == StreamsConfigUtils.ProcessingMode.EXACTLY_ONCE_ALPHA) {
            props.put("internal.auto.downgrade.txn.commit", true);
        }

        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, originals().get(BOOTSTRAP_SERVERS_CONFIG));
        // add client id with stream client id prefix
        props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);

        return props;
    }
Get the configs for the {@link KafkaProducer producer}.
Properties using the prefix {@link #PRODUCER_PREFIX} will be used in favor over their non-prefixed versions
except in the case of {@link ProducerConfig#BOOTSTRAP_SERVERS_CONFIG} where we always use the non-prefixed
version as we only support reading/writing from/to the same Kafka Cluster.

@param clientId clientId
@return Map of the producer configuration.
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java