KafkaMessageSource Blocks on Initial Poll When No Records

[garyrussell]

5.5.4.RELEASE

KMS increases the poll timeout for the first poll (assignment).

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

With the default pollTimeout (5s), when there are no records, the poll blocks for 100s.

This causes a delay, for example, with a PollableConsumer in SCSt the poll is not responsive.

Wake the consumer when the partitions are assigned.