Kafka binder used with pollables makes fail the bindings actuator endpoint

[JordiMartinezVicent]

Describe the issue
Given an application with two instances and:

• Uses the kafka binder
• Declares a pollable

The actuator endpoint /actuator/bindings lasts around 100s if the call is done to the instance which has no assigned partitions.

To Reproduce

scs-pollables.zip

With the attached application the behavior can be reproduced. In order to simulate two instances of the same application, it declares two pollables with the same group pointing to the same topic.

To startup the application, a kafka broker is needed.

To reproduce:

• Start up the attached application
• Call to the endpoint http://localhost:8080/actuator/bindings

Expected behavior
The call to the endpoint should not last around 100s

Version of the framework

Spring Boot: 3.0.4
spring-cloud-dependencies: 2022.0.1

Additional context

The issue is due to the class org.springframework.integration.kafka.inbound.KafkaMessageSource is synchronized and used for polling and to retrieve the binding state from the bindings endpoint.

On the one hand, it defines the following timeouts:

this.pollTimeout = Duration.ofMillis(consumerProperties.getPollTimeout());
this.assignTimeout = Duration.ofMillis(Math.max(this.pollTimeout.toMillis() * 20, MIN_ASSIGN_TIMEOUT)); // NOSONAR - magic 		

Note that the default pollTimeout is 5 seconds, so the assignTimeout will be 100s.

And it uses them here at the method pollRecord:

ConsumerRecords<K, V> records = this.consumer
  poll(this.assignedPartitions.isEmpty() ? this.assignTimeout : this.pollTimeout);

Note that the assignTimeout is used if there are no assignedPartitions

On the other hand, when calling the endpoint, the method of the same class isRunning is called (which is syncronized).

Maybe to avoid the synchronization point, the state running and stoped could be manage at the class DefaultBinding, same as it does with the "paused" state. In that manner, the polling and the retrieving of the state would not collision.

[JordiMartinezVicent]

Hi guys, any update with this?
@olegz @garyrussell

[sobychacko]

Closing this in lieu of spring-projects/spring-integration#8778

The fix will be available in Spring Cloud Stream 4.1.0 and other patch releases.