GH-1845: Idle Between Polls Improvements

#1845

- don't idle until partitions are assigned - it caused delayed assignment
- use `ListenerUtils.stoppableSleep()` - don't hold up container stop while idle
- `ListenerUtils.stoppableSleep()` - reduce sleep time for small intervals to
  improve resolution

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -1526,13 +1526,14 @@ private void checkIdle() {
 
 		private void idleBetweenPollIfNecessary() {
 			long idleBetweenPolls = this.containerProperties.getIdleBetweenPolls();
-			if (idleBetweenPolls > 0) {
+			Collection<TopicPartition> assigned = getAssignedPartitions();
+			if (idleBetweenPolls > 0 && assigned != null && assigned.size() > 0) {
 				idleBetweenPolls = Math.min(idleBetweenPolls,
 						this.maxPollInterval - (System.currentTimeMillis() - this.lastPoll)
 								- 5000); // NOSONAR - less by five seconds to avoid race condition with rebalance
 				if (idleBetweenPolls > 0) {
 					try {
-						TimeUnit.MILLISECONDS.sleep(idleBetweenPolls);
+						ListenerUtils.stoppableSleep(KafkaMessageListenerContainer.this, idleBetweenPolls);
 					}
 					catch (InterruptedException ex) {
 						Thread.currentThread().interrupt();

spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java

@@ -49,7 +49,11 @@ private ListenerUtils() {
 
 	private static final ThreadLocal<Boolean> LOG_METADATA_ONLY = new ThreadLocal<>();
 
-	private static final int SLEEP_INTERVAL = 100;
+	private static final int DEFAULT_SLEEP_INTERVAL = 100;
+
+	private static final int SMALL_SLEEP_INTERVAL = 10;
+
+	private static final long SMALL_INTERVAL_THRESHOLD = 500;
 
 	/**
 	 * Determine the type of the listener.
@@ -243,8 +247,9 @@ public static void unrecoverableBackOff(BackOff backOff, ThreadLocal<BackOffExec
 	 */
 	public static void stoppableSleep(MessageListenerContainer container, long interval) throws InterruptedException {
 		long timeout = System.currentTimeMillis() + interval;
+		long sleepInterval = interval > SMALL_INTERVAL_THRESHOLD ? DEFAULT_SLEEP_INTERVAL : SMALL_SLEEP_INTERVAL;
 		do {
-			Thread.sleep(SLEEP_INTERVAL);
+			Thread.sleep(sleepInterval);
 			if (!container.isRunning()) {
 				break;
 			}

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -2508,6 +2508,39 @@ public void testInitialSeek() throws Exception {
 		container.stop();
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	public void testIdleEarlyExit() throws Exception {
+		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
+		Consumer<Integer, String> consumer = mock(Consumer.class);
+		given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer);
+		ConsumerRecords<Integer, String> emptyRecords = new ConsumerRecords<>(Collections.emptyMap());
+		final CountDownLatch latch = new CountDownLatch(1);
+		given(consumer.poll(any(Duration.class))).willAnswer(i -> {
+			latch.countDown();
+			Thread.sleep(50);
+			return emptyRecords;
+		});
+		ContainerProperties containerProps = new ContainerProperties("foo");
+		containerProps.setGroupId("grp");
+		containerProps.setAckMode(AckMode.RECORD);
+		containerProps.setClientId("clientId");
+		containerProps.setMessageListener((MessageListener) r -> { });
+		containerProps.setMissingTopicsFatal(false);
+		containerProps.setIdleBetweenPolls(60_000L);
+		containerProps.setShutdownTimeout(20_000);
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		new DirectFieldAccessor(container).setPropertyValue("listenerConsumer.assignedPartitions",
+				Arrays.asList(new TopicPartition("foo", 0)));
+		Thread.sleep(500);
+		long t1 = System.currentTimeMillis();
+		container.stop();
+		assertThat(System.currentTimeMillis() - t1).isLessThan(10_000L);
+	}
+
 	@Test
 	public void testExceptionWhenCommitAfterRebalance() throws Exception {
 		final CountDownLatch rebalanceLatch = new CountDownLatch(2);