Skip to content

Commit 25db334

Browse files
kkondakakkondaka
committed
CloudWatchLogs Sink: Update max event size and drop error events if DLQ is not configured
Signed-off-by: Kondaka <krishkdk@amazon.com>
1 parent 06503f6 commit 25db334

10 files changed

Lines changed: 157 additions & 30 deletions

File tree

‎data-prepper-plugins/cloudwatch-logs/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/CloudWatchLogsIT.java‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,8 @@ public class CloudWatchLogsIT {
104104
@Mock
105105
private Counter eventsFailedCounter;
106106
@Mock
107+
private Counter largeEventsDroppedCounter;
108+
@Mock
107109
private Counter requestsFailedCounter;
108110
@Mock
109111
private Counter dlqSuccessCounter;
@@ -119,6 +121,7 @@ public class CloudWatchLogsIT {
119121
private AtomicInteger eventsFailedCount;
120122
private AtomicInteger requestsFailedCount;
121123
private AtomicInteger dlqSuccessCount;
124+
private AtomicInteger largeEventsDroppedCount;
122125
private CloudWatchLogsClient cloudWatchLogsClient;
123126
private ObjectMapper objectMapper;
124127
private AwsCredentialsProvider awsCredentialsProvider;
@@ -130,6 +133,7 @@ void setUp() {
130133
eventsSuccessCount = new AtomicInteger(0);
131134
requestsSuccessCount = new AtomicInteger(0);
132135
eventsFailedCount = new AtomicInteger(0);
136+
largeEventsDroppedCount = new AtomicInteger(0);
133137
requestsFailedCount = new AtomicInteger(0);
134138
dlqSuccessCount = new AtomicInteger(0);
135139
objectMapper = new ObjectMapper();
@@ -154,6 +158,7 @@ void setUp() {
154158
eventsSuccessCounter = mock(Counter.class);
155159
requestsSuccessCounter = mock(Counter.class);
156160
eventsFailedCounter = mock(Counter.class);
161+
largeEventsDroppedCounter = mock(Counter.class);
157162
requestsFailedCounter = mock(Counter.class);
158163
dlqSuccessCounter = mock(Counter.class);
159164
lenient().doAnswer((a)-> {
@@ -166,6 +171,11 @@ void setUp() {
166171
eventsFailedCount.addAndGet(v);
167172
return null;
168173
}).when(eventsFailedCounter).increment(any(Double.class));
174+
lenient().doAnswer((a)-> {
175+
int v = (int)(double)(a.getArgument(0));
176+
largeEventsDroppedCount.addAndGet(v);
177+
return null;
178+
}).when(largeEventsDroppedCounter).increment(any(Double.class));
169179
lenient().doAnswer((a)-> {
170180
requestsSuccessCount.addAndGet(1);
171181
return null;
@@ -199,6 +209,9 @@ void setUp() {
199209
if (s.equals(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_EVENTS_FAILED)) {
200210
return eventsFailedCounter;
201211
}
212+
if (s.equals(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_LARGE_EVENTS_DROPPED)) {
213+
return largeEventsDroppedCounter;
214+
}
202215
if (s.contains("NumDlqSuccess")) {
203216
return dlqSuccessCounter;
204217
}
@@ -437,6 +450,53 @@ void testWithLargeSingleMessagesSentToDLQ() {
437450

438451
}
439452

453+
@Test
454+
void testWithLargeSingleMessagesWhenDLQNotConfigured() {
455+
s3Client = S3Client.builder()
456+
.credentialsProvider(awsCredentialsProvider)
457+
.region(Region.of(awsRegion))
458+
.build();
459+
460+
long startTime = Instant.now().toEpochMilli();
461+
when(thresholdConfig.getBatchSize()).thenReturn(NUM_RECORDS);
462+
when(thresholdConfig.getMaxEventSizeBytes()).thenReturn(200L);
463+
when(thresholdConfig.getMaxRequestSizeBytes()).thenReturn(1000L);
464+
when(cloudWatchLogsSinkConfig.getDlq()).thenReturn(null);
465+
466+
sink = createObjectUnderTest();
467+
Collection<Record<Event>> records = getRecordList(NUM_RECORDS);
468+
Record<Event> largeRecord = getLargeRecord(200);
469+
records.add(largeRecord);
470+
471+
sink.doOutput(records);
472+
await().atMost(Duration.ofSeconds(30))
473+
.untilAsserted(() -> {
474+
long endTime = Instant.now().toEpochMilli();
475+
GetLogEventsRequest getRequest = GetLogEventsRequest
476+
.builder()
477+
.logGroupName(logGroupName)
478+
.logStreamName(logStreamName)
479+
.startTime(startTime)
480+
.endTime(endTime)
481+
.build();
482+
GetLogEventsResponse response = cloudWatchLogsClient.getLogEvents(getRequest);
483+
List<OutputLogEvent> events = response.events();
484+
assertThat(events.size(), equalTo(NUM_RECORDS));
485+
for (int i = 0; i < events.size(); i++) {
486+
String message = events.get(i).message();
487+
Map<String, Object> event = objectMapper.readValue(message, Map.class);
488+
assertThat(event.get("name"), equalTo("Person"+i));
489+
assertThat(event.get("age"), equalTo(Integer.toString(i)));
490+
}
491+
});
492+
assertThat(eventsSuccessCount.get(), equalTo(NUM_RECORDS));
493+
assertThat(requestsSuccessCount.get(), equalTo(1));
494+
assertThat(largeEventsDroppedCount.get(), equalTo(1));
495+
assertThat(dlqSuccessCount.get(), equalTo(0));
496+
verify(eventHandle, times(NUM_RECORDS+1)).release(true);
497+
498+
}
499+
440500
@Test
441501
void testWithBadCredentials_AllEventsToDLQ() {
442502
s3Client = S3Client.builder()

‎data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/CloudWatchLogsSink.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,7 @@ public CloudWatchLogsSink(final PluginSetting pluginSetting,
9191
.cloudWatchLogsClient(cloudWatchLogsClient)
9292
.cloudWatchLogsMetrics(cloudWatchLogsMetrics)
9393
.dlqPushHandler(dlqPushHandler)
94+
.dropIfDlqNotConfigured(true)
9495
.logGroup(cloudWatchLogsSinkConfig.getLogGroup())
9596
.logStream(cloudWatchLogsSinkConfig.getLogStream())
9697
.retryCount(cloudWatchLogsSinkConfig.getMaxRetries())
@@ -104,7 +105,7 @@ public CloudWatchLogsSink(final PluginSetting pluginSetting,
104105
throw new InvalidBufferTypeException("Error loading buffer!");
105106
}
106107

107-
cloudWatchLogsService = new CloudWatchLogsService(buffer, cloudWatchLogsLimits, cloudWatchLogsDispatcher, dlqPushHandler);
108+
cloudWatchLogsService = new CloudWatchLogsService(buffer, cloudWatchLogsMetrics, cloudWatchLogsLimits, cloudWatchLogsDispatcher, dlqPushHandler, true);
108109
}
109110

110111
@Override

‎data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/client/CloudWatchLogsDispatcher.java‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -35,13 +35,15 @@ public class CloudWatchLogsDispatcher {
3535
private CloudWatchLogsClient cloudWatchLogsClient;
3636
private CloudWatchLogsMetrics cloudWatchLogsMetrics;
3737
private DlqPushHandler dlqPushHandler;
38+
private boolean dropIfDlqNotConfigured;
3839
private Executor executor;
3940
private String logGroup;
4041
private String logStream;
4142
private int retryCount;
4243
public CloudWatchLogsDispatcher(final CloudWatchLogsClient cloudWatchLogsClient,
4344
final CloudWatchLogsMetrics cloudWatchLogsMetrics,
4445
final DlqPushHandler dlqPushHandler,
46+
final boolean dropIfDlqNotConfigured,
4547
final Executor executor,
4648
final String logGroup,
4749
final String logStream,
@@ -52,6 +54,7 @@ public CloudWatchLogsDispatcher(final CloudWatchLogsClient cloudWatchLogsClient,
5254
this.logStream = logStream;
5355
this.retryCount = retryCount;
5456
this.dlqPushHandler = dlqPushHandler;
57+
this.dropIfDlqNotConfigured = dropIfDlqNotConfigured;
5558

5659
this.executor = executor;
5760
}
@@ -95,6 +98,7 @@ public void dispatchLogs(List<InputLogEvent> inputLogEvents, List<EventHandle> e
9598
.dlqPushHandler(dlqPushHandler)
9699
.putLogEventsRequest(putLogEventsRequest)
97100
.eventHandles(eventHandles)
101+
.dropIfDlqNotConfigured(dropIfDlqNotConfigured)
98102
.totalEventCount(inputLogEvents.size())
99103
.retryCount(retryCount)
100104
.build());
@@ -111,6 +115,7 @@ protected static class Uploader implements Runnable {
111115
private final List<EventHandle> eventHandles;
112116
private final int totalEventCount;
113117
private final int retryCount;
118+
private boolean dropIfDlqNotConfigured;
114119

115120
@Override
116121
public void run() {
@@ -154,7 +159,7 @@ public void upload() {
154159
cloudWatchLogsMetrics.increaseLogEventFailCounter(totalEventCount);
155160
List<InputLogEvent> logEvents = putLogEventsRequest.logEvents();
156161
for (int i = 0; i < logEvents.size(); i++) {
157-
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, eventHandles.get(i), logEvents.get(i).message(), failureMessage, dlqPushHandler);
162+
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, eventHandles.get(i), logEvents.get(i).message(), failureMessage, dlqPushHandler, dropIfDlqNotConfigured);
158163
if (dlqObject != null) {
159164
dlqObjects.add(dlqObject);
160165
}
@@ -179,7 +184,7 @@ List<DlqObject> getDlqObjectsFromResponse(PutLogEventsResponse putLogEventsRespo
179184
if (endIndex != null) {
180185
int i = 0;
181186
for (InputLogEvent logEvent : logEvents.subList(0, endIndex)) {
182-
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, eventHandles.get(i), logEvent.message(), "Too old log event", dlqPushHandler);
187+
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, eventHandles.get(i), logEvent.message(), "Too old log event", dlqPushHandler, dropIfDlqNotConfigured);
183188
if (dlqObject != null) {
184189
dlqObjects.add(dlqObject);
185190
}
@@ -190,7 +195,7 @@ List<DlqObject> getDlqObjectsFromResponse(PutLogEventsResponse putLogEventsRespo
190195
if (startIndex != null) {
191196
int i = 0;
192197
for (InputLogEvent logEvent : logEvents.subList(startIndex, logEvents.size())) {
193-
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, eventHandles.get(startIndex + i), logEvent.message(), "Too old log event", dlqPushHandler);
198+
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, eventHandles.get(startIndex + i), logEvent.message(), "Too old log event", dlqPushHandler, dropIfDlqNotConfigured);
194199
if (dlqObject != null) {
195200
dlqObjects.add(dlqObject);
196201
}

‎data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/client/CloudWatchLogsMetrics.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,16 +18,19 @@ public class CloudWatchLogsMetrics {
1818
public static final String CLOUDWATCH_LOGS_EVENTS_SUCCEEDED = "cloudWatchLogsEventsSucceeded";
1919
public static final String CLOUDWATCH_LOGS_EVENTS_FAILED = "cloudWatchLogsEventsFailed";
2020
public static final String CLOUDWATCH_LOGS_REQUESTS_FAILED = "cloudWatchLogsRequestsFailed";
21+
public static final String CLOUDWATCH_LOGS_LARGE_EVENTS_DROPPED = "cloudWatchLogsLargeEventsDropped";
2122
private final Counter logEventSuccessCounter;
2223
private final Counter logEventFailCounter;
2324
private final Counter requestSuccessCount;
2425
private final Counter requestFailCount;
26+
private final Counter logLargeEventsDroppedCounter;
2527

2628
public CloudWatchLogsMetrics(final PluginMetrics pluginMetrics) {
2729
this.logEventSuccessCounter = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_EVENTS_SUCCEEDED);
2830
this.requestFailCount = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_REQUESTS_FAILED);
2931
this.logEventFailCounter = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_EVENTS_FAILED);
3032
this.requestSuccessCount = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_REQUESTS_SUCCEEDED);
33+
this.logLargeEventsDroppedCounter = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_LARGE_EVENTS_DROPPED);
3134
}
3235

3336
public void increaseLogEventSuccessCounter(int value) {
@@ -45,4 +48,8 @@ public void increaseLogEventFailCounter(int value) {
4548
public void increaseRequestFailCounter(int value) {
4649
requestFailCount.increment(value);
4750
}
51+
52+
public void increaseLogLargeEventsDroppedCounter(int value) {
53+
logLargeEventsDroppedCounter.increment(value);
54+
}
4855
}

‎data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/client/CloudWatchLogsService.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,22 +36,28 @@ public class CloudWatchLogsService {
3636
private final CloudWatchLogsDispatcher cloudWatchLogsDispatcher;
3737
private final Buffer buffer;
3838
private final CloudWatchLogsLimits cloudWatchLogsLimits;
39+
private CloudWatchLogsMetrics cloudWatchLogsMetrics;
3940
private final SinkStopWatch sinkStopWatch;
4041
private final ReentrantLock processLock;
4142
private final DlqPushHandler dlqPushHandler;
43+
private final boolean dropIfDlqNotConfigured;
4244
public CloudWatchLogsService(final Buffer buffer,
45+
final CloudWatchLogsMetrics cloudWatchLogsMetrics,
4346
final CloudWatchLogsLimits cloudWatchLogsLimits,
4447
final CloudWatchLogsDispatcher cloudWatchLogsDispatcher,
45-
final DlqPushHandler dlqPushHandler) {
48+
final DlqPushHandler dlqPushHandler,
49+
final boolean dropIfDlqNotConfigured) {
4650

4751
this.buffer = buffer;
4852
this.cloudWatchLogsLimits = cloudWatchLogsLimits;
4953

54+
this.cloudWatchLogsMetrics = cloudWatchLogsMetrics;
5055
processLock = new ReentrantLock();
5156
sinkStopWatch = new SinkStopWatch();
5257

5358
this.cloudWatchLogsDispatcher = cloudWatchLogsDispatcher;
5459
this.dlqPushHandler = dlqPushHandler;
60+
this.dropIfDlqNotConfigured = dropIfDlqNotConfigured;
5561
}
5662

5763
/**
@@ -79,9 +85,11 @@ public void processLogEvents(final Collection<Record<Event>> logs) {
7985

8086
if (cloudWatchLogsLimits.isGreaterThanMaxEventSize(logLength)) {
8187
final String failureMessage = String.format("Event blocked due to Max Size restriction! Event Size : %s", (logLength + CloudWatchLogsLimits.APPROXIMATE_LOG_EVENT_OVERHEAD_SIZE));
82-
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, log.getData().getEventHandle(), logString, failureMessage, dlqPushHandler);
88+
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, log.getData().getEventHandle(), logString, failureMessage, dlqPushHandler, dropIfDlqNotConfigured);
8389
if (dlqObject != null) {
8490
dlqObjects.add(dlqObject);
91+
} else if (dropIfDlqNotConfigured) {
92+
cloudWatchLogsMetrics.increaseLogLargeEventsDroppedCounter(1);
8593
}
8694
continue;
8795
}

‎data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/config/ThresholdConfig.java‎

Lines changed: 18 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
import com.fasterxml.jackson.annotation.JsonProperty;
99
import jakarta.validation.constraints.Min;
1010
import jakarta.validation.constraints.Max;
11-
import jakarta.validation.constraints.Size;
11+
import jakarta.validation.constraints.AssertTrue;
1212
import org.hibernate.validator.constraints.time.DurationMax;
1313
import org.hibernate.validator.constraints.time.DurationMin;
1414
import org.opensearch.dataprepper.model.types.ByteCount;
@@ -21,22 +21,22 @@
2121
* restrictions.
2222
*/
2323
public class ThresholdConfig {
24+
public static final long ONE_MB = 1048576L;
2425
public static final int DEFAULT_BATCH_SIZE = 25;
25-
public static final String DEFAULT_EVENT_SIZE = "256kb";
26-
public static final String DEFAULT_SIZE_OF_REQUEST = "1mb";
26+
public static final String DEFAULT_MAX_EVENT_SIZE = "1mb";
27+
public static final String DEFAULT_MAX_REQUEST_SIZE = "1mb";
2728
public static final long DEFAULT_FLUSH_INTERVAL = 60;
2829

2930
@JsonProperty(value = "batch_size", defaultValue="25")
3031
@Min(1)
3132
@Max(10000)
3233
private int batchSize = DEFAULT_BATCH_SIZE;
3334

34-
@JsonProperty(value = "max_event_size", defaultValue="256k")
35-
@Size(min = 1, max = 256, message = "max_event_size amount should be between 1 to 256 kilobytes")
36-
private String maxEventSize = DEFAULT_EVENT_SIZE;
35+
@JsonProperty(value = "max_event_size", defaultValue="1mb")
36+
private String maxEventSize = DEFAULT_MAX_EVENT_SIZE;
3737

3838
@JsonProperty(value = "max_request_size", defaultValue="1mb")
39-
private String maxRequestSize = DEFAULT_SIZE_OF_REQUEST;
39+
private String maxRequestSize = DEFAULT_MAX_REQUEST_SIZE;
4040

4141
@JsonProperty("flush_interval")
4242
@DurationMin(seconds = 60)
@@ -59,4 +59,15 @@ public long getFlushInterval() {
5959
return flushInterval.getSeconds();
6060
}
6161

62+
@AssertTrue(message = "Both the maximum event size and maximum request size must be configured with a value greater than zero (0) and less than 1 megabyte (MB)")
63+
boolean isValidConfig() {
64+
long maxEventSizeBytes = ByteCount.parse(maxEventSize).getBytes();
65+
long maxRequestSizeBytes = ByteCount.parse(maxRequestSize).getBytes();
66+
67+
return maxEventSizeBytes > 0 &&
68+
maxEventSizeBytes < ONE_MB &&
69+
maxRequestSizeBytes > 0 &&
70+
maxRequestSizeBytes < ONE_MB;
71+
}
72+
6273
}

‎data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/utils/CloudWatchLogsSinkUtils.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,12 +16,12 @@
1616

1717
public class CloudWatchLogsSinkUtils {
1818
private static final Logger LOG = LoggerFactory.getLogger(CloudWatchLogsSinkUtils.class);
19-
public static DlqObject createDlqObject(final int status, final EventHandle eventHandle, final String message, final String failureMessage, final DlqPushHandler dlqPushHandler) {
19+
public static DlqObject createDlqObject(final int status, final EventHandle eventHandle, final String message, final String failureMessage, final DlqPushHandler dlqPushHandler, final boolean dropIfDlqNotConfigured) {
2020
if (dlqPushHandler != null) {
2121
CloudWatchLogsSinkDlqData cloudWatchLogsSinkDlqData = CloudWatchLogsSinkDlqData.createDlqData(status, message, failureMessage);
2222
return DlqObject.createDlqObject(dlqPushHandler.getPluginSetting(), List.of(eventHandle), cloudWatchLogsSinkDlqData);
2323
} else {
24-
eventHandle.release(false);
24+
eventHandle.release(dropIfDlqNotConfigured);
2525
}
2626
return null;
2727
}

‎data-prepper-plugins/cloudwatch-logs/src/test/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/client/CloudWatchLogsDispatcherTest.java‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@ CloudWatchLogsDispatcher getCloudWatchLogsDispatcher() {
7676
.logGroup(LOG_GROUP)
7777
.logStream(LOG_STREAM)
7878
.retryCount(RETRY_COUNT)
79+
.dropIfDlqNotConfigured(true)
7980
.build();
8081
}
8182

@@ -122,8 +123,8 @@ void GIVEN_too_old_events_SHOULD_not_release_old_events() {
122123
executeDispatcherRunnable();
123124

124125
// First two events should not be released
125-
verify(eventHandles.get(0), never()).release(true);
126-
verify(eventHandles.get(1), never()).release(true);
126+
verify(eventHandles.get(0)).release(true);
127+
verify(eventHandles.get(1)).release(true);
127128

128129
// Remaining events should be released
129130
for (int i = 2; i < eventHandles.size(); i++) {
@@ -157,7 +158,7 @@ void GIVEN_too_new_events_SHOULD_not_release_new_events() {
157158

158159
// Last three events should not be released
159160
for (int i = eventHandles.size() - 3; i < eventHandles.size(); i++) {
160-
verify(eventHandles.get(i), never()).release(true);
161+
verify(eventHandles.get(i)).release(true);
161162
}
162163
}
163164

@@ -181,17 +182,17 @@ void GIVEN_both_old_and_new_rejected_events_SHOULD_only_release_valid_events() {
181182
executeDispatcherRunnable();
182183

183184
// First two events should not be released (too old)
184-
verify(eventHandles.get(0), never()).release(true);
185-
verify(eventHandles.get(1), never()).release(true);
185+
verify(eventHandles.get(0)).release(true);
186+
verify(eventHandles.get(1)).release(true);
186187

187188
// Middle events should be released
188189
for (int i = 2; i < eventHandles.size() - 2; i++) {
189190
verify(eventHandles.get(i)).release(true);
190191
}
191192

192193
// Last two events should not be released (too new)
193-
verify(eventHandles.get(eventHandles.size() - 2), never()).release(true);
194-
verify(eventHandles.get(eventHandles.size() - 1), never()).release(true);
194+
verify(eventHandles.get(eventHandles.size() - 2)).release(true);
195+
verify(eventHandles.get(eventHandles.size() - 1)).release(true);
195196
}
196197

197198
@Test
@@ -247,6 +248,6 @@ void GIVEN_max_retries_exceeded_SHOULD_not_release_events() {
247248
verify(mockCloudWatchLogsMetrics, never()).increaseRequestSuccessCounter(1);
248249

249250
// No events should be released after max retries
250-
eventHandles.forEach(eventHandle -> verify(eventHandle, never()).release(true));
251+
eventHandles.forEach(eventHandle -> verify(eventHandle).release(true));
251252
}
252253
}

0 commit comments

Comments
 (0)