Add a configuration property for KLC's idleBetweenPolls

See gh-23048

Author: s50600822

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java

@@ -187,6 +187,7 @@ private void configureContainer(ContainerProperties container) {
 		map.from(properties::getIdleEventInterval).as(Duration::toMillis).to(container::setIdleEventInterval);
 		map.from(properties::getMonitorInterval).as(Duration::getSeconds).as(Number::intValue)
 				.to(container::setMonitorInterval);
+		map.from(properties.getIdleBetweenPolls()).as(Duration::toMillis).to(container::setIdleBetweenPolls);
 		map.from(properties::getLogContainerConfig).to(container::setLogContainerConfig);
 		map.from(properties::isMissingTopicsFatal).to(container::setMissingTopicsFatal);
 		map.from(this.transactionManager).to(container::setTransactionManager);

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java

@@ -890,6 +890,12 @@ public enum Type {
 		 */
 		private Duration idleEventInterval;
 
+		/**
+		 * The sleep interval in milliseconds between
+		 * {@link org.apache.kafka.clients.consumer.Consumer#poll(Duration)} calls.
+		 */
+		private Duration idleBetweenPolls;
+
 		/**
 		 * Time between checks for non-responsive consumers. If a duration suffix is not
 		 * specified, seconds will be used.
@@ -980,6 +986,14 @@ public void setIdleEventInterval(Duration idleEventInterval) {
 			this.idleEventInterval = idleEventInterval;
 		}
 
+		public Duration getIdleBetweenPolls() {
+			return idleBetweenPolls;
+		}
+
+		public void setIdleBetweenPolls(Duration idleBetweenPolls) {
+			this.idleBetweenPolls = idleBetweenPolls;
+		}
+
 		public Duration getMonitorInterval() {
 			return this.monitorInterval;
 		}
@@ -1003,7 +1017,6 @@ public boolean isMissingTopicsFatal() {
 		public void setMissingTopicsFatal(boolean missingTopicsFatal) {
 			this.missingTopicsFatal = missingTopicsFatal;
 		}
-
 	}
 
 	public static class Ssl {