IllegalStateException in StreamProcessor : Cannot complete future, the future is already completed

[deepthidevaki]

Describe the bug

java.lang.IllegalStateException: Cannot complete future, the future is already completed exceptionally with 'Expected to claim segment of size 20033584, but can't claim more than 4194304 bytes.' 
	at io.camunda.zeebe.scheduler.future.CompletableActorFuture.completeExceptionally(CompletableActorFuture.java:186) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.future.CompletableActorFuture.completeExceptionally(CompletableActorFuture.java:193) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.retry.AbortableRetryStrategy.run(AbortableRetryStrategy.java:50) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.ActorJob.invoke(ActorJob.java:92) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.ActorJob.execute(ActorJob.java:45) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.ActorTask.execute(ActorTask.java:119) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.ActorThread.executeCurrentTask(ActorThread.java:106) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.ActorThread.doWork(ActorThread.java:87) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.ActorThread.run(ActorThread.java:198) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
Caused by: java.lang.IllegalArgumentException: Expected to claim segment of size 20033584, but can't claim more than 4194304 bytes.
	at io.camunda.zeebe.dispatcher.Dispatcher.offer(Dispatcher.java:207) ~[zeebe-dispatcher-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.dispatcher.Dispatcher.claimFragmentBatch(Dispatcher.java:164) ~[zeebe-dispatcher-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.logstreams.impl.log.LogStreamBatchWriterImpl.claimBatchForEvents(LogStreamBatchWriterImpl.java:235) ~[zeebe-logstreams-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.logstreams.impl.log.LogStreamBatchWriterImpl.tryWrite(LogStreamBatchWriterImpl.java:212) ~[zeebe-logstreams-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.streamprocessor.ProcessingScheduleServiceImpl.lambda$toRunnable$4(ProcessingScheduleServiceImpl.java:127) ~[zeebe-workflow-engine-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.retry.ActorRetryMechanism.run(ActorRetryMechanism.java:36) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.retry.AbortableRetryStrategy.run(AbortableRetryStrategy.java:45) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	... 6 more

StreamProcessor actor failed and the partition is unhealthy.

To Reproduce

Observed this when running an e2e test with a new process added in this PR

Expected behavior

StreamProcessor should not fail.

Log/Stacktrace

Logs

Log explorer

Logs before this error that might be relevant

2022-08-31 16:28:33.521 CEST
Expected to find job with key 6755399441066071, but no job found
2022-08-31 16:28:33.521 CEST
Expected to find job with key 6755399441066104, but no job found
2022-08-31 16:28:33.521 CEST
Expected to find job with key 6755399441067170, but no job found
2022-08-31 16:28:33.521 CEST
Expected to find job with key 6755399441067940, but no job found
2022-08-31 16:31:10.218 CEST
Failed to write command MESSAGE_SUBSCRIPTION CREATE from 0 to logstream
2022-08-31 16:31:10.218 CEST
Failed to write command MESSAGE_SUBSCRIPTION CREATE from 0 to logstream
2022-08-31 16:31:10.219 CEST
Failed to write command MESSAGE_SUBSCRIPTION CREATE from 0 to logstream
2022-08-31 16:31:10.219 CEST
.....
Failed to write command MESSAGE_SUBSCRIPTION CREATE from 0 to logstream
2022-08-31 16:31:40.226 CEST
Failed to write command MESSAGE_SUBSCRIPTION CREATE from 0 to logstream
2022-08-31 16:31:40.226 CEST
Failed to write command MESSAGE_SUBSCRIPTION CREATE from 0 to logstream
2022-08-31 16:31:48.722 CEST
Actor Broker-2-StreamProcessor-3 failed in phase STARTED.
2022-08-31 16:31:48.728 CEST
Discard job io.camunda.zeebe.scheduler.retry.AbortableRetryStrategy$$Lambda$2505/0x00000008016d7530 QUEUED from fastLane of Actor Broker-2-StreamProcessor-3.
2022-08-31 16:31:48.728 CEST
Discard job io.camunda.zeebe.streamprocessor.StreamProcessor$$Lambda$2317/0x0000000801677500 QUEUED from fastLane of Actor Broker-2-StreamProcessor-3.

Environment:

• OS: camunda cloud
• Zeebe Version: 8.1.0-SNAPSHOT

[megglos]

Triage: Streamprocessor got stuck unhealthy and it didn't seem to recover.
Seems related to recent refactoring.

[npepinpe]

Interesting that the error is that it tried to claim almost 20MB from the dispatcher 🤯 Is that expected from your test?

[deepthidevaki]

Interesting that the error is that it tried to claim almost 20MB from the dispatcher exploding_head Is that expected from your test?

Not expected, I guess 🤔 . Variables are small. It is running workers with default config (except for the timeout).

[npepinpe]

Could it be possible we have an unexpected loop? :s Perhaps putting submit instead of run in the retry is causing more issues than expected. I'll have to reproduce it locally to confirm.

[deepthidevaki]

It occurred again here. I can keep the cluster running if it is useful for you. Let me know if you need any other help in reproducing it.

java.lang.IllegalStateException: Cannot complete future, the future is already completed  with value true
	at io.camunda.zeebe.scheduler.future.CompletableActorFuture.completeExceptionally(CompletableActorFuture.java:186) ~[zeebe-scheduler-8.1.0-alpha5.jar:8.1.0-alpha5]
	at io.camunda.zeebe.scheduler.future.CompletableActorFuture.completeExceptionally(CompletableActorFuture.java:193) ~[zeebe-scheduler-8.1.0-alpha5.jar:8.1.0-alpha5]
	at io.camunda.zeebe.scheduler.retry.AbortableRetryStrategy.run(AbortableRetryStrategy.java:50) ~[zeebe-scheduler-8.1.0-alpha5.jar:8.1.0-alpha5]
	at io.camunda.zeebe.scheduler.ActorJob.invoke(ActorJob.java:92) ~[zeebe-scheduler-8.1.0-alpha5.jar:8.1.0-alpha5]
	at io.camunda.zeebe.scheduler.ActorJob.execute(ActorJob.java:45) ~[zeebe-scheduler-8.1.0-alpha5.jar:8.1.0-alpha5]
	at io.camunda.zeebe.scheduler.ActorTask.execute(ActorTask.java:119) ~[zeebe-scheduler-8.1.0-alpha5.jar:8.1.0-alpha5]
	at io.camunda.zeebe.scheduler.ActorThread.executeCurrentTask(ActorThread.java:106) ~[zeebe-scheduler-8.1.0-alpha5.jar:8.1.0-alpha5]
	at io.camunda.zeebe.scheduler.ActorThread.doWork(ActorThread.java:87) ~[zeebe-scheduler-8.1.0-alpha5.jar:8.1.0-alpha5]
	at io.camunda.zeebe.scheduler.ActorThread.run(ActorThread.java:198) ~[zeebe-scheduler-8.1.0-alpha5.jar:8.1.0-alpha5]
Caused by: java.lang.IllegalStateException: Cannot complete future, the future is already completed  with value true
	at io.camunda.zeebe.scheduler.future.CompletableActorFuture.complete(CompletableActorFuture.java:166) ~[zeebe-scheduler-8.1.0-alpha5.jar:8.1.0-alpha5]
	at io.camunda.zeebe.scheduler.retry.ActorRetryMechanism.run(ActorRetryMechanism.java:37) ~[zeebe-scheduler-8.1.0-alpha5.jar:8.1.0-alpha5]
	at io.camunda.zeebe.scheduler.retry.AbortableRetryStrategy.run(AbortableRetryStrategy.java:45) ~[zeebe-scheduler-8.1.0-alpha5.jar:8.1.0-alpha5]
	... 6 more

[npepinpe]

I think removing runUntilDone created some unexpected side effects with the current refactoring.

Since we moved from runUntilDone to a run followed by submit calls, it's possible for records to be written to the dispatcher out of order, in theory (in practice, hard to say, possibly there is no failure case where this is possible).

Say the processing schedule service wants to write a result, but for whatever reason the first try fails. It will then submit a job to retry. At this point we can continue processing. Possibly readNextRecord was already enqueued. So we process the record, then write the results. This calls run via the AbortableRetryStrategy, so the task to write here is prepended. Considering the commands we wrote may have been based on state before the record was processed, this could be wrong. I think the only way to ensure correct ordering is to use run everywhere or submit everywhere, but not both.

It's kind of late though, so I'll think on it again in the morning with fresh eyes, maybe I'm missing something. It also doesn't quite explain why the error occurred, though it hints that there are some concurrent issues.

[npepinpe]

I think our assumption was that any state machine should still guard against the execution being interleaved, but I don't know if we do that in every case 😄 It seems very brittle without something like the facilities to declare how FSMs handle which messages in which state in Akka.

[npepinpe]

I have some hint that the ProcessingScheduleServiceImpl may be an issue. It reuses its own writeRetryStrategy, and calls runWithRetry every time a submitted runnable runs. This means we overwrite the underlying ActorRetryMechanism on every runWithRetry call. But what if we had already enqueued a retry?

Say we schedule two runnables, both will want to write some commands. Say they were to expire/be executed one right after the other, e.g. the actor queue is [runnable1, runnable2]. The first one runs and calls runWithRetry, which creates a new future in AbortableRetryStrategy, sets it to ActorRetryMechanism. This calls prepends AbortableRetryStrategy::run on the actor via actor.run. So the queue is now: [AbortableRetryStrategy::run, runnable2]. Let's say the operation fails and we retry. So we'll submit the retried operation. The queue would look like [runnable2, retry1] (not exactly, as we have two queues actually, fun times, but you get the idea). runnable2 runs, calls runWithRetry, which creates a new future (!) and overwrites it on the ActorRetryMechanism (!). Then it calls actor.run(AbortableRetryStrategy::run). So the queue is now: [AbortableRetryStrategy::run, retry1], and the first future will never be completely (sad trombone). If the run now succeeds, the future is completed, then retry1 executes and tries to complete the future. If the run didn't succeed, it's now submitted again, and the same thing happens (e.g. the future will be "completed" twice).

I think this doesn't happen in the processing state machine because it's a state machine, so it's always in one state and the futures are always completed before the retry strategy is reused.

[deepthidevaki]

Makes sense.

As I understood, this happens because we submit for retry. If it succeeds in the first attempt itself, this is not a problem because the order is preserved due to run. So the retryMechanism and the currentFuture won't be overwritten. Is that correct?

[npepinpe]

Correct. I think if we switch to run instead of submit, it will be mostly fine, but as this is quite hard to test for (and the class itself is hard to test), I can't fully guarantee it without running through more hand-made scenarios 😞

Maybe we should think about property based testing this as well.

So if we replace submit with run, then we are always prepending the next retry to the fast lane queue, since we're always doing this from the same actor. And since it's from the same actor, no other calls can prepend a job before that job is executed, because:

• call will submit a job
• subscriptions (incl. timers) will also submit jobs to the fast lane (so append)
• run will only prepend a job if the run call is being done by the current actor. If the run call is done outside of the actor, then it is appended.

So I would suggest as a quick fix to go back to using run instead of submit for places where we were using runUntilDone.

[npepinpe]

Now I just need to figure out how to write a regression test for this before writing the fix 🤔

[deepthidevaki]

I was also wondering why we are reusing the currentFuture and retryMechanism in AbortableRetryStrategy. Can we remove that? Each retry task will keep track of its future and the task to rerun. So concurrent retryable tasks are not a problem.

[npepinpe]

That's a good idea, but I'm wondering if that's enough (without reverting submit to run) when it comes to side effect, since now the records (from the different runWithRetry calls) may be written in different orders. Is that...safe?

Another question, in the ProcessingScheduleService, shouldn't we reschedule the next run only when the write future is completed? Right now, we reschedule it after the task run. If the write needs to be retried, we may schedule the next iteration already before the write is successful. If it required multiple iterations, then for whatever reason it could be the next task would be scheduled. It's probably safe at the moment, but only because of the nature of the scheduled tasks...?

[deepthidevaki]

That's a good idea, but I'm wondering if that's enough (without reverting submit to run) when it comes to side effect, since now the records (from the different runWithRetry calls) may be written in different orders. Is that...safe?

I'm not sure about the expected semantics for side effects.

Another question, in the ProcessingScheduleService, shouldn't we reschedule the next run only when the write future is completed?

I don't understand. Rescheduling is done by AboratableRetryStrategy, right? ProcessSchedulingService is only waiting for the future to log incase it failed. And it is only retried after the "write" failed.

[npepinpe]

Yes, but the scheduling service reschedules the original timer-based task immediately before the write is retried. The next timer can then call runWithRetry again. So for example, if you have the message time to live checker, it might run, try to write some commands, be retried, then the next time to live checker run executes and writes more commands.

I think that's generally OK-ish because we always write commands at the moment, and these may be rejected, but if we ever wrote state updates it would be a big problem I think.

[deepthidevaki]

Ok. This is for runAtFixedRate. Yes, it is possible.

[npepinpe]

Follow up issue: #10306