diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc index 4a57a516f..5c435d6f3 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc @@ -129,7 +129,10 @@ The property `spring.cloud.stream.instanceCount` must typically be greater than Default: `true`. autoCommitOffset:: Whether to autocommit offsets when a message has been processed. -If set to `false`, an `Acknowledgment` header will be available in the message headers for late acknowledgment. +If set to `false`, a header with the key `kafka_acknowledgment` of the type `org.springframework.kafka.support.Acknowledgment` header will be present in the inbound message. +Applications may use this header for acknowledging messages. +See the examples section for details. +When this property is set to `false`, Kafka binder will set the ack mode to `org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode.MANUAL`. + Default: `true`. autoCommitOnError:: @@ -199,6 +202,34 @@ If a topic already exists with a larger number of partitions than the maximum of In this section, we illustrate the use of the above properties for specific scenarios. +==== Example: Setting `autoCommitOffset` false and relying on manual acking. + +This example illustrates how one may manually acknowledge offsets in a consumer application. + +This example requires that `spring.cloud.stream.kafka.bindings.input.consumer.autoCommitOffset` is set to false. +Use the corresponding input channel name for your example. + +[source] +---- +@SpringBootApplication +@EnableBinding(Sink.class) +public class ManuallyAcknowdledgingConsumer { + + public static void main(String[] args) { + SpringApplication.run(ManuallyAcknowdledgingConsumer.class, args); + } + + @StreamListener(Sink.INPUT) + public void process(Message message) { + Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); + if (acknowledgment != null) { + System.out.println("Acknowledgment provided"); + acknowledgment.acknowledge(); + } + } +} +---- + ==== Example: security configuration Apache Kafka 0.9 supports secure connections between client and brokers. diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 8378ad3ce..b1e69e14f 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -47,6 +47,8 @@ import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -60,6 +62,7 @@ import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; /** * @author Soby Chacko @@ -638,6 +641,81 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests producerBinding = binder.bindProducer("foo.x", moduleOutputChannel, + createProducerProperties()); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setAutoCommitOffset(false); + + Binding consumerBinding = binder.bindConsumer("foo.x", "test", moduleInputChannel, + consumerProperties); + + String testPayload1 = "foo" + UUID.randomUUID().toString(); + Message message1 = org.springframework.integration.support.MessageBuilder.withPayload( + testPayload1.getBytes()).build(); + + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel.send(message1); + + Message receivedMessage = receive(moduleInputChannel); + assertThat(receivedMessage).isNotNull(); + assertThat(receivedMessage.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT)).isNotNull(); + Acknowledgment acknowledgment = receivedMessage.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); + try { + acknowledgment.acknowledge(); + } + catch (Exception e) { + fail("Acknowledge must not throw an exception"); + } + finally { + producerBinding.unbind(); + consumerBinding.unbind(); + } + } + + @Test + @SuppressWarnings("unchecked") + public void testManualAckIsNotPossibleWhenAutoCommitOffsetIsEnabledOnTheBinder() throws Exception { + Binder binder = getBinder(); + + DirectChannel moduleOutputChannel = createBindableChannel("output", + createProducerBindingProperties(createProducerProperties())); + QueueChannel moduleInputChannel = new QueueChannel(); + + Binding producerBinding = binder.bindProducer("foo.x", moduleOutputChannel, + createProducerProperties()); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + + Binding consumerBinding = binder.bindConsumer("foo.x", "test", moduleInputChannel, + consumerProperties); + + String testPayload1 = "foo" + UUID.randomUUID().toString(); + Message message1 = org.springframework.integration.support.MessageBuilder.withPayload( + testPayload1.getBytes()).build(); + + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel.send(message1); + + Message receivedMessage = receive(moduleInputChannel); + assertThat(receivedMessage).isNotNull(); + assertThat(receivedMessage.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT)).isNull(); + + producerBinding.unbind(); + consumerBinding.unbind(); + } + @Test @Override @SuppressWarnings("unchecked")