Manual acking tests

Adding tests to verify manual acking when auto commit
offsets are disabled/enabled.
Docs changes.

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.

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<Abstr
 		consumerBinding2.unbind();
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testManualAckSucceedsWhenAutoCommitOffsetIsTurnedOff() throws Exception {
+		Binder binder = getBinder();
+
+		DirectChannel moduleOutputChannel = createBindableChannel("output",
+				createProducerBindingProperties(createProducerProperties()));
+		QueueChannel moduleInputChannel = new QueueChannel();
+
+		Binding<MessageChannel> producerBinding = binder.bindProducer("foo.x", moduleOutputChannel,
+				createProducerProperties());
+
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		consumerProperties.getExtension().setAutoCommitOffset(false);
+
+		Binding<MessageChannel> 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<MessageChannel> producerBinding = binder.bindProducer("foo.x", moduleOutputChannel,
+				createProducerProperties());
+
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+
+		Binding<MessageChannel> 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")