Skip to content

Commit 352e767

Browse files
ChrisKujawaChrisKujawa
committed
style: apply codestyle
(cherry picked from commit 3fbde5f)
1 parent 6181e7d commit 352e767

File tree

2 files changed

+119
-84
lines changed

2 files changed

+119
-84
lines changed

‎engine/src/test/java/io/camunda/zeebe/engine/util/StreamPlatform.java

+3-4
Original file line numberDiff line numberDiff line change
@@ -101,7 +101,8 @@ public StreamPlatform(
101101
actorScheduler.submitActor(writeActor);
102102

103103
defaultMockedRecordProcessor = mock(RecordProcessor.class);
104-
when(defaultMockedRecordProcessor.process(any(), any())).thenReturn(EmptyProcessingResult.INSTANCE);
104+
when(defaultMockedRecordProcessor.process(any(), any()))
105+
.thenReturn(EmptyProcessingResult.INSTANCE);
105106
when(defaultMockedRecordProcessor.onProcessingError(any(), any(), any()))
106107
.thenReturn(EmptyProcessingResult.INSTANCE);
107108
when(defaultMockedRecordProcessor.accepts(any())).thenReturn(true);
@@ -205,8 +206,6 @@ public StreamProcessor buildStreamProcessor(
205206
final var storage = createRuntimeFolder(stream);
206207
final var snapshot = storage.getParent().resolve(SNAPSHOT_FOLDER);
207208

208-
209-
210209
final ZeebeDb<?> zeebeDb;
211210
if (snapshotWasTaken) {
212211
zeebeDb = zeebeDbFactory.createDb(snapshot.toFile());
@@ -232,7 +231,7 @@ public StreamProcessor buildStreamProcessor(
232231
final var openFuture = streamProcessor.openAsync(false);
233232

234233
if (awaitOpening) { // and recovery
235-
verify(mockProcessorLifecycleAware, timeout(15 * 1000)).onRecovered(any());
234+
verify(mockProcessorLifecycleAware, timeout(15 * 1000)).onRecovered(any());
236235
}
237236
openFuture.join(15, TimeUnit.SECONDS);
238237

‎engine/src/test/java/io/camunda/zeebe/streamprocessor/StreamProcessorTest.java

+116-80
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,9 @@ public void shouldCallStreamProcessorLifecycle() throws Exception {
8989
public void shouldCallStreamProcessorLifecycleOnFail() {
9090
// given
9191
final var mockProcessorLifecycleAware = streamPlatform.getMockProcessorLifecycleAware();
92-
doThrow(new RuntimeException("force fail")).when(mockProcessorLifecycleAware).onRecovered(any());
92+
doThrow(new RuntimeException("force fail"))
93+
.when(mockProcessorLifecycleAware)
94+
.onRecovered(any());
9395

9496
// when
9597
streamPlatform.startStreamProcessor();
@@ -140,7 +142,9 @@ public void shouldCallOnErrorWhenProcessingFails() {
140142
inOrder.verify(defaultRecordProcessor, TIMEOUT).init(any());
141143
inOrder.verify(defaultRecordProcessor, TIMEOUT).accepts(ValueType.PROCESS_INSTANCE);
142144
inOrder.verify(defaultRecordProcessor, TIMEOUT).process(any(), any());
143-
inOrder.verify(defaultRecordProcessor, TIMEOUT).onProcessingError(eq(processingError), any(), any());
145+
inOrder
146+
.verify(defaultRecordProcessor, TIMEOUT)
147+
.onProcessingError(eq(processingError), any(), any());
144148
inOrder.verifyNoMoreInteractions();
145149
}
146150

@@ -163,7 +167,9 @@ public void shouldLoopWhenOnErrorFails() {
163167
inOrder.verify(defaultRecordProcessor, TIMEOUT).init(any());
164168
inOrder.verify(defaultRecordProcessor, TIMEOUT).accepts(ValueType.PROCESS_INSTANCE);
165169
inOrder.verify(defaultRecordProcessor, TIMEOUT).process(any(), any());
166-
inOrder.verify(defaultRecordProcessor, TIMEOUT.atLeast(5)).onProcessingError(eq(processingError), any(), any());
170+
inOrder
171+
.verify(defaultRecordProcessor, TIMEOUT.atLeast(5))
172+
.onProcessingError(eq(processingError), any(), any());
167173
}
168174

169175
@Test
@@ -221,14 +227,14 @@ public void shouldWriteFollowUpEventsAndCommands() {
221227
// given
222228
final var defaultRecordProcessor = streamPlatform.getDefaultMockedRecordProcessor();
223229
final var firstResultBuilder = new BufferedProcessingResultBuilder((c, v) -> true);
224-
firstResultBuilder.appendRecordReturnEither(1, RecordType.EVENT,
225-
ACTIVATE_ELEMENT, RejectionType.NULL_VAL, "", RECORD);
226-
firstResultBuilder.appendRecordReturnEither(2, RecordType.COMMAND,
227-
ACTIVATE_ELEMENT, RejectionType.NULL_VAL, "", RECORD);
230+
firstResultBuilder.appendRecordReturnEither(
231+
1, RecordType.EVENT, ACTIVATE_ELEMENT, RejectionType.NULL_VAL, "", RECORD);
232+
firstResultBuilder.appendRecordReturnEither(
233+
2, RecordType.COMMAND, ACTIVATE_ELEMENT, RejectionType.NULL_VAL, "", RECORD);
228234

229235
final var secondResultBuilder = new BufferedProcessingResultBuilder((c, v) -> true);
230-
secondResultBuilder.appendRecordReturnEither(3, RecordType.EVENT,
231-
ACTIVATE_ELEMENT, RejectionType.NULL_VAL, "", RECORD);
236+
secondResultBuilder.appendRecordReturnEither(
237+
3, RecordType.EVENT, ACTIVATE_ELEMENT, RejectionType.NULL_VAL, "", RECORD);
232238

233239
when(defaultRecordProcessor.process(any(), any()))
234240
.thenReturn(firstResultBuilder.build())
@@ -238,17 +244,21 @@ public void shouldWriteFollowUpEventsAndCommands() {
238244
streamPlatform.startStreamProcessor();
239245

240246
// when
241-
streamPlatform.writeBatch(
242-
command().processInstance(ACTIVATE_ELEMENT, RECORD));
247+
streamPlatform.writeBatch(command().processInstance(ACTIVATE_ELEMENT, RECORD));
243248

244249
// then
245250
verify(defaultRecordProcessor, TIMEOUT.times(2)).process(any(), any());
246-
await("Last written position should be updated").untilAsserted(() ->
247-
assertThat(streamPlatform.getLogStream().getLastWrittenPosition()).isEqualTo(4)
248-
);
249-
await("Last processed position should be updated").untilAsserted(() ->
250-
assertThat(streamPlatform.getStreamProcessor().getLastProcessedPositionAsync().join()).isEqualTo(3));
251+
await("Last written position should be updated")
252+
.untilAsserted(
253+
() -> assertThat(streamPlatform.getLogStream().getLastWrittenPosition()).isEqualTo(4));
254+
await("Last processed position should be updated")
255+
.untilAsserted(
256+
() ->
257+
assertThat(
258+
streamPlatform.getStreamProcessor().getLastProcessedPositionAsync().join())
259+
.isEqualTo(3));
251260
}
261+
252262
@Test
253263
public void shouldExecutePostCommitTask() {
254264
// given
@@ -261,8 +271,7 @@ public void shouldExecutePostCommitTask() {
261271
streamPlatform.startStreamProcessor();
262272

263273
// when
264-
streamPlatform.writeBatch(
265-
command().processInstance(ACTIVATE_ELEMENT, RECORD));
274+
streamPlatform.writeBatch(command().processInstance(ACTIVATE_ELEMENT, RECORD));
266275

267276
// then
268277
verify(mockPostCommitTask, TIMEOUT).flush();
@@ -282,8 +291,7 @@ public void shouldRepeatExecutePostCommitTask() {
282291
streamPlatform.startStreamProcessor();
283292

284293
// when
285-
streamPlatform.writeBatch(
286-
command().processInstance(ACTIVATE_ELEMENT, RECORD));
294+
streamPlatform.writeBatch(command().processInstance(ACTIVATE_ELEMENT, RECORD));
287295

288296
// then
289297
verify(mockPostCommitTask, TIMEOUT.atLeast(5)).flush();
@@ -318,18 +326,20 @@ public void shouldNotRepeatPostCommitOnException() throws Exception {
318326
public void shouldUpdateStateOnSuccessfulProcessing() {
319327
// given
320328
final var testProcessor = spy(new TestProcessor());
321-
testProcessor.processingAction = (ctx) -> {
322-
final var zeebeDb = ctx.getZeebeDb();
323-
final var keyGenerator = new DbKeyGenerator(1, zeebeDb, ctx.getTransactionContext());
324-
keyGenerator.nextKey();
325-
keyGenerator.nextKey();
326-
keyGenerator.nextKey();
327-
};
329+
testProcessor.processingAction =
330+
(ctx) -> {
331+
final var zeebeDb = ctx.getZeebeDb();
332+
final var keyGenerator = new DbKeyGenerator(1, zeebeDb, ctx.getTransactionContext());
333+
keyGenerator.nextKey();
334+
keyGenerator.nextKey();
335+
keyGenerator.nextKey();
336+
};
328337
// in order to not mark the processing as skipped we need to return a result
329338
testProcessor.processingResult = new BufferedProcessingResultBuilder((c, s) -> true).build();
330339
doCallRealMethod()
331340
.doReturn(EmptyProcessingResult.INSTANCE)
332-
.when(testProcessor).process(any(), any());
341+
.when(testProcessor)
342+
.process(any(), any());
333343
streamPlatform.withRecordProcessors(List.of(testProcessor)).startStreamProcessor();
334344

335345
final var zeebeDb = testProcessor.recordProcessorContext.getZeebeDb();
@@ -352,20 +362,22 @@ public void shouldUpdateStateOnSuccessfulProcessing() {
352362
public void shouldNotUpdateStateOnExceptionInProcessing() {
353363
// given
354364
final var testProcessor = spy(new TestProcessor());
355-
testProcessor.processingAction = (ctx) -> {
356-
final var zeebeDb = ctx.getZeebeDb();
357-
final var keyGenerator = new DbKeyGenerator(1, zeebeDb, ctx.getTransactionContext());
358-
keyGenerator.nextKey();
359-
keyGenerator.nextKey();
360-
keyGenerator.nextKey();
361-
362-
throw new RuntimeException("expected");
363-
};
365+
testProcessor.processingAction =
366+
(ctx) -> {
367+
final var zeebeDb = ctx.getZeebeDb();
368+
final var keyGenerator = new DbKeyGenerator(1, zeebeDb, ctx.getTransactionContext());
369+
keyGenerator.nextKey();
370+
keyGenerator.nextKey();
371+
keyGenerator.nextKey();
372+
373+
throw new RuntimeException("expected");
374+
};
364375
// in order to not mark the processing as skipped we need to return a result
365376
testProcessor.processingResult = new BufferedProcessingResultBuilder((c, s) -> true).build();
366377
doCallRealMethod()
367378
.doReturn(EmptyProcessingResult.INSTANCE)
368-
.when(testProcessor).process(any(), any());
379+
.when(testProcessor)
380+
.process(any(), any());
369381
streamPlatform.withRecordProcessors(List.of(testProcessor)).startStreamProcessor();
370382

371383
final var zeebeDb = testProcessor.recordProcessorContext.getZeebeDb();
@@ -388,19 +400,22 @@ public void shouldNotUpdateStateOnExceptionInProcessing() {
388400
public void shouldUpdateStateOnProcessingErrorCall() {
389401
// given
390402
final var testProcessor = spy(new TestProcessor());
391-
testProcessor.processingAction = (ctx) -> {
392-
throw new RuntimeException("expected");
393-
};
394-
testProcessor.onProcessingErrorAction = (ctx) -> {
395-
final var zeebeDb = ctx.getZeebeDb();
396-
final var keyGenerator = new DbKeyGenerator(1, zeebeDb, ctx.getTransactionContext());
397-
keyGenerator.nextKey();
398-
keyGenerator.nextKey();
399-
keyGenerator.nextKey();
400-
};
403+
testProcessor.processingAction =
404+
(ctx) -> {
405+
throw new RuntimeException("expected");
406+
};
407+
testProcessor.onProcessingErrorAction =
408+
(ctx) -> {
409+
final var zeebeDb = ctx.getZeebeDb();
410+
final var keyGenerator = new DbKeyGenerator(1, zeebeDb, ctx.getTransactionContext());
411+
keyGenerator.nextKey();
412+
keyGenerator.nextKey();
413+
keyGenerator.nextKey();
414+
};
401415
doCallRealMethod()
402416
.doReturn(EmptyProcessingResult.INSTANCE)
403-
.when(testProcessor).process(any(), any());
417+
.when(testProcessor)
418+
.process(any(), any());
404419
streamPlatform.withRecordProcessors(List.of(testProcessor)).startStreamProcessor();
405420

406421
final var zeebeDb = testProcessor.recordProcessorContext.getZeebeDb();
@@ -423,24 +438,28 @@ public void shouldUpdateStateOnProcessingErrorCall() {
423438
public void shouldNotUpdateStateOnExceptionOnProcessingErrorCall() {
424439
// given
425440
final var testProcessor = spy(new TestProcessor());
426-
testProcessor.processingAction = (ctx) -> {
427-
throw new RuntimeException("expected");
428-
};
429-
testProcessor.onProcessingErrorAction = (ctx) -> {
430-
final var zeebeDb = ctx.getZeebeDb();
431-
final var keyGenerator = new DbKeyGenerator(1, zeebeDb, ctx.getTransactionContext());
432-
keyGenerator.nextKey();
433-
keyGenerator.nextKey();
434-
keyGenerator.nextKey();
435-
436-
throw new RuntimeException("expected");
437-
};
441+
testProcessor.processingAction =
442+
(ctx) -> {
443+
throw new RuntimeException("expected");
444+
};
445+
testProcessor.onProcessingErrorAction =
446+
(ctx) -> {
447+
final var zeebeDb = ctx.getZeebeDb();
448+
final var keyGenerator = new DbKeyGenerator(1, zeebeDb, ctx.getTransactionContext());
449+
keyGenerator.nextKey();
450+
keyGenerator.nextKey();
451+
keyGenerator.nextKey();
452+
453+
throw new RuntimeException("expected");
454+
};
438455
doCallRealMethod()
439456
.doReturn(EmptyProcessingResult.INSTANCE)
440-
.when(testProcessor).process(any(), any());
457+
.when(testProcessor)
458+
.process(any(), any());
441459
doCallRealMethod()
442460
.doReturn(EmptyProcessingResult.INSTANCE)
443-
.when(testProcessor).onProcessingError(any(), any(), any());
461+
.when(testProcessor)
462+
.onProcessingError(any(), any(), any());
444463
streamPlatform.withRecordProcessors(List.of(testProcessor)).startStreamProcessor();
445464

446465
final var zeebeDb = testProcessor.recordProcessorContext.getZeebeDb();
@@ -465,7 +484,16 @@ public void shouldWriteResponse() {
465484
final var defaultMockedRecordProcessor = streamPlatform.getDefaultMockedRecordProcessor();
466485

467486
final var resultBuilder = new BufferedProcessingResultBuilder((c, s) -> true);
468-
resultBuilder.withResponse(RecordType.EVENT, 3, ELEMENT_ACTIVATING, RECORD, ValueType.PROCESS_INSTANCE, RejectionType.NULL_VAL, "", 1, 12);
487+
resultBuilder.withResponse(
488+
RecordType.EVENT,
489+
3,
490+
ELEMENT_ACTIVATING,
491+
RECORD,
492+
ValueType.PROCESS_INSTANCE,
493+
RejectionType.NULL_VAL,
494+
"",
495+
1,
496+
12);
469497
when(defaultMockedRecordProcessor.process(any(), any()))
470498
.thenReturn(resultBuilder.build())
471499
.thenReturn(EmptyProcessingResult.INSTANCE);
@@ -480,8 +508,7 @@ public void shouldWriteResponse() {
480508
// then
481509
verify(defaultMockedRecordProcessor, TIMEOUT.times(2)).process(any(), any());
482510

483-
final var commandResponseWriter =
484-
streamPlatform.getMockCommandResponseWriter();
511+
final var commandResponseWriter = streamPlatform.getMockCommandResponseWriter();
485512

486513
verify(commandResponseWriter, TIMEOUT.times(1)).key(3);
487514
verify(commandResponseWriter, TIMEOUT.times(1))
@@ -500,7 +527,16 @@ public void shouldWriteResponseOnFailedEventProcessing() {
500527
.thenReturn(EmptyProcessingResult.INSTANCE);
501528

502529
final var resultBuilder = new BufferedProcessingResultBuilder((c, s) -> true);
503-
resultBuilder.withResponse(RecordType.EVENT, 3, ELEMENT_ACTIVATING, RECORD, ValueType.PROCESS_INSTANCE, RejectionType.NULL_VAL, "", 1, 12);
530+
resultBuilder.withResponse(
531+
RecordType.EVENT,
532+
3,
533+
ELEMENT_ACTIVATING,
534+
RECORD,
535+
ValueType.PROCESS_INSTANCE,
536+
RejectionType.NULL_VAL,
537+
"",
538+
1,
539+
12);
504540
when(defaultMockedRecordProcessor.onProcessingError(any(), any(), any()))
505541
.thenReturn(resultBuilder.build())
506542
.thenReturn(EmptyProcessingResult.INSTANCE);
@@ -516,8 +552,7 @@ public void shouldWriteResponseOnFailedEventProcessing() {
516552
verify(defaultMockedRecordProcessor, TIMEOUT.times(2)).process(any(), any());
517553
verify(defaultMockedRecordProcessor, TIMEOUT.times(1)).onProcessingError(any(), any(), any());
518554

519-
final var commandResponseWriter =
520-
streamPlatform.getMockCommandResponseWriter();
555+
final var commandResponseWriter = streamPlatform.getMockCommandResponseWriter();
521556

522557
verify(commandResponseWriter, TIMEOUT.times(1)).key(3);
523558
verify(commandResponseWriter, TIMEOUT.times(1))
@@ -534,8 +569,7 @@ public void shouldInvokeOnProcessedListenerWhenReturnResult() {
534569
final var defaultMockedRecordProcessor = streamPlatform.getDefaultMockedRecordProcessor();
535570

536571
final var resultBuilder = new BufferedProcessingResultBuilder((c, s) -> true);
537-
when(defaultMockedRecordProcessor.process(any(), any()))
538-
.thenReturn(resultBuilder.build());
572+
when(defaultMockedRecordProcessor.process(any(), any())).thenReturn(resultBuilder.build());
539573
streamPlatform.startStreamProcessor();
540574

541575
// when
@@ -623,12 +657,14 @@ public void shouldNotUpdateLastWrittenPositionWhenSkipped() {
623657

624658
// then
625659
verify(defaultRecordProcessor, TIMEOUT.times(2)).process(any(), any());
626-
await("Last written position should be updated").untilAsserted(() ->
627-
assertThat(streamPlatform.getStreamProcessor().getLastWrittenPositionAsync().join()).isEqualTo(-1)
628-
);
660+
await("Last written position should be updated")
661+
.untilAsserted(
662+
() ->
663+
assertThat(streamPlatform.getStreamProcessor().getLastWrittenPositionAsync().join())
664+
.isEqualTo(-1));
629665
}
630666

631-
private final static class TestProcessor implements RecordProcessor {
667+
private static final class TestProcessor implements RecordProcessor {
632668

633669
ProcessingResult processingResult = EmptyProcessingResult.INSTANCE;
634670
ProcessingResult processingResultOnError = EmptyProcessingResult.INSTANCE;
@@ -647,18 +683,18 @@ public boolean accepts(final ValueType valueType) {
647683
}
648684

649685
@Override
650-
public void replay(final TypedRecord record) {
651-
}
686+
public void replay(final TypedRecord record) {}
652687

653688
@Override
654-
public ProcessingResult process(final TypedRecord record,
655-
final ProcessingResultBuilder processingResultBuilder) {
689+
public ProcessingResult process(
690+
final TypedRecord record, final ProcessingResultBuilder processingResultBuilder) {
656691
processingAction.accept(recordProcessorContext);
657692
return processingResult;
658693
}
659694

660695
@Override
661-
public ProcessingResult onProcessingError(final Throwable processingException,
696+
public ProcessingResult onProcessingError(
697+
final Throwable processingException,
662698
final TypedRecord record,
663699
final ProcessingResultBuilder processingResultBuilder) {
664700
onProcessingErrorAction.accept(recordProcessorContext);

0 commit comments

Comments
 (0)