Impossible to create Listener Containers with different Kafka consumer properties

[sergeylemekhov]

Affects Version(s): 2.7.0

🎁 Enhancement

When manually creating a listener container via listener container factory org.springframework.kafka.config.AbstractKafkaListenerContainerFactory#createContainer(java.lang.String...) method the listener container is initialized with properties from the factory.
The org.springframework.kafka.config.AbstractKafkaListenerContainerFactory#initializeContainer method sets listener container's kafkaConsumerProperties as a reference to the factory consumerProperties.
When calling code then changes kafkaConsumerProperties of the created listener container the changes affect all other listener containers which is the problem.

Because of this it is impossible to create two listener containers with different offset reset policies for example:

ConcurrentMessageListenerContainer<String, String> container1 = listenerContainerFactory.createContainer("topic");
container1.getContainerProperties().setGroupId("group1");
container1.getContainerProperties().getKafkaConsumerProperties().setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "lateset");

ConcurrentMessageListenerContainer<String, String> container2 = listenerContainerFactory.createContainer("topic");
container2.getContainerProperties().setGroupId("group2");
container2.getContainerProperties().getKafkaConsumerProperties().setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

Both listeners in the snippet end up with "earliest" offset reset policy.

A way to fix this would be creating a copy of factory consumerProperties before setting it for the listener container.

[garyrussell]

Something is amiss here; each container gets a new ContainerProperties:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/config/ConcurrentKafkaListenerContainerFactory.java

Lines 69 to 71 in bd15452

	if (!topics.isEmpty()) { // NOSONAR
	ContainerProperties properties = new ContainerProperties(topics.toArray(new String[0]));
	return new ConcurrentMessageListenerContainer<>(getConsumerFactory(), properties);

We then copy properties from the factory's properties into the instance properties:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

Lines 403 to 433 in bd15452

	ContainerProperties properties = instance.getContainerProperties();
	BeanUtils.copyProperties(this.containerProperties, properties, "topics", "topicPartitions", "topicPattern",
	"messageListener", "ackCount", "ackTime", "subBatchPerPartition");
	JavaUtils.INSTANCE
	.acceptIfNotNull(this.afterRollbackProcessor, instance::setAfterRollbackProcessor)
	.acceptIfCondition(this.containerProperties.getAckCount() > 0, this.containerProperties.getAckCount(),
	properties::setAckCount)
	.acceptIfCondition(this.containerProperties.getAckTime() > 0, this.containerProperties.getAckTime(),
	properties::setAckTime)
	.acceptIfNotNull(this.containerProperties.getSubBatchPerPartition(),
	properties::setSubBatchPerPartition)
	.acceptIfNotNull(this.errorHandler, instance::setGenericErrorHandler)
	.acceptIfNotNull(this.missingTopicsFatal, instance.getContainerProperties()::setMissingTopicsFatal);
	Boolean autoStart = endpoint.getAutoStartup();
	if (autoStart != null) {
	instance.setAutoStartup(autoStart);
	}
	else if (this.autoStartup != null) {
	instance.setAutoStartup(this.autoStartup);
	}
	instance.setRecordInterceptor(this.recordInterceptor);
	instance.setBatchInterceptor(this.batchInterceptor);
	JavaUtils.INSTANCE
	.acceptIfNotNull(this.phase, instance::setPhase)
	.acceptIfNotNull(this.applicationContext, instance::setApplicationContext)
	.acceptIfNotNull(this.applicationEventPublisher, instance::setApplicationEventPublisher)
	.acceptIfHasText(endpoint.getGroupId(), instance.getContainerProperties()::setGroupId)
	.acceptIfHasText(endpoint.getClientIdPrefix(), instance.getContainerProperties()::setClientId)
	.acceptIfNotNull(endpoint.getConsumerProperties(),
	instance.getContainerProperties()::setKafkaConsumerProperties);
	}

So it's not at all clear how we could get cross-talk between container properties.

[garyrussell]

Oh; I see - it's the containerProperies.kafkaConsumerProperties that is the same - it should be omitted from the BeanUtils copy.

As a work around, give each container a new Properties, instead of getting it from the container properties.

	/**
	 * 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) {
		Assert.notNull(kafkaConsumerProperties, "'kafkaConsumerProperties' cannot be null");
		this.kafkaConsumerProperties = kafkaConsumerProperties;
	}

[sergeylemekhov]

@garyrussell I've also created a PR #1810
I've added a wrapper for Properties instance that delegates default values to factory instance and added a separate test.