Skip to content

Update the DefaultProducerFactory to support config overrides of transactional producers  #2070

@frosiere

Description

@frosiere

In some specific use cases, it may be required to update the configuration entries of a transactional consumer before creating it.

To allow this, a slight update of the DefaultKafkaProducer#doCreateTxProducer method may be done as follow

// private -> protected
// new rawConfigs parameters
protected CloseSafeProducer<K, V> doCreateTxProducer(String prefix, String suffix,
        BiPredicate<CloseSafeProducer<K, V>, Duration> remover, Map<String, Object> rawConfigs) {
    // same code as now
}

This update would allow a pre-processing of configuration entries as follow

public class CustomDefaultKafkaProducerFactory<K, V> extends DefaultKafkaProducerFactory<K, V> {
	
	@Override
	protected CloseSafeProducer<K, V> doCreateTxProducer(String prefix, String suffix, BiPredicate<CloseSafeProducer<K, V>, Duration> remover, Map<String, Object> rawConfigs) {
		final Map<String, Object> newProducerConfigs = new HashMap<>(rawConfigs);
		// overrides
		return super.doCreateTxProducer(prefix, suffix, remover, newProducerConfigs);
	}
}

This issue follows a Stackoverflow discussion described here.

Metadata

Metadata

Assignees

No one assigned

    Type

    No type

    Projects

    No projects

    Milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions