Remove BackPressureHandlerLimiter from the library and make it user-code in tests only (#1251)

Author: loicrouchon

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractContainerOptions.java

@@ -22,6 +22,7 @@
 import io.micrometer.observation.ObservationConvention;
 import io.micrometer.observation.ObservationRegistry;
 import java.time.Duration;
+import java.util.function.Supplier;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.lang.Nullable;
 import org.springframework.retry.backoff.BackOffPolicy;
@@ -57,7 +58,7 @@ public abstract class AbstractContainerOptions<O extends ContainerOptions<O, B>,
 
 	private final BackPressureMode backPressureMode;
 
-	private final BackPressureLimiter backPressureLimiter;
+	private final Supplier<BackPressureHandler> backPressureHandlerSupplier;
 
 	private final ListenerMode listenerMode;
 
@@ -95,7 +96,7 @@ protected AbstractContainerOptions(Builder<?, ?> builder) {
 		this.listenerShutdownTimeout = builder.listenerShutdownTimeout;
 		this.acknowledgementShutdownTimeout = builder.acknowledgementShutdownTimeout;
 		this.backPressureMode = builder.backPressureMode;
-		this.backPressureLimiter = builder.backPressureLimiter;
+		this.backPressureHandlerSupplier = builder.backPressureHandlerSupplier;
 		this.listenerMode = builder.listenerMode;
 		this.messageConverter = builder.messageConverter;
 		this.acknowledgementMode = builder.acknowledgementMode;
@@ -174,8 +175,8 @@ public BackPressureMode getBackPressureMode() {
 	}
 
 	@Override
-	public BackPressureLimiter getBackPressureLimiter() {
-		return this.backPressureLimiter;
+	public Supplier<BackPressureHandler> getBackPressureHandlerSupplier() {
+		return this.backPressureHandlerSupplier;
 	}
 
 	@Override
@@ -250,7 +251,7 @@ protected abstract static class Builder<B extends ContainerOptionsBuilder<B, O>,
 
 		private static final BackPressureMode DEFAULT_THROUGHPUT_CONFIGURATION = BackPressureMode.AUTO;
 
-		private static final BackPressureLimiter DEFAULT_BACKPRESSURE_LIMITER = null;
+		private static final Supplier<BackPressureHandler> DEFAULT_BACKPRESSURE_LIMITER = null;
 
 		private static final ListenerMode DEFAULT_MESSAGE_DELIVERY_STRATEGY = ListenerMode.SINGLE_MESSAGE;
 
@@ -276,7 +277,7 @@ protected abstract static class Builder<B extends ContainerOptionsBuilder<B, O>,
 
 		private BackPressureMode backPressureMode = DEFAULT_THROUGHPUT_CONFIGURATION;
 
-		private BackPressureLimiter backPressureLimiter = DEFAULT_BACKPRESSURE_LIMITER;
+		private Supplier<BackPressureHandler> backPressureHandlerSupplier = DEFAULT_BACKPRESSURE_LIMITER;
 
 		private Duration listenerShutdownTimeout = DEFAULT_LISTENER_SHUTDOWN_TIMEOUT;
 
@@ -320,7 +321,7 @@ protected Builder(AbstractContainerOptions<?, ?> options) {
 			this.listenerShutdownTimeout = options.listenerShutdownTimeout;
 			this.acknowledgementShutdownTimeout = options.acknowledgementShutdownTimeout;
 			this.backPressureMode = options.backPressureMode;
-			this.backPressureLimiter = options.backPressureLimiter;
+			this.backPressureHandlerSupplier = options.backPressureHandlerSupplier;
 			this.listenerMode = options.listenerMode;
 			this.messageConverter = options.messageConverter;
 			this.acknowledgementMode = options.acknowledgementMode;
@@ -423,8 +424,8 @@ public B backPressureMode(BackPressureMode backPressureMode) {
 		}
 
 		@Override
-		public B backPressureLimiter(BackPressureLimiter backPressureLimiter) {
-			this.backPressureLimiter = backPressureLimiter;
+		public B backPressureHandlerSupplier(Supplier<BackPressureHandler> backPressureHandlerSupplier) {
+			this.backPressureHandlerSupplier = backPressureHandlerSupplier;
 			return self();
 		}
 

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractPipelineMessageListenerContainer.java

@@ -232,17 +232,14 @@ private TaskExecutor validateCustomExecutor(TaskExecutor taskExecutor) {
 
 	protected BackPressureHandler createBackPressureHandler() {
 		O containerOptions = getContainerOptions();
-		List<BackPressureHandler> backPressureHandlers = new ArrayList<>(2);
+		if (containerOptions.getBackPressureHandlerSupplier() != null) {
+			return containerOptions.getBackPressureHandlerSupplier().get();
+		}
 		Duration acquireTimeout = containerOptions.getMaxDelayBetweenPolls();
 		int batchSize = containerOptions.getMaxMessagesPerPoll();
-		backPressureHandlers.add(SemaphoreBackPressureHandler.builder().batchSize(batchSize)
+		return SemaphoreBackPressureHandler.builder().batchSize(batchSize)
 				.totalPermits(containerOptions.getMaxConcurrentMessages()).acquireTimeout(acquireTimeout)
-				.throughputConfiguration(containerOptions.getBackPressureMode()).build());
-		if (containerOptions.getBackPressureLimiter() != null) {
-			backPressureHandlers.add(new BackPressureHandlerLimiter(containerOptions.getBackPressureLimiter(),
-					acquireTimeout, containerOptions.getStandbyLimitPollingInterval(), batchSize));
-		}
-		return new CompositeBackPressureHandler(backPressureHandlers, batchSize);
+				.throughputConfiguration(containerOptions.getBackPressureMode()).build();
 	}
 
 	protected TaskExecutor createSourcesTaskExecutor() {

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/BackPressureHandler.java

@@ -57,6 +57,18 @@ public interface BackPressureHandler {
 	 */
 	void release(int amount, ReleaseReason reason);
 
+	/**
+	 * Release the specified amount of permits. Each message that has been processed should release one permit, whether
+	 * processing was successful or not.
+	 * @param amount the amount of permits to release.
+	 *
+	 * @deprecated This method is deprecated and will not be called by the Spring Cloud AWS SQS listener anymore.
+	 * Implement {@link #release(int, ReleaseReason)} instead.
+	 */
+	@Deprecated
+	default void release(int amount) {
+	}
+
 	/**
 	 * Attempts to acquire all permits up to the specified timeout. If successful, means all permits were returned and
 	 * thus no activity is left in the {@link io.awspring.cloud.sqs.listener.source.MessageSource}.

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/BackPressureHandlerLimiter.java

@@ -30,4 +30,28 @@ public interface BatchAwareBackPressureHandler extends BackPressureHandler {
 	 * @throws InterruptedException if the Thread is interrupted while waiting for permits.
 	 */
 	int requestBatch() throws InterruptedException;
+
+	/**
+	 * Release a batch of permits. This has the semantics of letting the {@link BackPressureHandler} know that all
+	 * permits from a batch are being released, in opposition to {@link #release(int)} in which any number of permits
+	 * can be specified.
+	 *
+	 * @deprecated This method is deprecated and will not be called by the Spring Cloud AWS SQS listener anymore.
+	 * Implement {@link BackPressureHandler#release(int, ReleaseReason)} instead.
+	 */
+	@Deprecated
+	default void releaseBatch() {
+	}
+
+	/**
+	 * Return the configured batch size for this handler.
+	 * @return the batch size.
+	 *
+	 * @deprecated This method is deprecated and will not be used by the Spring Cloud AWS SQS listener anymore.
+	 */
+	@Deprecated
+	default int getBatchSize() {
+		return 0;
+	}
+
 }

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/BackPressureLimiter.java

@@ -22,6 +22,7 @@
 import io.micrometer.observation.ObservationRegistry;
 import java.time.Duration;
 import java.util.Collection;
+import java.util.function.Supplier;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.lang.Nullable;
 import org.springframework.retry.backoff.BackOffPolicy;
@@ -63,8 +64,6 @@ public interface ContainerOptions<O extends ContainerOptions<O, B>, B extends Co
 	/**
 	 * {@return the amount of time to wait before checking again for the current limit when the queue processing is on
 	 * standby} Default is 100 milliseconds.
-	 *
-	 * @see BackPressureLimiter#limit()
 	 */
 	Duration getStandbyLimitPollingInterval();
 
@@ -138,10 +137,10 @@ default BackOffPolicy getPollBackOffPolicy() {
 	BackPressureMode getBackPressureMode();
 
 	/**
-	 * Return the {@link BackPressureLimiter} for this container.
-	 * @return the backpressure limiter.
+	 * Return the a {@link Supplier} to create a {@link BackPressureHandler} for this container.
+	 * @return the BackPressureHandler supplier.
 	 */
-	BackPressureLimiter getBackPressureLimiter();
+	Supplier<BackPressureHandler> getBackPressureHandlerSupplier();
 
 	/**
 	 * Return the {@link ListenerMode} mode for this container.

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/BatchAwareBackPressureHandler.java

@@ -20,6 +20,7 @@
 import io.awspring.cloud.sqs.support.converter.MessagingMessageConverter;
 import io.micrometer.observation.ObservationRegistry;
 import java.time.Duration;
+import java.util.function.Supplier;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.retry.backoff.BackOffPolicy;
 
@@ -63,7 +64,6 @@ public interface ContainerOptionsBuilder<B extends ContainerOptionsBuilder<B, O>
 	 *
 	 * @param standbyLimitPollingInterval the limit polling interval when the queue processing is on standby.
 	 * @return this instance.
-	 * @see BackPressureLimiter#limit()
 	 */
 	B standbyLimitPollingInterval(Duration standbyLimitPollingInterval);
 
@@ -157,12 +157,12 @@ default B pollBackOffPolicy(BackOffPolicy pollBackOffPolicy) {
 	B backPressureMode(BackPressureMode backPressureMode);
 
 	/**
-	 * Set the {@link BackPressureLimiter} for this container. Default is {@code null}.
+	 * Set the {@link Supplier} of {@link BackPressureHandler} for this container. Default is {@code null}.
 	 *
-	 * @param backPressureLimiter the backpressure limiter.
+	 * @param backPressureHandlerSupplier the BackPressureHandler supplier.
 	 * @return this instance.
 	 */
-	B backPressureLimiter(BackPressureLimiter backPressureLimiter);
+	B backPressureHandlerSupplier(Supplier<BackPressureHandler> backPressureHandlerSupplier);
 
 	/**
 	 * Set the maximum interval between acknowledgements for batch acknowledgements. The default depends on the specific