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

Abierto
  • ¿Algo no está bien?
#3,316 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
3/5
Tiempo estimado
1-2 días
Aptitud para principiantes
74/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
java
Área
backend-api-design

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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.

Lenguaje dominante
Java
Estrellas
1.1k
Forks
651
Merge medio
1 d 16 h
PR fusionados (30 d)
16

Preparar el entorno

Este proyecto no incluye contenedor de desarrollo, Dockerfile ni guía de contribución, así que la configuración corre por tu cuenta: empieza por su README y consulta nuestra guía para la primera contribución para los pasos generales.

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de spring-cloud/spring-cloud-stream

Todos los issues de spring-cloud/spring-cloud-stream

Issues similares

Más issues de Java

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.