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

Aperta
  • Qualcosa non va?
#3,316 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
3/5
Tempo stimato
1-2 giorni
Idoneità per principianti
74/100
Tipo di issue
Bug
Chiarezza
Specificata chiaramente
Stato di attività
Attiva
Stack tecnologico
java
Ambito
backend-api-design

Direzione di ricerca

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.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

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.

Lingua principale
Java
Stelle
1.1k
Fork
651
Merge medio
1g 15h
PR unite (30g)
17

Preparare l'ambiente

Questo progetto non fornisce container di sviluppo, Dockerfile né guida per i contributori, quindi l'ambiente è a tuo carico: parti dal suo README e consulta la nostra guida al primo contributo per i passaggi generali.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di spring-cloud/spring-cloud-stream

Tutte le issue di spring-cloud/spring-cloud-stream

Issue simili

Altre issue su Java

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.