DefaultKafkaConsumerFactory Should have a way to set the configs after instantiation

[jbotuck]

As of spring boot 2.3 it is possible to use customizers to configure some of boot's autoconfigured kafka client beans

https://docs.spring.io/spring-boot/docs/current/api/org/springframework/boot/autoconfigure/kafka/DefaultKafkaConsumerFactoryCustomizer.html

https://docs.spring.io/spring-boot/docs/current/api/org/springframework/boot/autoconfigure/kafka/DefaultKafkaProducerFactoryCustomizer.html

https://docs.spring.io/spring-boot/docs/current/api/org/springframework/boot/autoconfigure/kafka/StreamsBuilderFactoryBeanCustomizer.html

The problem is that the consumer customizer has limited utility since its currently impossible to modify the config in the consumer factory. Is it possible to add a way to modify the config? Similar to the updateConfigs method in the producer factory? https://docs.spring.io/spring-kafka/docs/current/api/org/springframework/kafka/core/ProducerFactory.html#updateConfigs(java.util.Map)

[garyrussell]

I agree it should be made symmetric.

However, the consumer factory has an overloaded createConsumer method:

	/**
	 * Create a consumer with an explicit group id; in addition, the
	 * client id suffix is appended to the clientIdPrefix which overrides the
	 * {@code client.id} property, if present. In addition, consumer properties can
	 * be overridden if the factory implementation supports it.
	 * @param groupId the group id.
	 * @param clientIdPrefix the prefix.
	 * @param clientIdSuffix the suffix.
	 * @param properties the properties to override.
	 * @return the consumer.
	 * @since 2.2.4
	 */
	Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
			@Nullable String clientIdSuffix, @Nullable Properties properties);

It is used by the listener container.

With the listener container, you can set overrides there:

container.getContainerProperties().setKafkaConsumerProperties(...);

	/**
	 * Set the consumer properties that will be merged with the consumer properties
	 * provided by the consumer factory; properties here will supersede any with the same
	 * name(s) in the consumer factory.
	 * {@code group.id} and {@code client.id} are ignored.
	 * Property keys must be {@link String}s.
	 * @param kafkaConsumerProperties the properties.
	 * @see org.apache.kafka.clients.consumer.ConsumerConfig
	 * @see #setGroupId(String)
	 * @see #setClientId(String)
	 */
	public void setKafkaConsumerProperties(Properties kafkaConsumerProperties) {
           ...
	}

[jbotuck]

Cool, I didn't know about that other way of setting the properties. But Boot doesn't really provide a great way to customize the autoconfigured containerfactory bean.

My current workaround is to have my code create the ConsumerFactory Bean manually, instead of my preferred flow of letting boot create the factory and my code supplying the factory customizer

[garyrussell]

You can add a customizer bean to customize Boot's container factory:

@Component
class Customizer {

	public Customizer(ConcurrentKafkaListenerContainerFactory<?, ?> factory) {
		factory.setContainerCustomizer(container -> {
			if (container.getContainerProperties().getGroupId().equals("slowGroup")) {
				container.getContainerProperties().setIdleBetweenPolls(60_000);
			}
		});
	}

}

This is safe because the only code in the factory's afterPropertiesSet() is some validation of the error handler type.