Corrupt partition prevents startup of node

[megglos]

Describe the bug

A benchmark node became unhealthy after a restart because of being stuck at starting the Partition Server for partition 2.

To Reproduce

Zeebe node 2 in medic-cw-37-de38e9e086-benchmark is stuck at this issue.

Expected behavior

Would it make sense to discard the log and/or snapshot in this case to recover?

Log/Stacktrace

Full Stacktrace

java.util.concurrent.ExecutionException: Startup failed in the following steps: [Partition Manager]. See suppressed exceptions for details.
	at io.camunda.zeebe.scheduler.future.CompletableActorFuture.get(CompletableActorFuture.java:142) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.future.CompletableActorFuture.get(CompletableActorFuture.java:109) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.FutureUtil.join(FutureUtil.java:21) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.future.CompletableActorFuture.join(CompletableActorFuture.java:198) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.broker.Broker.internalStart(Broker.java:101) ~[zeebe-broker-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.util.LogUtil.doWithMDC(LogUtil.java:23) ~[zeebe-util-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.broker.Broker.start(Broker.java:83) ~[zeebe-broker-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.broker.StandaloneBroker.run(StandaloneBroker.java:95) ~[camunda-zeebe-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:771) ~[spring-boot-2.7.3.jar:2.7.3]
	at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:755) ~[spring-boot-2.7.3.jar:2.7.3]
	at org.springframework.boot.SpringApplication.run(SpringApplication.java:315) ~[spring-boot-2.7.3.jar:2.7.3]
	at io.camunda.zeebe.broker.StandaloneBroker.main(StandaloneBroker.java:81) ~[camunda-zeebe-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
Caused by: io.camunda.zeebe.scheduler.startup.StartupProcessException: Startup failed in the following steps: [Partition Manager]. See suppressed exceptions for details.
	at io.camunda.zeebe.scheduler.startup.StartupProcess.aggregateExceptionsSynchronized(StartupProcess.java:282) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.startup.StartupProcess.completeStartupFutureExceptionallySynchronized(StartupProcess.java:183) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.startup.StartupProcess.lambda$proceedWithStartupSynchronized$3(StartupProcess.java:167) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.future.FutureContinuationRunnable.run(FutureContinuationRunnable.java:33) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
	at io.camunda.zeebe.scheduler.ActorJob.invoke(ActorJob.java:94) ~[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]
	Suppressed: io.camunda.zeebe.scheduler.startup.StartupProcessStepException: Bootstrap step Partition Manager failed
		at io.camunda.zeebe.scheduler.startup.StartupProcess.completeStartupFutureExceptionallySynchronized(StartupProcess.java:185) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.camunda.zeebe.scheduler.startup.StartupProcess.lambda$proceedWithStartupSynchronized$3(StartupProcess.java:167) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.camunda.zeebe.scheduler.future.FutureContinuationRunnable.run(FutureContinuationRunnable.java:33) ~[zeebe-scheduler-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.camunda.zeebe.scheduler.ActorJob.invoke(ActorJob.java:94) ~[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.util.concurrent.CompletionException: java.lang.IllegalStateException: Expected to find a snapshot at index >= log's first index 213419494, but found snapshot 213252958. A previous snapshot is most likely corrupted.
		at java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source) ~[?:?]
		at java.util.concurrent.CompletableFuture.uniApplyNow(Unknown Source) ~[?:?]
		at java.util.concurrent.CompletableFuture.uniApplyStage(Unknown Source) ~[?:?]
		at java.util.concurrent.CompletableFuture.thenApply(Unknown Source) ~[?:?]
		at io.atomix.raft.partition.RaftPartition.open(RaftPartition.java:106) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.atomix.raft.partition.RaftPartitionGroup.lambda$join$6(RaftPartitionGroup.java:193) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at java.util.stream.ReferencePipeline$3$1.accept(Unknown Source) ~[?:?]
		at java.util.HashMap$KeySpliterator.forEachRemaining(Unknown Source) ~[?:?]
		at java.util.stream.AbstractPipeline.copyInto(Unknown Source) ~[?:?]
		at java.util.stream.AbstractPipeline.wrapAndCopyInto(Unknown Source) ~[?:?]
		at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(Unknown Source) ~[?:?]
		at java.util.stream.AbstractPipeline.evaluate(Unknown Source) ~[?:?]
		at java.util.stream.ReferencePipeline.collect(Unknown Source) ~[?:?]
		at io.atomix.raft.partition.RaftPartitionGroup.join(RaftPartitionGroup.java:195) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.atomix.primitive.partition.impl.DefaultPartitionService.start(DefaultPartitionService.java:63) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.camunda.zeebe.broker.partitioning.PartitionManagerImpl.start(PartitionManagerImpl.java:125) ~[zeebe-broker-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.camunda.zeebe.broker.bootstrap.PartitionManagerStep.lambda$startupInternal$4(PartitionManagerStep.java:42) ~[zeebe-broker-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at java.util.concurrent.CompletableFuture$AsyncRun.run(Unknown Source) ~[?:?]
		at java.util.concurrent.CompletableFuture$AsyncRun.exec(Unknown Source) ~[?:?]
		at java.util.concurrent.ForkJoinTask.doExec(Unknown Source) ~[?:?]
		at java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(Unknown Source) ~[?:?]
		at java.util.concurrent.ForkJoinPool.scan(Unknown Source) ~[?:?]
		at java.util.concurrent.ForkJoinPool.runWorker(Unknown Source) ~[?:?]
		at java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source) ~[?:?]
	Caused by: java.lang.IllegalStateException: Expected to find a snapshot at index >= log's first index 213419494, but found snapshot 213252958. A previous snapshot is most likely corrupted.
		at io.atomix.raft.impl.RaftContext.verifySnapshotLogConsistent(RaftContext.java:226) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.atomix.raft.impl.RaftContext.<init>(RaftContext.java:203) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.atomix.raft.impl.DefaultRaftServer$Builder.build(DefaultRaftServer.java:258) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.atomix.raft.impl.DefaultRaftServer$Builder.build(DefaultRaftServer.java:232) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.atomix.raft.partition.impl.RaftPartitionServer.buildServer(RaftPartitionServer.java:190) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.atomix.raft.partition.impl.RaftPartitionServer.initServer(RaftPartitionServer.java:156) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.atomix.raft.partition.impl.RaftPartitionServer.start(RaftPartitionServer.java:114) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.atomix.raft.partition.RaftPartition.open(RaftPartition.java:106) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.atomix.raft.partition.RaftPartitionGroup.lambda$join$6(RaftPartitionGroup.java:193) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at java.util.stream.ReferencePipeline$3$1.accept(Unknown Source) ~[?:?]
		at java.util.HashMap$KeySpliterator.forEachRemaining(Unknown Source) ~[?:?]
		at java.util.stream.AbstractPipeline.copyInto(Unknown Source) ~[?:?]
		at java.util.stream.AbstractPipeline.wrapAndCopyInto(Unknown Source) ~[?:?]
		at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(Unknown Source) ~[?:?]
		at java.util.stream.AbstractPipeline.evaluate(Unknown Source) ~[?:?]
		at java.util.stream.ReferencePipeline.collect(Unknown Source) ~[?:?]
		at io.atomix.raft.partition.RaftPartitionGroup.join(RaftPartitionGroup.java:195) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.atomix.primitive.partition.impl.DefaultPartitionService.start(DefaultPartitionService.java:63) ~[zeebe-atomix-cluster-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.camunda.zeebe.broker.partitioning.PartitionManagerImpl.start(PartitionManagerImpl.java:125) ~[zeebe-broker-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at io.camunda.zeebe.broker.bootstrap.PartitionManagerStep.lambda$startupInternal$4(PartitionManagerStep.java:42) ~[zeebe-broker-8.1.0-SNAPSHOT.jar:8.1.0-SNAPSHOT]
		at java.util.concurrent.CompletableFuture$AsyncRun.run(Unknown Source) ~[?:?]
		at java.util.concurrent.CompletableFuture$AsyncRun.exec(Unknown Source) ~[?:?]
		at java.util.concurrent.ForkJoinTask.doExec(Unknown Source) ~[?:?]
		at java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(Unknown Source) ~[?:?]
		at java.util.concurrent.ForkJoinPool.scan(Unknown Source) ~[?:?]
		at java.util.concurrent.ForkJoinPool.runWorker(Unknown Source) ~[?:?]
		at java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source) ~[?:?]

Environment:

• OS:
• Zeebe Version:
• Configuration:

Partition logs:
partition-2-logs.tar.gz

[Zelldon]

@megglos maybe you can attach also the data for the partition

[megglos]

I added the partition logs at least, the snapshot is 8-9GB. I'm downloading it now and will put it in a drive.

[npepinpe]

I've added the severity to be critical if we really suspect corruption, as imo that would be a release blocker (at least until we figure out whether this is already present in the latest release or if its something we introduced with 8.1)

Don't hesitate to reach out for more 👀 and 🧠 to solve this

[Zelldon]

Agree here with @npepinpe

[Zelldon]

I guess saving the logs would be also great.

[deepthidevaki]

I can take a look at it. if no one is. I'm waiting for my PRs to merge anyway 🤞

[deepthidevaki]

Here is what we observe in the log

Partition two starts receiving snapshot from the leader.

2022-09-22 02:49:42.423 CEST
2
zeebe
RaftServer{raft-partition-partition-2}{role=FOLLOWER} - Started receiving new snapshot FileBasedReceivedSnapshot{directory=/usr/local/zeebe/data/raft-partition/partitions/2/pending/213419493-39-623283423-623282489-4, snapshotStore=Broker-2-SnapshotStore-2, metadata=FileBasedSnapshotId{index=213419493, term=39, processedPosition=623283423, exporterPosition=623282489}} from 0

Then it resets the log, because it has to throw away the log when a follower receives a snapshot.

2022-09-22 02:54:43.631 CEST
2
zeebe
RaftServer{raft-partition-partition-2}{role=FOLLOWER} - Delete existing log (lastIndex '213418356') and replace with received snapshot (index '213419493'). First entry in the log will be at index 213419494

Immediately, the broker is shutting down.

2022-09-22 02:55:15.456 CEST
zeebe
Shutdown was called with context: io.camunda.zeebe.broker.bootstrap.BrokerStartupContextImpl@66475010

2022-09-22 02:55:15.472 CEST
2
zeebe
Closing ZeebePartition 2

But we don't see a log that says snapshot 213419493-39-623283423-623282489 is committed.

After restart, we see that the snapshot is not committed. So it is deleted and snapshot store loads the previous snapshot. (which is the correct behavior).

2022-09-22 02:59:29.073 CEST
2
zeebe
Snapshot /usr/local/zeebe/data/raft-partition/partitions/2/snapshots/213419493-39-623283423-623282489 does not have a checksum file, which most likely indicates a partial write (e.g. crash during move), and will be deleted
2022-09-22 02:59:29.137 CEST
2
zeebe
Deleting snapshots other than 213252958-39-622784900-622784131

But at this time the journal is at index 213419493, but the snapshot at that index is not committed. So startup consistency check fails - Expected to find a snapshot at index >= log's first index 213419494, but found snapshot 213252958.

[npepinpe]

Yikes, so we threw away the log before the snapshot was committed? :s

It seems we do this to avoid an existing edge case where we have a snapshot but the log was not reset. So I guess we need to make reset + commit somehow atomic 😄

[Zelldon]

Thanks for checking @deepthidevaki !

1. So recovery strategy (manually): delete data from that node?
2. Avoiding this issue is to only delete log after snapshot is committed?

[deepthidevaki]

Resetting the log before snapshot is committed was added recently in this PR. Assumption was that, if the system crashes before snapshot is committed, it will start with an empty state. This is ok because we can handle node restarts with data loss. But even though journal is empty, it's index is updated. This is why the check fails.

[deepthidevaki]

make reset + commit somehow atomic

This would be difficult to achieve, I assume 😄

Avoiding this issue is to only delete log after snapshot is committed?

This will re-introduce the edge case that we tried to fix, no? 🤔

I'm thinking about the following fix:
We will continue the current approach where we reset the log before committing the snapshot. In normal case this is fine. Snapshot will be committed. Log is at the correct index. If the system crashes after committing snapshot, then also the state is not corrupted. We have a snapshot and an empty log. If the system crashed before committing snapshot, then we have an empty journal with out a snapshot at that index.
On startup, do not fail consistency check if the journal is empty. If journal is empty and the corresponding snapshot does not exist, then reset the journal to the existing snapshot index. The current leader will replicate a newer snapshot or events, so no data is lost.

Would this hide any of the valid corruption cases?

[deepthidevaki]

1. So recovery strategy (manually): delete data from that node?

Yes. Manual recovery is possible in this case. But since this is not actually a corruption, but a bug, we should avoid this.

[lenaschoenburg]

I don't see any issues with this approach but since I also missed this in the initial review of #10183 I'm not overly confident in my assessment so a second opinion would be great 👀.

[deepthidevaki]

Let's look at different scenarios, where the snapshot is actually corrupted after they have been successfully committed.

1. Snapshot at index X is corrupted, first log index = X+1, log is not empty
   In this case consistency check will fail.

2. Snapshot at index X is corrupted, first log index = X+1, log is empty
   In this case, consistency check detects it as gap. But since the log is empty, it consider this as a valid state and the check passes. This can look like an issue, but I would argue that this is ok. Because, if the log is empty, we can safely assume that this node was a follower before restart and was not in the quorum. If the follower was in the quorum, there will be always at least one committed entry in the log. This is because we (zeebe snapshot director) always take a snapshot at previous index of snapshot position.
   Another case if the log is empty because it is also corrupted. But we have also persisted lastWrittenPostiion in the meta store. So if the log is corrupted, it would be detected even earlier. Now if meta store is also corrupted - in the best case it fails when trying to read it, in worst case it reads a wrong lastWrittenPosition. So if the snapshot is corrupted, the log is empty because it is corrupted, and meta store is corrupted and reads a wrong lastWrittenPosition, then we wouldn't be able to detect that the state is invalid.

[deepthidevaki]

So if the snapshot is corrupted, the log is empty because it is corrupted, and meta store is corrupted and reads a wrong lastWrittenPosition, then we wouldn't be able to detect that the state is invalid.

In summary, this is one case where we won't be able to detect the corrupted state. But this can happen even now.