Add tests for ConcurrencyLimiterBlockingBackPressureHandler (#1251)

Author: loicrouchon

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

@@ -147,7 +147,7 @@ public static CompositeBackPressureHandler compositeBackPressureHandler(Containe
 	public static ConcurrencyLimiterBlockingBackPressureHandler concurrencyLimiterBackPressureHandler(
 			ContainerOptions<?, ?> options) {
 		return ConcurrencyLimiterBlockingBackPressureHandler.builder().batchSize(options.getMaxMessagesPerPoll())
-				.totalPermits(options.getMaxConcurrentMessages()).throughputConfiguration(options.getBackPressureMode())
+				.totalPermits(options.getMaxConcurrentMessages())
 				.acquireTimeout(options.getMaxDelayBetweenPolls()).build();
 	}
 

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

@@ -126,8 +126,6 @@ public static class Builder {
 
 		private Duration acquireTimeout;
 
-		private BackPressureMode backPressureMode;
-
 		public Builder batchSize(int batchSize) {
 			this.batchSize = batchSize;
 			return this;
@@ -143,14 +141,9 @@ public Builder acquireTimeout(Duration acquireTimeout) {
 			return this;
 		}
 
-		public Builder throughputConfiguration(BackPressureMode backPressureConfiguration) {
-			this.backPressureMode = backPressureConfiguration;
-			return this;
-		}
-
 		public ConcurrencyLimiterBlockingBackPressureHandler build() {
 			Assert.noNullElements(
-					Arrays.asList(this.batchSize, this.totalPermits, this.acquireTimeout, this.backPressureMode),
+					Arrays.asList(this.batchSize, this.totalPermits, this.acquireTimeout),
 					"Missing configuration");
 			Assert.isTrue(this.batchSize > 0, "The batch size must be greater than 0");
 			Assert.isTrue(this.totalPermits >= this.batchSize, "Total permits must be greater than the batch size");

spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/ConcurrencyLimiterBlockingBackPressureHandlerTest.java

@@ -0,0 +1,46 @@
+package io.awspring.cloud.sqs.listener;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ConcurrencyLimiterBlockingBackPressureHandlerTest {
+
+	private static final int BATCH_SIZE = 5;
+	private static final int TOTAL_PERMITS = 10;
+
+	private ConcurrencyLimiterBlockingBackPressureHandler handler;
+
+    @BeforeEach
+    void setUp() {
+        handler = ConcurrencyLimiterBlockingBackPressureHandler.builder()
+                .totalPermits(TOTAL_PERMITS)
+                .batchSize(BATCH_SIZE)
+                .acquireTimeout(Duration.ofMillis(100))
+                .build();
+    }
+
+    @Test
+    void request_shouldAcquirePermits() throws InterruptedException {
+		// Requesting a first batch should acquire the permits
+        assertThat(handler.request(BATCH_SIZE)).isEqualTo(BATCH_SIZE);
+		// Requesting a second batch should acquire the remaining permits
+        assertThat(handler.request(BATCH_SIZE)).isEqualTo(BATCH_SIZE);
+        // No permits left
+        assertThat(handler.request(1)).isZero();
+    }
+
+    @Test
+    void release_shouldAllowFurtherRequests() throws InterruptedException {
+		// Given all permits are acquired
+		assertThat(handler.request(TOTAL_PERMITS)).isEqualTo(TOTAL_PERMITS);
+        assertThat(handler.request(1)).isZero();
+		// When releasing some permits, new requests should be allowed
+        handler.release(3, BackPressureHandler.ReleaseReason.PROCESSED);
+		assertThat(handler.request(5)).isEqualTo(3); // Only 3 permits were released so far
+    }
+}
+

spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/ThroughputBackPressureHandlerTest.java

@@ -40,7 +40,7 @@ void lowThroughputMode_shouldReturnZeroUntilRelease(BackPressureHandler.ReleaseR
 		int batchSize = 5;
 		assertThat(handler.request(batchSize)).isEqualTo(batchSize);
 		// When a second batch is requested, it should return zero permits (because low throughput mode)
-		assertThat(handler.request(batchSize)).isEqualTo(0);
+		assertThat(handler.request(batchSize)).isZero();
 		// When a batch is requested after a release, the expected permits should be
 		// returned depending on the release reason
 		handler.release(1, releaseReason);
@@ -64,13 +64,13 @@ void highThroughputMode_shouldAllowMultipleConcurrentRequests() throws Interrupt
 		handler.release(5, BackPressureHandler.ReleaseReason.NONE_FETCHED);
 		assertThat(handler.request(batchSize)).isEqualTo(batchSize);
 		// And subsequent requests should return zero permits until the current batch finishes with NONE_FETCHED
-		assertThat(handler.request(batchSize)).isEqualTo(0);
-		assertThat(handler.request(batchSize)).isEqualTo(0);
+		assertThat(handler.request(batchSize)).isZero();
+		assertThat(handler.request(batchSize)).isZero();
 		handler.release(5, BackPressureHandler.ReleaseReason.NONE_FETCHED);
 		assertThat(handler.request(batchSize)).isEqualTo(5);
 		// or until it (the current batch) finishes with PARTIAL_FETCH
-		assertThat(handler.request(batchSize)).isEqualTo(0);
-		assertThat(handler.request(batchSize)).isEqualTo(0);
+		assertThat(handler.request(batchSize)).isZero();
+		assertThat(handler.request(batchSize)).isZero();
 		handler.release(3, BackPressureHandler.ReleaseReason.PARTIAL_FETCH);
 		assertThat(handler.request(batchSize)).isEqualTo(5);
 	}
@@ -79,7 +79,7 @@ void highThroughputMode_shouldAllowMultipleConcurrentRequests() throws Interrupt
 	void drain_shouldSetDrainedAndReturnTrue() throws InterruptedException {
 		boolean result = handler.drain(Duration.ofSeconds(1));
 		assertThat(result).isTrue();
-		assertThat(handler.request(5)).isEqualTo(0);
+		assertThat(handler.request(5)).isZero();
 	}
 
 }