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

Đang mở
  • Có gì không ổn?
#3,316 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

Chưa có ai nhận issue này.

Đánh giá

Độ khó
3/5
Thời gian dự kiến
1-2 ngày
Mức phù hợp với người mới
74/100
Loại issue
Lỗi
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
java
Lĩnh vực
backend-api-design

Hướng nghiên cứu

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.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

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.

Ngôn ngữ chính
Java
Star
1.1k
Fork
651
Merge trung bình
2 ngày 15 giờ
Pull request đã merge (30 ngày)
8

Chuẩn bị môi trường

Dự án này không cung cấp dev container, Dockerfile hay hướng dẫn đóng góp, nên bạn cần tự thiết lập môi trường: hãy bắt đầu từ README và xem hướng dẫn đóng góp lần đầu của chúng tôi để biết các bước chung.

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của spring-cloud/spring-cloud-stream

Tất cả issue của spring-cloud/spring-cloud-stream

Issue tương tự

Thêm issue về Java

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.