Exceptions in RocksDBIncrementalSnapshotOperation#takeSnapshot cause job to get stuck

final RocksDBIncrementalSnapshotOperation<K> snapshotOperation =
	new RocksDBIncrementalSnapshotOperation<>(
		this,
		checkpointStreamFactory,
		checkpointId,
		checkpointTimestamp);

snapshotOperation.takeSnapshot();

return new FutureTask<KeyedStateHandle>(
	new Callable<KeyedStateHandle>() {
		@Override
		public KeyedStateHandle call() throws Exception {
			return snapshotOperation.materializeSnapshot();
		}
	}
) {
	@Override
	public boolean cancel(boolean mayInterruptIfRunning) {
		snapshotOperation.stop();
		return super.cancel(mayInterruptIfRunning);
	}

	@Override
	protected void done() {
		snapshotOperation.releaseResources(isCancelled());
	}
};