Provide no-op AcknowledgementCallback implementation if MessageHeader is not present and Add no-op callback to TestChannelBinder MessageSource.

Resolves #1988

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,6 +22,8 @@ import java.util.function.BiConsumer;
 
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
@@ -59,12 +61,15 @@ import org.springframework.util.Assert;
  *
  * @author Gary Russell
  * @author Oleg Zhurakousky
+ * @author David Turanski
  * @since 2.0
  *
  */
 public class DefaultPollableMessageSource
 		implements PollableMessageSource, Lifecycle, RetryListener {
 
+	private static final Log log = LogFactory.getLog(DefaultPollableMessageSource.class);
+
 	protected static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<AttributeAccessor>();
 
 	private static final DirectChannel dummyChannel = new DirectChannel();
@@ -204,6 +209,11 @@ public class DefaultPollableMessageSource
 
 		AcknowledgmentCallback ackCallback = StaticMessageHeaderAccessor
 				.getAcknowledgmentCallback(message);
+
+		if (ackCallback == null) {
+			ackCallback = status -> log.warn("No AcknowledgementCallback defined. Status: " + status.name() + " " + message);
+		}
+
 		try {
 			if (this.retryTemplate == null) {
 				this.handle(message, handler);

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -62,6 +62,7 @@ import static org.mockito.Mockito.verify;
 /**
  * @author Gary Russell
  * @author Oleg Zhurakousky
+ * @author David Turanski
  * @since 2.0
  *
  */
@@ -77,6 +78,28 @@ public class PollableConsumerTests {
 				.getMessageConverterForAllRegistered();
 	}
 
+	@Test
+	public void testDefaultMessageSource() {
+		TestChannelBinder binder = createBinder();
+		MessageConverterConfigurer configurer = this.context
+				.getBean(MessageConverterConfigurer.class);
+
+		DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(
+				this.messageConverter);
+		configurer.configurePolledMessageSource(pollableSource, "foo");
+		ExtendedConsumerProperties<Object> properties = new ExtendedConsumerProperties<>(
+				null);
+		properties.setMaxAttempts(2);
+		properties.setBackOffInitialInterval(0);
+		binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
+		AtomicInteger count = new AtomicInteger();
+		assertThat(pollableSource.poll(message -> {
+			assertThat(message).isNotNull();
+			count.incrementAndGet();
+		})).isTrue();
+		assertThat(count.get()).isOne();
+	}
+
 	@Test
 	public void testSimple() {
 		TestChannelBinder binder = createBinder();
@@ -414,6 +437,39 @@ public class PollableConsumerTests {
 		verify(callback).acknowledge(Status.REQUEUE);
 	}
 
+	@Test
+	public void testRequeueWithNoAcknowledgementCallback() {
+		TestChannelBinder binder = createBinder();
+		MessageConverterConfigurer configurer = this.context
+				.getBean(MessageConverterConfigurer.class);
+
+		DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(
+				this.messageConverter);
+		configurer.configurePolledMessageSource(pollableSource, "foo");
+		pollableSource.addInterceptor(new ChannelInterceptor() {
+
+			@Override
+			public Message<?> preSend(Message<?> message, MessageChannel channel) {
+				return MessageBuilder.fromMessage(message)
+						.build();
+			}
+
+		});
+		ExtendedConsumerProperties<Object> properties = new ExtendedConsumerProperties<>(null);
+		properties.setMaxAttempts(2);
+		properties.setBackOffInitialInterval(0);
+		binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
+		final AtomicInteger count = new AtomicInteger();
+
+		assertThat(pollableSource.poll(received -> {
+			count.incrementAndGet();
+			throw new RequeueCurrentMessageException("test retry");
+		})).isTrue();
+
+		assertThat(count.get()).isEqualTo(2);
+
+	}
+
 	@Test
 	public void testRequeueFromErrorFlow() {
 		TestChannelBinder binder = createBinder();
@@ -482,19 +538,19 @@ public class PollableConsumerTests {
 	public void testAutoStartupOn() {
 		TestChannelBinder binder = createBinder();
 		binder.setMessageSourceDelegate(new LifecycleMessageSource(
-			() -> new GenericMessage<>("{\"foo\":\"bar\"}".getBytes())));
+				() -> new GenericMessage<>("{\"foo\":\"bar\"}".getBytes())));
 		MessageConverterConfigurer configurer = this.context
-			.getBean(MessageConverterConfigurer.class);
+				.getBean(MessageConverterConfigurer.class);
 
 		DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(
-			this.messageConverter);
+				this.messageConverter);
 		configurer.configurePolledMessageSource(pollableSource, "foo");
 		ExtendedConsumerProperties<Object> properties = new ExtendedConsumerProperties<>(
-			null);
+				null);
 		properties.setAutoStartup(true);
 
 		Binding<PollableSource<MessageHandler>> pollableSourceBinding = binder
-			.bindPollableConsumer("foo", "bar", pollableSource, properties);
+				.bindPollableConsumer("foo", "bar", pollableSource, properties);
 
 		assertThat(pollableSourceBinding.isRunning()).isTrue();
 	}
@@ -523,6 +579,7 @@ public class PollableConsumerTests {
 
 	public static class LifecycleMessageSource<T> implements MessageSource<T>, Lifecycle {
 		private final MessageSource<T> delegate;
+
 		private boolean running = false;
 
 		public LifecycleMessageSource(MessageSource<T> delegate) {

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinder.java

@@ -16,7 +16,6 @@
 
 package org.springframework.cloud.stream.binder.test;
 
-import java.util.Collections;
 import java.util.function.Consumer;
 
 import org.springframework.beans.factory.BeanFactory;
@@ -30,12 +29,15 @@ import org.springframework.cloud.stream.binder.test.TestChannelBinderProvisioner
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.core.AttributeAccessor;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.acks.AcknowledgmentCallback;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.handler.BridgeHandler;
 import org.springframework.integration.support.DefaultErrorMessageStrategy;
 import org.springframework.integration.support.ErrorMessageStrategy;
+import org.springframework.integration.support.MapBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
@@ -110,8 +112,10 @@ public class TestChannelBinder extends
 	private Message<?> lastError;
 
 	private MessageSource<?> messageSourceDelegate = () -> new GenericMessage<>(
-			"polled data",
-			Collections.singletonMap(MessageHeaders.CONTENT_TYPE, "text/plain"));
+			"polled data", new MapBuilder()
+					.put(MessageHeaders.CONTENT_TYPE, "text/plain")
+					.put(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, (AcknowledgmentCallback) status -> {
+					}).get());
 
 	public TestChannelBinder(TestChannelBinderProvisioner provisioningProvider) {
 		super(new String[] {}, provisioningProvider);