8a5c0377ef
Resolves https://github.com/spring-projects/spring-integration/issues/3637 If there are no records to receive, the poll blocked for `20*pollTimeout`. Wake the consumer during partition assignment; however, this changes the behavior when there are records present - the first poll always returns no records. Detect that the wakeup was due to assignment and perform another poll. Also add some debug logging, and don't create a new consumer if the source was stopped. **cherry-pick to 5.4.x**