Skip to content

Commit d9e809c

Browse files
jeromevdljeromevdl
committed
test parallel processing
1 parent 7dc7c34 commit d9e809c

File tree

6 files changed

+1208
-17
lines changed

6 files changed

+1208
-17
lines changed

‎powertools-batch/pom.xml

+16
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,16 @@
2121
<skip>true</skip>
2222
</configuration>
2323
</plugin>
24+
<plugin>
25+
<groupId>org.apache.maven.plugins</groupId>
26+
<artifactId>maven-surefire-plugin</artifactId>
27+
<configuration>
28+
<!-- enforce multiple threads for parallel processing tests -->
29+
<argLine>
30+
-Djava.util.concurrent.ForkJoinPool.common.parallelism=4
31+
</argLine>
32+
</configuration>
33+
</plugin>
2434
</plugins>
2535
</build>
2636

@@ -47,6 +57,12 @@
4757
<artifactId>junit-jupiter-api</artifactId>
4858
<scope>test</scope>
4959
</dependency>
60+
<dependency>
61+
<groupId>org.slf4j</groupId>
62+
<artifactId>slf4j-simple</artifactId>
63+
<version>2.0.7</version>
64+
<scope>test</scope>
65+
</dependency>
5066
<dependency>
5167
<groupId>org.assertj</groupId>
5268
<artifactId>assertj-core</artifactId>

‎powertools-batch/src/test/java/software/amazon/lambda/powertools/batch/DdbBatchProcessorTest.java

+80-8
Original file line numberDiff line numberDiff line change
@@ -20,16 +20,27 @@
2020
import com.amazonaws.services.lambda.runtime.events.DynamodbEvent;
2121
import com.amazonaws.services.lambda.runtime.events.StreamsEventResponse;
2222
import com.amazonaws.services.lambda.runtime.tests.annotations.Event;
23+
import java.util.ArrayList;
24+
import java.util.Collections;
25+
import java.util.List;
2326
import java.util.concurrent.atomic.AtomicBoolean;
27+
import org.junit.jupiter.api.AfterEach;
2428
import org.junit.jupiter.params.ParameterizedTest;
2529
import org.mockito.Mock;
2630
import software.amazon.lambda.powertools.batch.handler.BatchMessageHandler;
2731

28-
public class DdbBatchProcessorTest {
32+
class DdbBatchProcessorTest {
2933

3034
@Mock
3135
private Context context;
3236

37+
private final List<String> threadList = Collections.synchronizedList(new ArrayList<>());
38+
39+
@AfterEach
40+
public void clear() {
41+
threadList.clear();
42+
}
43+
3344
private void processMessageSucceeds(DynamodbEvent.DynamodbStreamRecord record, Context context) {
3445
// Great success
3546
}
@@ -40,9 +51,36 @@ private void processMessageFailsForFixedMessage(DynamodbEvent.DynamodbStreamReco
4051
}
4152
}
4253

54+
private void processMessageInParallelSucceeds(DynamodbEvent.DynamodbStreamRecord record, Context context) {
55+
String thread = Thread.currentThread().getName();
56+
if (!threadList.contains(thread)) {
57+
threadList.add(thread);
58+
}
59+
try {
60+
Thread.sleep(500); // simulate some processing
61+
} catch (InterruptedException e) {
62+
throw new RuntimeException(e);
63+
}
64+
}
65+
66+
private void processMessageInParallelFailsForFixedMessage(DynamodbEvent.DynamodbStreamRecord record, Context context) {
67+
String thread = Thread.currentThread().getName();
68+
if (!threadList.contains(thread)) {
69+
threadList.add(thread);
70+
}
71+
try {
72+
Thread.sleep(500); // simulate some processing
73+
} catch (InterruptedException e) {
74+
throw new RuntimeException(e);
75+
}
76+
if (record.getDynamodb().getSequenceNumber().equals("4421584500000000017450439091")) {
77+
throw new RuntimeException("fake exception");
78+
}
79+
}
80+
4381
@ParameterizedTest
4482
@Event(value = "dynamo_event.json", type = DynamodbEvent.class)
45-
public void batchProcessingSucceedsAndReturns(DynamodbEvent event) {
83+
void batchProcessingSucceedsAndReturns(DynamodbEvent event) {
4684
// Arrange
4785
BatchMessageHandler<DynamodbEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
4886
.withDynamoDbBatchHandler()
@@ -52,12 +90,28 @@ public void batchProcessingSucceedsAndReturns(DynamodbEvent event) {
5290
StreamsEventResponse dynamodbBatchResponse = handler.processBatch(event, context);
5391

5492
// Assert
55-
assertThat(dynamodbBatchResponse.getBatchItemFailures()).hasSize(0);
93+
assertThat(dynamodbBatchResponse.getBatchItemFailures()).isEmpty();
94+
}
95+
96+
@ParameterizedTest
97+
@Event(value = "dynamo_event_big.json", type = DynamodbEvent.class)
98+
void parallelBatchProcessingSucceedsAndReturns(DynamodbEvent event) {
99+
// Arrange
100+
BatchMessageHandler<DynamodbEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
101+
.withDynamoDbBatchHandler()
102+
.buildWithRawMessageHandler(this::processMessageInParallelSucceeds);
103+
104+
// Act
105+
StreamsEventResponse dynamodbBatchResponse = handler.processBatchInParallel(event, context);
106+
107+
// Assert
108+
assertThat(dynamodbBatchResponse.getBatchItemFailures()).isEmpty();
109+
assertThat(threadList).hasSizeGreaterThan(1);
56110
}
57111

58112
@ParameterizedTest
59113
@Event(value = "dynamo_event.json", type = DynamodbEvent.class)
60-
public void shouldAddMessageToBatchFailure_whenException_withMessage(DynamodbEvent event) {
114+
void shouldAddMessageToBatchFailure_whenException_withMessage(DynamodbEvent event) {
61115
// Arrange
62116
BatchMessageHandler<DynamodbEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
63117
.withDynamoDbBatchHandler()
@@ -72,9 +126,27 @@ public void shouldAddMessageToBatchFailure_whenException_withMessage(DynamodbEve
72126
assertThat(batchItemFailure.getItemIdentifier()).isEqualTo("4421584500000000017450439091");
73127
}
74128

129+
@ParameterizedTest
130+
@Event(value = "dynamo_event_big.json", type = DynamodbEvent.class)
131+
void parallelBatchProcessing_shouldAddMessageToBatchFailure_whenException_withMessage(DynamodbEvent event) {
132+
// Arrange
133+
BatchMessageHandler<DynamodbEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
134+
.withDynamoDbBatchHandler()
135+
.buildWithRawMessageHandler(this::processMessageInParallelFailsForFixedMessage);
136+
137+
// Act
138+
StreamsEventResponse dynamodbBatchResponse = handler.processBatchInParallel(event, context);
139+
140+
// Assert
141+
assertThat(dynamodbBatchResponse.getBatchItemFailures()).hasSize(1);
142+
StreamsEventResponse.BatchItemFailure batchItemFailure = dynamodbBatchResponse.getBatchItemFailures().get(0);
143+
assertThat(batchItemFailure.getItemIdentifier()).isEqualTo("4421584500000000017450439091");
144+
assertThat(threadList).hasSizeGreaterThan(1);
145+
}
146+
75147
@ParameterizedTest
76148
@Event(value = "dynamo_event.json", type = DynamodbEvent.class)
77-
public void failingFailureHandlerShouldntFailBatch(DynamodbEvent event) {
149+
void failingFailureHandlerShouldntFailBatch(DynamodbEvent event) {
78150
// Arrange
79151
AtomicBoolean wasCalledAndFailed = new AtomicBoolean(false);
80152
BatchMessageHandler<DynamodbEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
@@ -92,15 +164,15 @@ public void failingFailureHandlerShouldntFailBatch(DynamodbEvent event) {
92164

93165
// Assert
94166
assertThat(dynamodbBatchResponse).isNotNull();
95-
assertThat(dynamodbBatchResponse.getBatchItemFailures().size()).isEqualTo(1);
167+
assertThat(dynamodbBatchResponse.getBatchItemFailures()).hasSize(1);
96168
assertThat(wasCalledAndFailed.get()).isTrue();
97169
StreamsEventResponse.BatchItemFailure batchItemFailure = dynamodbBatchResponse.getBatchItemFailures().get(0);
98170
assertThat(batchItemFailure.getItemIdentifier()).isEqualTo("4421584500000000017450439091");
99171
}
100172

101173
@ParameterizedTest
102174
@Event(value = "dynamo_event.json", type = DynamodbEvent.class)
103-
public void failingSuccessHandlerShouldntFailBatchButShouldFailMessage(DynamodbEvent event) {
175+
void failingSuccessHandlerShouldntFailBatchButShouldFailMessage(DynamodbEvent event) {
104176
// Arrange
105177
AtomicBoolean wasCalledAndFailed = new AtomicBoolean(false);
106178
BatchMessageHandler<DynamodbEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
@@ -118,7 +190,7 @@ public void failingSuccessHandlerShouldntFailBatchButShouldFailMessage(DynamodbE
118190

119191
// Assert
120192
assertThat(dynamodbBatchResponse).isNotNull();
121-
assertThat(dynamodbBatchResponse.getBatchItemFailures().size()).isEqualTo(1);
193+
assertThat(dynamodbBatchResponse.getBatchItemFailures()).hasSize(1);
122194
assertThat(wasCalledAndFailed.get()).isTrue();
123195
StreamsEventResponse.BatchItemFailure batchItemFailure = dynamodbBatchResponse.getBatchItemFailures().get(0);
124196
assertThat(batchItemFailure.getItemIdentifier()).isEqualTo("4421584500000000017450439091");

‎powertools-batch/src/test/java/software/amazon/lambda/powertools/batch/KinesisBatchProcessorTest.java

+83-9
Original file line numberDiff line numberDiff line change
@@ -20,17 +20,28 @@
2020
import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
2121
import com.amazonaws.services.lambda.runtime.events.StreamsEventResponse;
2222
import com.amazonaws.services.lambda.runtime.tests.annotations.Event;
23+
import java.util.ArrayList;
24+
import java.util.Collections;
25+
import java.util.List;
2326
import java.util.concurrent.atomic.AtomicBoolean;
27+
import org.junit.jupiter.api.AfterEach;
2428
import org.junit.jupiter.params.ParameterizedTest;
2529
import org.mockito.Mock;
2630
import software.amazon.lambda.powertools.batch.handler.BatchMessageHandler;
2731
import software.amazon.lambda.powertools.batch.model.Product;
2832

29-
public class KinesisBatchProcessorTest {
33+
class KinesisBatchProcessorTest {
3034

3135
@Mock
3236
private Context context;
3337

38+
private final List<String> threadList = Collections.synchronizedList(new ArrayList<>());
39+
40+
@AfterEach
41+
public void clear() {
42+
threadList.clear();
43+
}
44+
3445
private void processMessageSucceeds(KinesisEvent.KinesisEventRecord record, Context context) {
3546
// Great success
3647
}
@@ -42,6 +53,34 @@ private void processMessageFailsForFixedMessage(KinesisEvent.KinesisEventRecord
4253
}
4354
}
4455

56+
private void processMessageInParallelSucceeds(KinesisEvent.KinesisEventRecord record, Context context) {
57+
String thread = Thread.currentThread().getName();
58+
if (!threadList.contains(thread)) {
59+
threadList.add(thread);
60+
}
61+
try {
62+
Thread.sleep(500); // simulate some processing
63+
} catch (InterruptedException e) {
64+
throw new RuntimeException(e);
65+
}
66+
}
67+
68+
private void processMessageInParallelFailsForFixedMessage(KinesisEvent.KinesisEventRecord record, Context context) {
69+
String thread = Thread.currentThread().getName();
70+
if (!threadList.contains(thread)) {
71+
threadList.add(thread);
72+
}
73+
try {
74+
Thread.sleep(500); // simulate some processing
75+
} catch (InterruptedException e) {
76+
throw new RuntimeException(e);
77+
}
78+
if (record.getKinesis().getSequenceNumber()
79+
.equals("49545115243490985018280067714973144582180062593244200961")) {
80+
throw new RuntimeException("fake exception");
81+
}
82+
}
83+
4584
// A handler that throws an exception for _one_ of the deserialized products in the same messages
4685
public void processMessageFailsForFixedProduct(Product product, Context context) {
4786
if (product.getId() == 1234) {
@@ -51,7 +90,7 @@ public void processMessageFailsForFixedProduct(Product product, Context context)
5190

5291
@ParameterizedTest
5392
@Event(value = "kinesis_event.json", type = KinesisEvent.class)
54-
public void batchProcessingSucceedsAndReturns(KinesisEvent event) {
93+
void batchProcessingSucceedsAndReturns(KinesisEvent event) {
5594
// Arrange
5695
BatchMessageHandler<KinesisEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
5796
.withKinesisBatchHandler()
@@ -61,12 +100,28 @@ public void batchProcessingSucceedsAndReturns(KinesisEvent event) {
61100
StreamsEventResponse kinesisBatchResponse = handler.processBatch(event, context);
62101

63102
// Assert
64-
assertThat(kinesisBatchResponse.getBatchItemFailures()).hasSize(0);
103+
assertThat(kinesisBatchResponse.getBatchItemFailures()).isEmpty();
104+
}
105+
106+
@ParameterizedTest
107+
@Event(value = "kinesis_event_big.json", type = KinesisEvent.class)
108+
void batchProcessingInParallelSucceedsAndReturns(KinesisEvent event) {
109+
// Arrange
110+
BatchMessageHandler<KinesisEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
111+
.withKinesisBatchHandler()
112+
.buildWithRawMessageHandler(this::processMessageInParallelSucceeds);
113+
114+
// Act
115+
StreamsEventResponse kinesisBatchResponse = handler.processBatchInParallel(event, context);
116+
117+
// Assert
118+
assertThat(kinesisBatchResponse.getBatchItemFailures()).isEmpty();
119+
assertThat(threadList).hasSizeGreaterThan(1);
65120
}
66121

67122
@ParameterizedTest
68123
@Event(value = "kinesis_event.json", type = KinesisEvent.class)
69-
public void shouldAddMessageToBatchFailure_whenException_withMessage(KinesisEvent event) {
124+
void shouldAddMessageToBatchFailure_whenException_withMessage(KinesisEvent event) {
70125
// Arrange
71126
BatchMessageHandler<KinesisEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
72127
.withKinesisBatchHandler()
@@ -82,9 +137,28 @@ public void shouldAddMessageToBatchFailure_whenException_withMessage(KinesisEven
82137
"49545115243490985018280067714973144582180062593244200961");
83138
}
84139

140+
@ParameterizedTest
141+
@Event(value = "kinesis_event_big.json", type = KinesisEvent.class)
142+
void batchProcessingInParallel_shouldAddMessageToBatchFailure_whenException_withMessage(KinesisEvent event) {
143+
// Arrange
144+
BatchMessageHandler<KinesisEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
145+
.withKinesisBatchHandler()
146+
.buildWithRawMessageHandler(this::processMessageInParallelFailsForFixedMessage);
147+
148+
// Act
149+
StreamsEventResponse kinesisBatchResponse = handler.processBatchInParallel(event, context);
150+
151+
// Assert
152+
assertThat(kinesisBatchResponse.getBatchItemFailures()).hasSize(1);
153+
StreamsEventResponse.BatchItemFailure batchItemFailure = kinesisBatchResponse.getBatchItemFailures().get(0);
154+
assertThat(batchItemFailure.getItemIdentifier()).isEqualTo(
155+
"49545115243490985018280067714973144582180062593244200961");
156+
assertThat(threadList).hasSizeGreaterThan(1);
157+
}
158+
85159
@ParameterizedTest
86160
@Event(value = "kinesis_event.json", type = KinesisEvent.class)
87-
public void shouldAddMessageToBatchFailure_whenException_withProduct(KinesisEvent event) {
161+
void shouldAddMessageToBatchFailure_whenException_withProduct(KinesisEvent event) {
88162
// Arrange
89163
BatchMessageHandler<KinesisEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
90164
.withKinesisBatchHandler()
@@ -102,7 +176,7 @@ public void shouldAddMessageToBatchFailure_whenException_withProduct(KinesisEven
102176

103177
@ParameterizedTest
104178
@Event(value = "kinesis_event.json", type = KinesisEvent.class)
105-
public void failingFailureHandlerShouldntFailBatch(KinesisEvent event) {
179+
void failingFailureHandlerShouldntFailBatch(KinesisEvent event) {
106180
// Arrange
107181
AtomicBoolean wasCalled = new AtomicBoolean(false);
108182
BatchMessageHandler<KinesisEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
@@ -118,7 +192,7 @@ public void failingFailureHandlerShouldntFailBatch(KinesisEvent event) {
118192

119193
// Assert
120194
assertThat(kinesisBatchResponse).isNotNull();
121-
assertThat(kinesisBatchResponse.getBatchItemFailures().size()).isEqualTo(1);
195+
assertThat(kinesisBatchResponse.getBatchItemFailures()).hasSize(1);
122196
assertThat(wasCalled.get()).isTrue();
123197
StreamsEventResponse.BatchItemFailure batchItemFailure = kinesisBatchResponse.getBatchItemFailures().get(0);
124198
assertThat(batchItemFailure.getItemIdentifier()).isEqualTo(
@@ -127,7 +201,7 @@ public void failingFailureHandlerShouldntFailBatch(KinesisEvent event) {
127201

128202
@ParameterizedTest
129203
@Event(value = "kinesis_event.json", type = KinesisEvent.class)
130-
public void failingSuccessHandlerShouldntFailBatchButShouldFailMessage(KinesisEvent event) {
204+
void failingSuccessHandlerShouldntFailBatchButShouldFailMessage(KinesisEvent event) {
131205
// Arrange
132206
AtomicBoolean wasCalledAndFailed = new AtomicBoolean(false);
133207
BatchMessageHandler<KinesisEvent, StreamsEventResponse> handler = new BatchMessageHandlerBuilder()
@@ -146,7 +220,7 @@ public void failingSuccessHandlerShouldntFailBatchButShouldFailMessage(KinesisEv
146220

147221
// Assert
148222
assertThat(kinesisBatchResponse).isNotNull();
149-
assertThat(kinesisBatchResponse.getBatchItemFailures().size()).isEqualTo(1);
223+
assertThat(kinesisBatchResponse.getBatchItemFailures()).hasSize(1);
150224
assertThat(wasCalledAndFailed.get()).isTrue();
151225
StreamsEventResponse.BatchItemFailure batchItemFailure = kinesisBatchResponse.getBatchItemFailures().get(0);
152226
assertThat(batchItemFailure.getItemIdentifier()).isEqualTo(

0 commit comments

Comments
 (0)