GH-3193 Add ackCount and ackTime to KafkaConsumerProperties

Signed-off-by: akenra <37288280+akenra@users.noreply.github.com>

Author: akenra

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java

@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.stream.binder.kafka.properties;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -99,6 +100,24 @@ public enum StandardHeaders {
 	 */
 	private ContainerProperties.AckMode ackMode;
 
+	/**
+	 * Consumer ack count. When set, the container will send an acknowledgment after the specified number
+	 * of messages have been consumed. Only applicable for COUNT or COUNT_TIME ack modes.
+	 * <p>
+	 * This property is binding-level and takes precedence over the global
+	 * {@code spring.kafka.listener.ackCount} when explicitly configured on the binding.
+	 */
+	private Integer ackCount;
+
+	/**
+	 * Consumer ack time. When set, the container will send an acknowledgment after the specified duration
+	 * has elapsed since the last message was consumed. Only applicable for TIME or COUNT_TIME ack modes.
+	 * <p>
+	 * This property is binding-level and takes precedence over the global
+	 * {@code spring.kafka.listener.ackTime} when explicitly configured on the binding.
+	 */
+	private Duration ackTime;
+
 	/**
 	 * Flag to enable auto commit on error in polled consumers.
 	 */
@@ -223,6 +242,34 @@ public void setAckMode(ContainerProperties.AckMode ackMode) {
 		this.ackMode = ackMode;
 	}
 
+	/**
+	 * @return consumer ack count, or null if not set at the binding level.
+	 * <p>
+	 * Binding-level value takes precedence over the global
+	 * {@code spring.kafka.listener.ackCount} when explicitly configured.
+	 */
+	public Integer getAckCount() {
+		return this.ackCount;
+	}
+
+	public void setAckCount(Integer ackCount) {
+		this.ackCount = ackCount;
+	}
+
+	/**
+	 * @return consumer ack time, or null if not set at the binding level.
+	 * <p>
+	 * Binding-level value takes precedence over the global
+	 * {@code spring.kafka.listener.ackTime} when explicitly configured.
+	 */
+	public Duration getAckTime() {
+		return this.ackTime;
+	}
+
+	public void setAckTime(Duration ackTime) {
+		this.ackTime = ackTime;
+	}
+
 	/**
 	 * @return start offset
 	 *

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -27,6 +27,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
+import java.time.Duration;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
@@ -710,6 +711,18 @@ else if (applicationContext != null) {
 			}
 		}
 
+		// Apply binding-level ackCount and ackTime after global propagation
+		// These take precedence over global spring.kafka.listener values when explicitly set
+		KafkaConsumerProperties consumerExt = extendedConsumerProperties.getExtension();
+		Integer bindingAckCount = consumerExt.getAckCount();
+		if (bindingAckCount != null) {
+			messageListenerContainer.getContainerProperties().setAckCount(bindingAckCount);
+		}
+		Duration bindingAckTime = consumerExt.getAckTime();
+		if (bindingAckTime != null) {
+			messageListenerContainer.getContainerProperties().setAckTime(bindingAckTime.toMillis());
+		}
+
 		if (this.logger.isDebugEnabled()) {
 			this.logger.debug("Listened partitions: "
 					+ StringUtils.collectionToCommaDelimitedString(listenedPartitions));

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java

@@ -2015,6 +2015,29 @@ void manualAckIsNotPossibleWhenAutoCommitOffsetIsEnabledOnTheBinder()
 		consumerBinding.unbind();
 	}
 
+	@Test
+	void bindingAckCountAndAckTimeAreAppliedToContainerProperties() {
+		Binder<MessageChannel, ExtendedConsumerProperties<KafkaConsumerProperties>, ExtendedProducerProperties<KafkaProducerProperties>> binder = getBinder();
+		var moduleInputChannel = new QueueChannel();
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		consumerProperties.getExtension().setAckCount(10);
+		consumerProperties.getExtension().setAckTime(Duration.ofSeconds(5));
+
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer(
+				"testBindingAckCountAndAckTime" + UUID.randomUUID(), "test", moduleInputChannel,
+				consumerProperties);
+		try {
+			AbstractMessageListenerContainer<?, ?> container = TestUtils.getPropertyValue(
+					consumerBinding, "lifecycle.messageListenerContainer",
+					AbstractMessageListenerContainer.class);
+			assertThat(container.getContainerProperties().getAckCount()).isEqualTo(10);
+			assertThat(container.getContainerProperties().getAckTime()).isEqualTo(5000L);
+		}
+		finally {
+			consumerBinding.unbind();
+		}
+	}
+
 	@Test
 	@Override
 	@SuppressWarnings("unchecked")

docs/modules/ROOT/pages/kafka/kafka-binder/config-options.adoc

@@ -189,6 +189,16 @@ ackMode::
 Specify the container ack mode.
 This is based on the AckMode enumeration defined in Spring Kafka.
 If `ackEachRecord` property is set to `true` and consumer is not in batch mode, then this will use the ack mode of `RECORD`, otherwise, use the provided ack mode using this property.
+ackCount::
+The number of records that must be processed before pending offsets are committed when the `ackMode` is `COUNT` or `COUNT_TIME`.
+When set, this binding-level value takes precedence over the global `spring.kafka.listener.ackCount`.
++
+Default: not set.
+ackTime::
+The time after which pending offsets are committed when the `ackMode` is `TIME` or `COUNT_TIME`.
+When set, this binding-level value takes precedence over the global `spring.kafka.listener.ackTime`.
++
+Default: not set.
 
 autoCommitOnError::
 In pollable consumers, if set to `true`, it always auto commits on error.