DefaultPollableMessageSource: invoke afterSendCompletion for already-applied interceptors when a message is discarded

Start in DefaultPollableMessageSource, especially ReceiveAdvice.invoke, setSource(), and the poll() overload that converts to ParameterizedTypeReference. Trace how interceptors are applied and where completion callbacks run for discarded messages and conversion exceptions. Add focused tests based on the two reproductions; done means each interceptor whose preSend succeeds gets exactly one completion callback on both paths, and the tests pass.

Describe the issue

Follow-up to #3131 / #3253. Since 5.0.4, DefaultPollableMessageSource invokes the full ChannelInterceptor contract, but there are two paths where interceptors whose preSend already succeeded never receive afterSendCompletion, unlike AbstractMessageChannel (which calls triggerAfterSendCompletion for the already-applied interceptors):

1. A later interceptor returns null from preSend: the preSend loop in ReceiveAdvice (the local class created in setSource() that wraps MessageSource.receive()) returns null, poll() returns false, and no afterSendCompletion is invoked for the interceptors already applied.
2. Payload conversion fails in receive(ParameterizedTypeReference): the MessageConversionException is thrown after ReceiveAdvice has run preSend, but before the try/finally of poll(), so afterSendCompletion is skipped for all interceptors.

Interceptors that open resources in preSend and release them in afterSendCompletion (tracing scope, MDC, thread locals) leak them on the poller thread. A typical case is a filtering interceptor that discards messages by returning null, registered after a tracing interceptor.

To Reproduce

Plain JUnit test, no application context needed (spring-cloud-stream 5.0.4). Both tests fail at the completion assertion:

package com.example.demoStreamPollable;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.util.ArrayList;
import java.util.List;

import org.junit.jupiter.api.Test;
import org.springframework.cloud.stream.binder.DefaultPollableMessageSource;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.converter.StringMessageConverter;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.GenericMessage;

class PollableInterceptorCompletionTests {

	private final List<String> events = new ArrayList<>();

	@Test
	void afterSendCompletionIsInvokedForAppliedInterceptorsWhenLaterPreSendReturnsNull() {
		DefaultPollableMessageSource source = newSource();
		source.setSource(() -> new GenericMessage<>("foo"));
		source.addInterceptor(recorder("A"));
		source.addInterceptor(new ChannelInterceptor() {
			@Override
			public Message<?> preSend(Message<?> message, MessageChannel channel) {
				return null; // e.g. a filtering interceptor discarding the message
			}
		});

		assertThat(source.poll(message -> { })).isFalse();

		// Same contract as AbstractMessageChannel: A's preSend succeeded, so it must be completed
		assertThat(this.events).containsExactly("A.preSend", "A.afterSendCompletion(sent=false)");
	}

	@Test
	void afterSendCompletionIsInvokedWhenPayloadConversionFails() {
		DefaultPollableMessageSource source = newSource();
		source.setSource(() -> new GenericMessage<>("not a Foo"));
		source.addInterceptor(recorder("A"));

		assertThatThrownBy(() -> source.poll(message -> { }, new ParameterizedTypeReference<Foo>() { }))
			.isInstanceOf(RuntimeException.class);

		assertThat(this.events).contains("A.preSend").anyMatch(e -> e.startsWith("A.afterSendCompletion"));
	}

	private static DefaultPollableMessageSource newSource() {
		return new DefaultPollableMessageSource(new CompositeMessageConverter(List.of(new StringMessageConverter())));
	}

	private ChannelInterceptor recorder(String name) {
		return new ChannelInterceptor() {
			@Override
			public Message<?> preSend(Message<?> message, MessageChannel channel) {
				events.add(name + ".preSend");
				return message;
			}

			@Override
			public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
				events.add(name + ".afterSendCompletion(sent=" + sent + ")");
			}
		};
	}

	record Foo(String value) {
	}

}

Version of the framework

Spring Cloud Stream 5.0.4 (Spring Cloud 2025.1.4), Spring Boot 4.1.2.

Expected behavior

Every interceptor whose preSend returned a message receives afterSendCompletion exactly once, also when the message is discarded by a later interceptor or when conversion fails, consistent with AbstractMessageChannel.

Additional context

Proposal:

• In the preSend loop of ReceiveAdvice.invoke, track the interceptors whose preSend returned a message, and call afterSendCompletion(message, DUMMY_CHANNEL, false, null) in reverse order when a later one returns null, the same as AbstractMessageChannel.
• Run the completion when conversion fails as well (for example, by moving the conversion inside the try of poll(), or by completing with the exception).

Happy to submit a PR.