Exceptions swallowed in Actor

[deepthidevaki]

Describe the bug

An error from atomix Raft prevented logstream to open. However instead of completing the future exceptionally, the exception was caught in FutureContinuationRunnable. The exception was logged and the actor continued running. However installation of partition is not completed and health check did not detect it because the future was not completed exceptionally.

I 2020-03-16T21:43:21.016163543Z 2020-03-16 21:43:21.016 [Broker-2-ZeebePartition-1] [Broker-2-zb-actors-0] DEBUG io.zeebe.broker.system - Installing follower partition service for partition PartitionId{id=1, group=raft-partition}
 
I 2020-03-16T21:43:21.016971210Z 2020-03-16 21:43:21.016 [Broker-2-ZeebePartition-1] [Broker-2-zb-actors-0] DEBUG io.zeebe.util.actor - Continuing on future completion failed
 
I 2020-03-16T21:43:21.016986561Z java.lang.IllegalStateException: journal not open
	at com.google.common.base.Preconditions.checkState(Preconditions.java:508) ~[guava-28.2-jre.jar:?]
	at io.atomix.storage.journal.SegmentedJournal.assertOpen(SegmentedJournal.java:279) ~[atomix-storage-3.2.0-20200316.115405-137.jar:?]
	at io.atomix.storage.journal.SegmentedJournal.getSegment(SegmentedJournal.java:409) ~[atomix-storage-3.2.0-20200316.115405-137.jar:?]
	at io.atomix.storage.journal.SegmentedJournalReader.initialize(SegmentedJournalReader.java:42) ~[atomix-storage-3.2.0-20200316.115405-137.jar:?]
	at io.atomix.storage.journal.SegmentedJournalReader.<init>(SegmentedJournalReader.java:35) ~[atomix-storage-3.2.0-20200316.115405-137.jar:?]
	at io.atomix.storage.journal.SegmentedJournal.openReader(SegmentedJournal.java:232) ~[atomix-storage-3.2.0-20200316.115405-137.jar:?]
	at io.atomix.protocols.raft.storage.log.RaftLog.openReader(RaftLog.java:62) ~[atomix-raft-3.2.0-20200316.115448-137.jar:?]
	at io.atomix.protocols.raft.partition.impl.RaftPartitionServer.openReader(RaftPartitionServer.java:198) ~[atomix-raft-3.2.0-20200316.115448-137.jar:?]
	at io.zeebe.logstreams.storage.atomix.AtomixRaftServer.create(AtomixRaftServer.java:30) ~[zeebe-logstreams-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.logstreams.storage.atomix.AtomixReaderFactory.create(AtomixReaderFactory.java:18) ~[zeebe-logstreams-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.logstreams.storage.atomix.AtomixReaderFactory.create(AtomixReaderFactory.java:22) ~[zeebe-logstreams-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.logstreams.storage.atomix.AtomixLogStorage.newReader(AtomixLogStorage.java:40) ~[zeebe-logstreams-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.logstreams.impl.log.LogStreamReaderImpl.<init>(LogStreamReaderImpl.java:41) ~[zeebe-logstreams-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.logstreams.impl.log.LogStreamImpl.<init>(LogStreamImpl.java:84) ~[zeebe-logstreams-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.logstreams.impl.log.LogStreamBuilderImpl.buildAsync(LogStreamBuilderImpl.java:68) ~[zeebe-logstreams-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.broker.system.partitions.ZeebePartition.openLogStream(ZeebePartition.java:641) ~[zeebe-broker-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.broker.system.partitions.ZeebePartition.basePartitionInstallation(ZeebePartition.java:308) ~[zeebe-broker-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.broker.system.partitions.ZeebePartition.installFollowerPartition(ZeebePartition.java:242) ~[zeebe-broker-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.broker.system.partitions.ZeebePartition.lambda$transitionToFollower$6(ZeebePartition.java:234) ~[zeebe-broker-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.util.sched.future.FutureContinuationRunnable.run(FutureContinuationRunnable.java:28) [zeebe-util-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.util.sched.ActorJob.invoke(ActorJob.java:76) [zeebe-util-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.util.sched.ActorJob.execute(ActorJob.java:39) [zeebe-util-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.util.sched.ActorTask.execute(ActorTask.java:115) [zeebe-util-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.util.sched.ActorThread.executeCurrentTask(ActorThread.java:107) [zeebe-util-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.util.sched.ActorThread.doWork(ActorThread.java:91) [zeebe-util-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]
	at io.zeebe.util.sched.ActorThread.run(ActorThread.java:195) [zeebe-util-0.23.0-SNAPSHOT.jar:0.23.0-SNAPSHOT]

To Reproduce
Will come up with a reproducer code later.

Expected behavior
Exceptions should lead to futures completed exceptions sothat they can be detected and handled properly.

[deepthidevaki]

Since the exception occurred in a continuation job which is only a consumer of previously completed future, we cannot complete the future exceptionally. Ideally exceptions should be handled by the task itself. But in case it is not handled by the task, the actor should have a generic handler. As a solution, instead of catching the exception in FutureContinuationRunnable, we should handle them in actor::handleFailure the same way other exceptions from an actor task is handled.