GH-2675: DefaultBinding isRunning enhancements

[sobychacko]

• Instead of relying on the lifecycle object's isRunning method which could possibly be synchronized and cause a contention for the lock, maintain the running state separately inside DefaultBinding via a field (similar to how it tracks the pausable state).

  Without this change, in the pollable Kafka consumer based applications, there is a possiblity for a longer wait when queyring the binding status in DefaultBinding (for example via the binding actuator endpoint) due to this aforementioned lock contention.

Resolves #2675

[garyrussell]

This needs to be fixed in Spring Integration, not here.

cc/ @artembilan

[sobychacko]

@garyrussell @artembilan What do you think about temporarily having this change here? If it's not worth it, we will close this PR here.

[artembilan]

I'm sorry, I fail to determine what is wrong and what should be fixed in the KafkaMessageSource.
Note: in 6.2 we have moved to an internal Lock instead of synchronized on those methods.
Maybe that can fix it? Then we simply can back-port the fix to previous versions.
But again: where is a dead lock, please?
Thanks

[sobychacko]

@artembilan The deadlock happens when the /binding actuator endpoint makes a call to the KafkaMessageSource#isRunning() method. Because of a timeout set there, there is a code path, that a thread may wait for the lock to release, which happens only after the timeout. This only happens for pollable consumers. See this issue for more details. #2675

[sobychacko]

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

[garyrussell]

I think we can just change the lifecycle synchronization to use an AtomicBoolean for running.

I am not sure why it is set to true after creating the consumer (in doReceive()) - looks like that was added by an external contributor.

[artembilan]

The doReceive() is synchronized in the Spring Integration version and it is blocked waiting for result from the pollRecord() which ends up to be waiting for consumer.poll() where no partitions assigned to this consumer since the only one existing has been assigned to another one.
That's how it is dead-locking.
I think we can just narrow a synchronized/locked block in the doReceive() to exclude that pollRecord() call.

[garyrussell]

Right; we. should also call consumer.wakeup() in stopConsumer() in case the poller is stuck here.