feat: add scheduled executor pool size configuration for SQS batching operations

Author: khc41

docs/src/main/asciidoc/sqs.adoc

@@ -1008,6 +1008,7 @@ The Spring Boot Starter for SQS provides the following auto-configuration proper
 | `spring.cloud.aws.sqs.batch.wait-time-seconds` | Wait time for receiveMessage requests in batch. | No | AWS SDK default
 | `spring.cloud.aws.sqs.batch.system-attribute-names` | System attributes to request for receiveMessage calls. | No | None
 | `spring.cloud.aws.sqs.batch.attribute-names` | Message attributes to request for receiveMessage calls. | No | None
+| `spring.cloud.aws.sqs.batch.scheduled-executor-pool-size` | Size of the scheduled thread pool for batching operations. | No | 5
 |===
 
 

spring-cloud-aws-autoconfigure/src/main/java/io/awspring/cloud/autoconfigure/sqs/SqsAutoConfiguration.java

@@ -35,6 +35,7 @@
 import io.awspring.cloud.sqs.support.observation.SqsTemplateObservation;
 import io.micrometer.observation.ObservationRegistry;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
@@ -60,6 +61,7 @@
  * @author Maciej Walkowiak
  * @author Wei Jiang
  * @author Dongha Kim
+ * @author khc41
  * @since 3.0
  */
 @AutoConfiguration
@@ -91,14 +93,27 @@ public SqsAsyncClient sqsAsyncClient(AwsClientBuilderConfigurer awsClientBuilder
 	@ConditionalOnProperty(name = "spring.cloud.aws.sqs.batch.enabled", havingValue = "true")
 	@Bean
 	@Primary
-	public SqsAsyncClient batchSqsAsyncClient(SqsAsyncClient sqsAsyncClient) {
-		SqsAsyncBatchManager batchManager = createBatchManager(sqsAsyncClient);
+	public SqsAsyncClient batchSqsAsyncClient(SqsAsyncClient sqsAsyncClient,
+			ScheduledExecutorService sqsBatchingScheduledExecutor) {
+		SqsAsyncBatchManager batchManager = createBatchManager(sqsAsyncClient, sqsBatchingScheduledExecutor);
 		return new BatchingSqsClientAdapter(batchManager);
 	}
 
-	private SqsAsyncBatchManager createBatchManager(SqsAsyncClient sqsAsyncClient) {
-		return SqsAsyncBatchManager.builder().client(sqsAsyncClient)
-				.scheduledExecutor(Executors.newScheduledThreadPool(5))
+	@ConditionalOnProperty(name = "spring.cloud.aws.sqs.batch.enabled", havingValue = "true")
+	@ConditionalOnMissingBean(name = "sqsBatchingScheduledExecutor")
+	@Bean
+	public ScheduledExecutorService sqsBatchingScheduledExecutor() {
+		int poolSize = this.sqsProperties.getBatch().getScheduledExecutorPoolSize();
+		if (poolSize <= 0) {
+			throw new IllegalArgumentException(
+					"scheduledExecutorPoolSize must be greater than 0, but was: " + poolSize);
+		}
+		return Executors.newScheduledThreadPool(poolSize);
+	}
+
+	private SqsAsyncBatchManager createBatchManager(SqsAsyncClient sqsAsyncClient,
+			ScheduledExecutorService scheduledExecutor) {
+		return SqsAsyncBatchManager.builder().client(sqsAsyncClient).scheduledExecutor(scheduledExecutor)
 				.overrideConfiguration(this::configurationProperties).build();
 	}
 

spring-cloud-aws-autoconfigure/src/main/java/io/awspring/cloud/autoconfigure/sqs/SqsProperties.java

@@ -235,6 +235,15 @@ public static class Batch {
 		@Nullable
 		private List<String> attributeNames;
 
+		/**
+		 * The size of the scheduled thread pool used for batching operations. This thread pool handles periodic batch
+		 * sending and other scheduled tasks.
+		 * 
+		 * <p>
+		 * Default is {@code 5}.
+		 */
+		private int scheduledExecutorPoolSize = 5;
+
 		public boolean isEnabled() {
 			return enabled;
 		}
@@ -297,6 +306,14 @@ public void setAttributeNames(List<String> attributeNames) {
 			this.attributeNames = attributeNames;
 		}
 
+		public int getScheduledExecutorPoolSize() {
+			return scheduledExecutorPoolSize;
+		}
+
+		public void setScheduledExecutorPoolSize(int scheduledExecutorPoolSize) {
+			this.scheduledExecutorPoolSize = scheduledExecutorPoolSize;
+		}
+
 	}
 
 }

spring-cloud-aws-autoconfigure/src/test/java/io/awspring/cloud/autoconfigure/sqs/SqsAutoConfigurationTest.java

@@ -408,6 +408,7 @@ void sqsBatchConfigurationPropertiesWithDefaults() {
 			assertThat(batchConfig.getWaitTimeSeconds()).isNull();
 			assertThat(batchConfig.getSystemAttributeNames()).isNull();
 			assertThat(batchConfig.getAttributeNames()).isNull();
+			assertThat(batchConfig.getScheduledExecutorPoolSize()).isEqualTo(5);
 
 			assertThat(context).hasSingleBean(SqsAsyncClient.class);
 			SqsAsyncClient client = context.getBean(SqsAsyncClient.class);
@@ -457,6 +458,42 @@ void sqsBatchConfigurationWithAttributeNames() {
 				});
 	}
 
+	@Test
+	void sqsBatchConfigurationWithDefaultScheduledExecutorPoolSize() {
+		this.contextRunner.withPropertyValues("spring.cloud.aws.sqs.batch.enabled:true").run(context -> {
+			assertThat(context).hasSingleBean(SqsProperties.class);
+			SqsProperties sqsProperties = context.getBean(SqsProperties.class);
+			SqsProperties.Batch batchConfig = sqsProperties.getBatch();
+
+			assertThat(batchConfig.isEnabled()).isTrue();
+			assertThat(batchConfig.getScheduledExecutorPoolSize()).isEqualTo(5);
+
+			assertThat(context).hasBean("sqsBatchingScheduledExecutor");
+		});
+	}
+
+	@Test
+	void sqsBatchConfigurationWithCustomScheduledExecutorPoolSize() {
+		this.contextRunner.withPropertyValues("spring.cloud.aws.sqs.batch.enabled:true",
+				"spring.cloud.aws.sqs.batch.scheduled-executor-pool-size:10").run(context -> {
+					assertThat(context).hasSingleBean(SqsProperties.class);
+					SqsProperties sqsProperties = context.getBean(SqsProperties.class);
+					SqsProperties.Batch batchConfig = sqsProperties.getBatch();
+
+					assertThat(batchConfig.isEnabled()).isTrue();
+					assertThat(batchConfig.getScheduledExecutorPoolSize()).isEqualTo(10);
+
+					assertThat(context).hasBean("sqsBatchingScheduledExecutor");
+				});
+	}
+
+	@Test
+	void sqsBatchConfigurationWithBatchDisabledDoesNotCreateScheduledExecutor() {
+		this.contextRunner.withPropertyValues("spring.cloud.aws.sqs.batch.enabled:false").run(context -> {
+			assertThat(context).doesNotHaveBean("sqsBatchingScheduledExecutor");
+		});
+	}
+
 	@Configuration(proxyBeanMethods = false)
 	static class CustomComponentsConfiguration {