- if (context.getId().equals(currentBehavior.getLeaderId())
- && captureSnapshot.isInstallSnapshotInitiated()) {
- // this would be call straight to the leader and won't initiate in serialization
- currentBehavior.handleMessage(context.getActor(), new SendInstallSnapshot(snapshot));
+ if (installSnapshotStream.isPresent()) {
+ if (context.getId().equals(currentBehavior.getLeaderId())) {
+ try {
+ ByteSource snapshotBytes = ((FileBackedOutputStream)installSnapshotStream.get()).asByteSource();
+ currentBehavior.handleMessage(context.getActor(),
+ new SendInstallSnapshot(snapshot, snapshotBytes));
+ } catch (IOException e) {
+ log.error("{}: Snapshot install failed due to an unrecoverable streaming error",
+ context.getId(), e);
+ }
+ } else {
+ ((FileBackedOutputStream)installSnapshotStream.get()).cleanup();
+ }