+ private List<ReplicatedLogEntry> getEntriesToSend(final FollowerLogInformation followerLogInfo,
+ final ActorSelection followerActor) {
+ // Try to get all the entries in the journal but not exceeding the max data size for a single AppendEntries
+ // message.
+ int maxEntries = (int) context.getReplicatedLog().size();
+ final int maxDataSize = context.getConfigParams().getSnapshotChunkSize();
+ final long followerNextIndex = followerLogInfo.getNextIndex();
+ List<ReplicatedLogEntry> entries = context.getReplicatedLog().getFrom(followerNextIndex,
+ maxEntries, maxDataSize);
+
+ // If the first entry's size exceeds the max data size threshold, it will be returned from the call above. If
+ // that is the case, then we need to slice it into smaller chunks.
+ if (!(entries.size() == 1 && entries.get(0).getData().size() > maxDataSize)) {
+ // Don't need to slice.
+ return entries;
+ }
+
+ log.debug("{}: Log entry size {} exceeds max payload size {}", logName(), entries.get(0).getData().size(),
+ maxDataSize);
+
+ // If an AppendEntries has already been serialized for the log index then reuse the
+ // SharedFileBackedOutputStream.
+ final Long logIndex = entries.get(0).getIndex();
+ SharedFileBackedOutputStream fileBackedStream = sharedSerializedAppendEntriesStreams.get(logIndex);
+ if (fileBackedStream == null) {
+ fileBackedStream = context.getFileBackedOutputStreamFactory().newSharedInstance();
+
+ final AppendEntries appendEntries = new AppendEntries(currentTerm(), context.getId(),
+ getLogEntryIndex(followerNextIndex - 1), getLogEntryTerm(followerNextIndex - 1), entries,
+ context.getCommitIndex(), getReplicatedToAllIndex(), context.getPayloadVersion());
+
+ log.debug("{}: Serializing {} for slicing for follower {}", logName(), appendEntries,
+ followerLogInfo.getId());
+
+ try (ObjectOutputStream out = new ObjectOutputStream(fileBackedStream)) {
+ out.writeObject(appendEntries);
+ } catch (IOException e) {
+ log.error("{}: Error serializing {}", logName(), appendEntries, e);
+ fileBackedStream.cleanup();
+ return Collections.emptyList();
+ }
+
+ sharedSerializedAppendEntriesStreams.put(logIndex, fileBackedStream);
+
+ fileBackedStream.setOnCleanupCallback(index -> {
+ log.debug("{}: On SharedFileBackedOutputStream cleanup for index {}", logName(), index);
+ sharedSerializedAppendEntriesStreams.remove(index);
+ }, logIndex);
+ } else {
+ log.debug("{}: Reusing SharedFileBackedOutputStream for follower {}", logName(), followerLogInfo.getId());
+ fileBackedStream.incrementUsageCount();
+ }
+
+ log.debug("{}: Slicing stream for index {}, follower {}", logName(), logIndex, followerLogInfo.getId());
+
+ // Record that slicing is in progress for the follower.
+ followerLogInfo.setSlicedLogEntryIndex(logIndex);
+
+ final FollowerIdentifier identifier = new FollowerIdentifier(followerLogInfo.getId());
+ appendEntriesMessageSlicer.slice(SliceOptions.builder().identifier(identifier)
+ .fileBackedOutputStream(fileBackedStream).sendTo(followerActor).replyTo(actor())
+ .onFailureCallback(failure -> {
+ log.error("{}: Error slicing AppendEntries for follower {}", logName(),
+ followerLogInfo.getId(), failure);
+ followerLogInfo.setSlicedLogEntryIndex(FollowerLogInformation.NO_INDEX);
+ }).build());
+
+ return Collections.emptyList();
+ }
+
+ private void sendAppendEntriesToFollower(final ActorSelection followerActor, final List<ReplicatedLogEntry> entries,
+ final FollowerLogInformation followerLogInformation) {
+ // In certain cases outlined below we don't want to send the actual commit index to prevent the follower from
+ // possibly committing and applying conflicting entries (those with same index, different term) from a prior
+ // term that weren't replicated to a majority, which would be a violation of raft.
+ // - if the follower isn't active. In this case we don't know the state of the follower and we send an
+ // empty AppendEntries as a heart beat to prevent election.
+ // - if we're in the process of installing a snapshot. In this case we don't send any new entries but still
+ // need to send AppendEntries to prevent election.
+ // - if we're in the process of slicing an AppendEntries with a large log entry payload. In this case we
+ // need to send an empty AppendEntries to prevent election.
+ boolean isInstallingSnaphot = followerLogInformation.getInstallSnapshotState() != null;
+ long leaderCommitIndex = isInstallingSnaphot || followerLogInformation.isLogEntrySlicingInProgress()
+ || !followerLogInformation.isFollowerActive() ? -1 : context.getCommitIndex();
+
+ long followerNextIndex = followerLogInformation.getNextIndex();