Introduce factory methods for creating back-pressure handlers (#1251)

Author: loicrouchon

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

@@ -222,7 +222,7 @@ protected abstract static class Builder<B extends ContainerOptionsBuilder<B, O>,
 
 		private static final BackPressureMode DEFAULT_THROUGHPUT_CONFIGURATION = BackPressureMode.AUTO;
 
-		private static final BackPressureHandlerFactory DEFAULT_BACKPRESSURE_FACTORY = buildDefaultBackPressureHandlerFactory();
+		private static final BackPressureHandlerFactory DEFAULT_BACKPRESSURE_FACTORY = BackPressureHandlerFactory::semaphoreBackPressureHandler;
 
 		private static final ListenerMode DEFAULT_MESSAGE_DELIVERY_STRATEGY = ListenerMode.SINGLE_MESSAGE;
 
@@ -428,12 +428,6 @@ private static BackOffPolicy buildDefaultBackOffPolicy() {
 			return BackOffPolicyBuilder.newBuilder().multiplier(DEFAULT_BACK_OFF_MULTIPLIER)
 					.delay(DEFAULT_BACK_OFF_DELAY).maxDelay(DEFAULT_BACK_OFF_MAX_DELAY).build();
 		}
-
-		private static BackPressureHandlerFactory buildDefaultBackPressureHandlerFactory() {
-			return options -> SemaphoreBackPressureHandler.builder().batchSize(options.getMaxMessagesPerPoll())
-					.totalPermits(options.getMaxConcurrentMessages()).acquireTimeout(options.getMaxDelayBetweenPolls())
-					.throughputConfiguration(options.getBackPressureMode()).build();
-		}
 	}
 
 }

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

@@ -15,6 +15,10 @@
  */
 package io.awspring.cloud.sqs.listener;
 
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
 /**
  * A factory for creating {@link BackPressureHandler} for managing queue consumption backpressure. Implementations can
  * configure each the {@link BackPressureHandler} according to its strategies, using the provided
@@ -82,4 +86,99 @@ public interface BackPressureHandlerFactory {
 	 * @return the created BackPressureHandler
 	 */
 	BackPressureHandler createBackPressureHandler(ContainerOptions<?, ?> containerOptions);
+
+	/**
+	 * Creates a new {@link SemaphoreBackPressureHandler} instance based on the provided {@link ContainerOptions}.
+	 *
+	 * @param options the container options.
+	 * @return the created SemaphoreBackPressureHandler.
+	 */
+	static BatchAwareBackPressureHandler semaphoreBackPressureHandler(ContainerOptions<?, ?> options) {
+		return SemaphoreBackPressureHandler.builder().batchSize(options.getMaxMessagesPerPoll())
+				.totalPermits(options.getMaxConcurrentMessages()).acquireTimeout(options.getMaxDelayBetweenPolls())
+				.throughputConfiguration(options.getBackPressureMode()).build();
+	}
+
+	/**
+	 * Creates a new {@link BackPressureHandler} instance based on the provided {@link ContainerOptions} combining a
+	 * {@link ConcurrencyLimiterBlockingBackPressureHandler}, a {@link ThroughputBackPressureHandler} and a
+	 * {@link FullBatchBackPressureHandler}. The exact combination of depends on the given {@link ContainerOptions}.
+	 *
+	 * @param options the container options.
+	 * @param maxIdleWaitTime the maximum amount of time to wait for a permit to be released in case no permits were
+	 *     obtained.
+	 * @return the created SemaphoreBackPressureHandler.
+	 */
+	static BatchAwareBackPressureHandler concurrencyLimiterBackPressureHandler(ContainerOptions<?, ?> options,
+			Duration maxIdleWaitTime) {
+		BackPressureMode backPressureMode = options.getBackPressureMode();
+
+		var concurrencyLimiterBlockingBackPressureHandler = concurrencyLimiterBackPressureHandler2(options);
+		if (backPressureMode == BackPressureMode.FIXED_HIGH_THROUGHPUT) {
+			return concurrencyLimiterBlockingBackPressureHandler;
+		}
+		var backPressureHandlers = new ArrayList<BackPressureHandler>();
+		backPressureHandlers.add(concurrencyLimiterBlockingBackPressureHandler);
+
+		// The ThroughputBackPressureHandler should run second in the chain as it is non-blocking.
+		// Running it first would result in more polls as it would potentially limit the
+		// ConcurrencyLimiterBlockingBackPressureHandler to a lower amount of requested permits
+		// which means the ConcurrencyLimiterBlockingBackPressureHandler blocking behavior would
+		// not be optimally leveraged.
+		if (backPressureMode == BackPressureMode.AUTO
+				|| backPressureMode == BackPressureMode.ALWAYS_POLL_MAX_MESSAGES) {
+			backPressureHandlers.add(throughputBackPressureHandler(options));
+		}
+
+		// The FullBatchBackPressureHandler should run last in the chain to ensure that a full batch is requested or not
+		if (backPressureMode == BackPressureMode.ALWAYS_POLL_MAX_MESSAGES) {
+			backPressureHandlers.add(fullBatchBackPressureHandler(options));
+		}
+		return compositeBackPressureHandler(options, maxIdleWaitTime, backPressureHandlers);
+	}
+
+	/**
+	 * Creates a new {@link ConcurrencyLimiterBlockingBackPressureHandler} instance based on the provided
+	 * {@link ContainerOptions}.
+	 *
+	 * @param options the container options.
+	 * @return the created ConcurrencyLimiterBlockingBackPressureHandler.
+	 */
+	static CompositeBackPressureHandler compositeBackPressureHandler(ContainerOptions<?, ?> options,
+			Duration maxIdleWaitTime, List<BackPressureHandler> backPressureHandlers) {
+		return new CompositeBackPressureHandler(List.copyOf(backPressureHandlers), options.getMaxMessagesPerPoll(),
+				maxIdleWaitTime);
+	}
+
+	/**
+	 * Creates a new {@link ConcurrencyLimiterBlockingBackPressureHandler} instance based on the provided
+	 * {@link ContainerOptions}.
+	 * @param options the container options.
+	 * @return the created ConcurrencyLimiterBlockingBackPressureHandler.
+	 */
+	static ConcurrencyLimiterBlockingBackPressureHandler concurrencyLimiterBackPressureHandler2(
+			ContainerOptions<?, ?> options) {
+		return ConcurrencyLimiterBlockingBackPressureHandler.builder().batchSize(options.getMaxMessagesPerPoll())
+				.totalPermits(options.getMaxConcurrentMessages()).throughputConfiguration(options.getBackPressureMode())
+				.acquireTimeout(options.getMaxDelayBetweenPolls()).build();
+	}
+
+	/**
+	 * Creates a new {@link ThroughputBackPressureHandler} instance based on the provided {@link ContainerOptions}.
+	 * @param options the container options.
+	 * @return the created ThroughputBackPressureHandler.
+	 */
+	static ThroughputBackPressureHandler throughputBackPressureHandler(ContainerOptions<?, ?> options) {
+		return ThroughputBackPressureHandler.builder().batchSize(options.getMaxMessagesPerPoll())
+				.totalPermits(options.getMaxConcurrentMessages()).build();
+	}
+
+	/**
+	 * Creates a new {@link FullBatchBackPressureHandler} instance based on the provided {@link ContainerOptions}.
+	 * @param options the container options.
+	 * @return the created FullBatchBackPressureHandler.
+	 */
+	static FullBatchBackPressureHandler fullBatchBackPressureHandler(ContainerOptions<?, ?> options) {
+		return FullBatchBackPressureHandler.builder().batchSize(options.getMaxMessagesPerPoll()).build();
+	}
 }

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

@@ -24,6 +24,28 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * Composite {@link BackPressureHandler} implementation that delegates the back-pressure handling to a list of
+ * {@link BackPressureHandler}s.
+ * <p>
+ * This class is used to combine multiple back-pressure handlers into a single one. It allows for more complex
+ * back-pressure handling strategies by combining different implementations.
+ * <p>
+ * The order in which the back-pressure handlers are registered in the {@link CompositeBackPressureHandler} is important
+ * as it will affect the blocking and limiting behaviour of the back-pressure handling.
+ * <p>
+ * When {@link #request(int amount)} is called, the first back-pressure handler in the list is called with
+ * {@code amount} as the requested amount of permits. The returned amount of permits (which is less than or equal to the
+ * initial amount) is then passed to the next back-pressure handler in the list. This process of reducing the amount to
+ * request for the next handlers in the chain is called "limiting". This process continues until all back-pressure
+ * handlers have been called or {@literal 0} permits has been returned.
+ * <p>
+ * Once the final amount of available permits have been computed, unused acquired permits on back-pressure handlers (due
+ * to later limiting happening in the chain) are released.
+ * <p>
+ * If no permits were obtained, the {@link #request(int)} method will wait up to {@code noPermitsReturnedWaitTimeout}
+ * for a release of permits before returning.
+ */
 public class CompositeBackPressureHandler implements BatchAwareBackPressureHandler, IdentifiableContainerComponent {
 
 	private static final Logger logger = LoggerFactory.getLogger(CompositeBackPressureHandler.class);
@@ -41,10 +63,10 @@ public class CompositeBackPressureHandler implements BatchAwareBackPressureHandl
 	private String id;
 
 	public CompositeBackPressureHandler(List<BackPressureHandler> backPressureHandlers, int batchSize,
-			Duration waitTimeout) {
+			Duration noPermitsReturnedWaitTimeout) {
 		this.backPressureHandlers = backPressureHandlers;
 		this.batchSize = batchSize;
-		this.noPermitsReturnedWaitTimeout = waitTimeout;
+		this.noPermitsReturnedWaitTimeout = noPermitsReturnedWaitTimeout;
 	}
 
 	@Override

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

@@ -15,6 +15,7 @@
  */
 package io.awspring.cloud.sqs.listener;
 
+import io.awspring.cloud.sqs.listener.source.PollingMessageSource;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.concurrent.Semaphore;
@@ -24,11 +25,10 @@
 import org.springframework.util.Assert;
 
 /**
- * {@link BackPressureHandler} implementation that uses a {@link Semaphore} for handling backpressure.
+ * Blocking {@link BackPressureHandler} implementation that uses a {@link Semaphore} for handling the number of
+ * concurrent messages being processed.
  *
- * @author Tomaz Fernandes
- * @see io.awspring.cloud.sqs.listener.source.PollingMessageSource
- * @since 3.0
+ * @see PollingMessageSource
  */
 public class ConcurrencyLimiterBlockingBackPressureHandler
 		implements BatchAwareBackPressureHandler, IdentifiableContainerComponent {
@@ -43,20 +43,17 @@ public class ConcurrencyLimiterBlockingBackPressureHandler
 
 	private final Duration acquireTimeout;
 
-	private final boolean alwaysPollMaxMessages;
-
 	private String id = getClass().getSimpleName();
 
 	private ConcurrencyLimiterBlockingBackPressureHandler(Builder builder) {
 		this.batchSize = builder.batchSize;
 		this.totalPermits = builder.totalPermits;
 		this.acquireTimeout = builder.acquireTimeout;
-		this.alwaysPollMaxMessages = BackPressureMode.ALWAYS_POLL_MAX_MESSAGES.equals(builder.backPressureMode);
-		this.semaphore = new Semaphore(totalPermits);
 		logger.debug(
 				"ConcurrencyLimiterBlockingBackPressureHandler created with configuration "
-						+ "totalPermits: {}, batchSize: {}, acquireTimeout: {}, an alwaysPollMaxMessages: {}",
-				this.totalPermits, this.batchSize, this.acquireTimeout, this.alwaysPollMaxMessages);
+						+ "totalPermits: {}, batchSize: {}, acquireTimeout: {}",
+				this.totalPermits, this.batchSize, this.acquireTimeout);
+		this.semaphore = new Semaphore(totalPermits);
 	}
 
 	public static Builder builder() {
@@ -81,7 +78,7 @@ public int requestBatch() throws InterruptedException {
 	@Override
 	public int request(int amount) throws InterruptedException {
 		int acquiredPermits = tryAcquire(amount, this.acquireTimeout);
-		if (alwaysPollMaxMessages || acquiredPermits > 0) {
+		if (acquiredPermits > 0) {
 			return acquiredPermits;
 		}
 		int availablePermits = Math.min(this.semaphore.availablePermits(), amount);

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

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2013-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.awspring.cloud.sqs.listener;
+
+import io.awspring.cloud.sqs.listener.source.PollingMessageSource;
+import java.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.Assert;
+
+/**
+ * Non-blocking {@link BackPressureHandler} implementation that ensures the exact batch size is requested.
+ * <p>
+ * If the amount of permits being requested is not equal to the batch size, permits will be limited to {@literal 0}. For
+ * this limiting mechanism to work, the {@link FullBatchBackPressureHandler} must be used in combination with another
+ * {@link BackPressureHandler} and be the last one in the chain of the {@link CompositeBackPressureHandler}
+ *
+ * @see PollingMessageSource
+ */
+public class FullBatchBackPressureHandler implements BatchAwareBackPressureHandler, IdentifiableContainerComponent {
+
+	private static final Logger logger = LoggerFactory.getLogger(FullBatchBackPressureHandler.class);
+
+	private final int batchSize;
+
+	private String id = getClass().getSimpleName();
+
+	private FullBatchBackPressureHandler(Builder builder) {
+		this.batchSize = builder.batchSize;
+		logger.debug("FullBatchBackPressureHandler created with configuration: batchSize: {}", this.batchSize);
+	}
+
+	public static Builder builder() {
+		return new Builder();
+	}
+
+	@Override
+	public void setId(String id) {
+		this.id = id;
+	}
+
+	@Override
+	public String getId() {
+		return this.id;
+	}
+
+	@Override
+	public int requestBatch() throws InterruptedException {
+		return request(this.batchSize);
+	}
+
+	@Override
+	public int request(int amount) throws InterruptedException {
+		if (amount == batchSize) {
+			return amount;
+		}
+		logger.warn("[{}] Could not acquire a full batch ({} / {}), cancelling current poll", this.id, amount,
+				this.batchSize);
+		return 0;
+	}
+
+	@Override
+	public void release(int amount, ReleaseReason reason) {
+		// NO-OP
+	}
+
+	@Override
+	public boolean drain(Duration timeout) {
+		return true;
+	}
+
+	public static class Builder {
+
+		private int batchSize;
+
+		public Builder batchSize(int batchSize) {
+			this.batchSize = batchSize;
+			return this;
+		}
+
+		public FullBatchBackPressureHandler build() {
+			Assert.notNull(this.batchSize, "Missing configuration for batch size");
+			Assert.isTrue(this.batchSize > 0, "The batch size must be greater than 0");
+			return new FullBatchBackPressureHandler(this);
+		}
+	}
+}