1+ /*
2+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+ *
4+ * Licensed under the Apache License, Version 2.0 (the "License").
5+ * You may not use this file except in compliance with the License.
6+ * A copy of the License is located at
7+ *
8+ * http://aws.amazon.com/apache2.0
9+ *
10+ * or in the "license" file accompanying this file. This file is distributed
11+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
12+ * express or implied. See the License for the specific language governing
13+ * permissions and limitations under the License.
14+ */
15+
16+ package software .amazon .awssdk .services .sqs .batchmanager ;
17+
18+ import static com .github .tomakehurst .wiremock .client .WireMock .*;
19+ import static com .github .tomakehurst .wiremock .core .WireMockConfiguration .wireMockConfig ;
20+ import static org .assertj .core .api .Assertions .assertThat ;
21+
22+ import com .github .tomakehurst .wiremock .junit5 .WireMockExtension ;
23+ import com .github .tomakehurst .wiremock .verification .LoggedRequest ;
24+ import com .google .common .util .concurrent .RateLimiter ;
25+ import java .net .URI ;
26+ import java .time .Duration ;
27+ import java .util .List ;
28+ import java .util .concurrent .CountDownLatch ;
29+ import java .util .concurrent .ExecutorService ;
30+ import java .util .concurrent .Executors ;
31+ import java .util .concurrent .TimeUnit ;
32+ import java .util .stream .Collectors ;
33+ import org .junit .jupiter .api .AfterEach ;
34+ import org .junit .jupiter .api .BeforeEach ;
35+ import org .junit .jupiter .api .Test ;
36+ import org .junit .jupiter .api .extension .RegisterExtension ;
37+ import software .amazon .awssdk .auth .credentials .AwsBasicCredentials ;
38+ import software .amazon .awssdk .auth .credentials .StaticCredentialsProvider ;
39+ import software .amazon .awssdk .services .sqs .SqsAsyncClient ;
40+
41+
42+ /**
43+ * Tests the batching efficiency of {@link SqsAsyncBatchManager} under various load scenarios.
44+ */
45+ public class BatchingEfficiencyUnderLoadTest {
46+
47+ private static final String QUEUE_URL = "https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue" ;
48+ private static final int CONCURRENT_THREADS = 50 ;
49+ private static final int MAX_BATCH_SIZE = 10 ;
50+ private static final int SEND_FREQUENCY_MILLIS = 5 ;
51+
52+ @ RegisterExtension
53+ static WireMockExtension wireMock = WireMockExtension .newInstance ()
54+ .options (wireMockConfig ().dynamicPort ())
55+ .configureStaticDsl (true )
56+ .build ();
57+
58+ private SqsAsyncClient client ;
59+ private SqsAsyncBatchManager batchManager ;
60+
61+ @ BeforeEach
62+ void setUp () {
63+ client = SqsAsyncClient .builder ()
64+ .endpointOverride (URI .create ("http://localhost:" + wireMock .getPort ()))
65+ .checksumValidationEnabled (false )
66+ .credentialsProvider (StaticCredentialsProvider .create (
67+ AwsBasicCredentials .create ("key" , "secret" )))
68+ .build ();
69+
70+ batchManager = SqsAsyncBatchManager .builder ()
71+ .client (client )
72+ .scheduledExecutor (Executors .newScheduledThreadPool (10 ))
73+ .overrideConfiguration (config -> config
74+ .sendRequestFrequency (Duration .ofMillis (SEND_FREQUENCY_MILLIS ))
75+ .maxBatchSize (MAX_BATCH_SIZE ))
76+ .build ();
77+ }
78+
79+ @ AfterEach
80+ void tearDown () {
81+ batchManager .close ();
82+ client .close ();
83+ }
84+
85+ /**
86+ * Test runs heavy load and expects average batch sizes to be close to max.
87+ */
88+ @ Test
89+ void shouldEfficientlyBatchMessagesUnderHighLoad () throws Exception {
90+ int expectedBatchSize = 25 ; // more than double the actual max of 10
91+ int rateLimit = 1000 / SEND_FREQUENCY_MILLIS * expectedBatchSize ;
92+ int messageCount = rateLimit * 2 ; // run it for 2 seconds
93+ runThroughputTest (messageCount , rateLimit );
94+
95+ // Then: Verify messages were efficiently batched
96+ List <LoggedRequest > batchRequests = findAll (postRequestedFor (anyUrl ()));
97+
98+ // Calculate batching metrics
99+ List <Integer > batchSizes = batchRequests .stream ()
100+ .map (req -> req .getBodyAsString ().split ("\" Id\" " ).length - 1 )
101+ .collect (Collectors .toList ());
102+
103+ double avgBatchSize = batchSizes .stream ()
104+ .mapToInt (Integer ::intValue )
105+ .average ()
106+ .orElse (0 );
107+
108+ double fullBatchRatio = batchSizes .stream ()
109+ .filter (size -> size >= 9 )
110+ .count () / (double ) batchSizes .size ();
111+
112+ // Assert efficient batching
113+ assertThat (avgBatchSize )
114+ .as ("Average batch size" )
115+ .isGreaterThan (8.0 );
116+
117+
118+ assertThat (fullBatchRatio )
119+ .as ("Ratio of nearly full batches (9-10 messages)" )
120+ .isGreaterThan (0.8 );
121+
122+ assertThat ((double )batchRequests .size ())
123+ .as ("Total batch requests for %d messages" , messageCount )
124+ .isLessThan (messageCount / 5d );
125+ }
126+
127+ /**
128+ * Test runs a load that should cause an average batch size of 5.
129+ */
130+ @ Test
131+ void shouldMakeHalfBatches () throws Exception {
132+ int expectedBatchSize = 5 ;
133+ int rateLimit = 1000 / SEND_FREQUENCY_MILLIS * expectedBatchSize ;
134+ int messageCount = rateLimit * 2 ; // run it for 2 seconds
135+ runThroughputTest (messageCount , rateLimit );
136+
137+ // Then: Verify batches were roughly half max size
138+ List <LoggedRequest > batchRequests = findAll (postRequestedFor (anyUrl ()));
139+
140+ // Calculate batching metrics
141+ List <Integer > batchSizes = batchRequests .stream ()
142+ .map (req -> req .getBodyAsString ().split ("\" Id\" " ).length - 1 )
143+ .collect (Collectors .toList ());
144+
145+ double avgBatchSize = batchSizes .stream ()
146+ .mapToInt (Integer ::intValue )
147+ .average ()
148+ .orElse (0 );
149+
150+ // Assert batch expected range
151+ assertThat (avgBatchSize )
152+ .as ("Average batch size" )
153+ .isLessThan (7.0 )
154+ .isGreaterThan (3.0 );
155+
156+ assertThat ((double )batchRequests .size ())
157+ .as ("Total batch requests for %d messages" , messageCount )
158+ .isLessThan (messageCount / 3d );
159+ }
160+
161+ @ Test
162+ void shouldMakeSmallBatches () throws Exception {
163+ int expectedBatchSize = 1 ;
164+ int rateLimit = 1000 / SEND_FREQUENCY_MILLIS * expectedBatchSize ;
165+ int messageCount = rateLimit * 2 ; // run it for 2 seconds
166+ runThroughputTest (messageCount , rateLimit );
167+
168+ // Then: Verify batches were roughly half max size
169+ List <LoggedRequest > batchRequests = findAll (postRequestedFor (anyUrl ()));
170+
171+ // Calculate batching metrics
172+ List <Integer > batchSizes = batchRequests .stream ()
173+ .map (req -> req .getBodyAsString ().split ("\" Id\" " ).length - 1 )
174+ .collect (Collectors .toList ());
175+
176+ double avgBatchSize = batchSizes .stream ()
177+ .mapToInt (Integer ::intValue )
178+ .average ()
179+ .orElse (0 );
180+
181+ // Assert batch expected range
182+ assertThat (avgBatchSize )
183+ .as ("Average batch size" )
184+ .isLessThan (2.0 );
185+
186+ assertThat ((double )batchRequests .size ())
187+ .as ("Total batch requests for %d messages" , messageCount )
188+ .isGreaterThan (messageCount * .5 );
189+ }
190+
191+ private void runThroughputTest (int messageCount , int rateLimit ) throws InterruptedException {
192+ // Given: SQS returns success for batch requests
193+ stubFor (post (anyUrl ())
194+ .willReturn (aResponse ()
195+ .withStatus (200 )
196+ .withBody ("{\" Successful\" : []}" )));
197+
198+ // When: Send rateLimit messages per second concurrently (using 50 threads)
199+ ExecutorService executor = Executors .newFixedThreadPool (CONCURRENT_THREADS );
200+
201+ // Rate limit to spread it out over a couple seconds; enough time to make
202+ // any orphaned scheduled flushes obvious.
203+ RateLimiter rateLimiter = RateLimiter .create (rateLimit );
204+
205+ for (int i = 0 ; i < messageCount ; i ++) {
206+ String messageBody = String .valueOf (i );
207+ rateLimiter .acquire ();
208+ executor .execute (() -> {
209+ try {
210+ batchManager .sendMessage (builder ->
211+ builder .queueUrl (QUEUE_URL )
212+ .messageBody (messageBody ));
213+ } catch (Exception ignored ) {
214+ // Test will fail on assertions if messages aren't sent
215+ }
216+ });
217+ }
218+
219+ executor .shutdown ();
220+ executor .awaitTermination (10 , TimeUnit .SECONDS );
221+ }
222+ }
0 commit comments