opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AppendEntries.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.raft.messages;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.annotations.VisibleForTesting;
  13 import com.google.common.collect.ImmutableList;
  14 import java.io.Externalizable;
  15 import java.io.IOException;
  16 import java.io.ObjectInput;
  17 import java.io.ObjectOutput;
  18 import java.util.List;
  19 import java.util.Optional;
  20 import org.eclipse.jdt.annotation.NonNull;
  21 import org.eclipse.jdt.annotation.Nullable;
  22 import org.opendaylight.controller.cluster.raft.RaftVersions;
  23 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
  24 import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
  25
  26 /**
  27  * Invoked by leader to replicate log entries (§5.3); also used as heartbeat (§5.2).
  28  */
  29 public final class AppendEntries extends AbstractRaftRPC {
  30     @java.io.Serial
  31     private static final long serialVersionUID = 1L;
  32
  33     // So that follower can redirect clients
  34     private final @NonNull String leaderId;
  35
  36     // Index of log entry immediately preceding new ones
  37     private final long prevLogIndex;
  38
  39     // term of prevLogIndex entry
  40     private final long prevLogTerm;
  41
  42     // log entries to store (empty for heart beat - may send more than one for efficiency)
  43     private final @NonNull List<ReplicatedLogEntry> entries;
  44
  45     // leader's commitIndex
  46     private final long leaderCommit;
  47
  48     // index which has been replicated successfully to all followers, -1 if none
  49     private final long replicatedToAllIndex;
  50
  51     private final short payloadVersion;
  52
  53     private final short recipientRaftVersion;
  54
  55     private final short leaderRaftVersion;
  56
  57     private final String leaderAddress;
  58
  59     AppendEntries(final long term, @NonNull final String leaderId, final long prevLogIndex,
  60             final long prevLogTerm, @NonNull final List<ReplicatedLogEntry> entries, final long leaderCommit,
  61             final long replicatedToAllIndex, final short payloadVersion, final short recipientRaftVersion,
  62             final short leaderRaftVersion, @Nullable final String leaderAddress) {
  63         super(term);
  64         this.leaderId = requireNonNull(leaderId);
  65         this.prevLogIndex = prevLogIndex;
  66         this.prevLogTerm = prevLogTerm;
  67         this.entries = requireNonNull(entries);
  68         this.leaderCommit = leaderCommit;
  69         this.replicatedToAllIndex = replicatedToAllIndex;
  70         this.payloadVersion = payloadVersion;
  71         this.recipientRaftVersion = recipientRaftVersion;
  72         this.leaderRaftVersion = leaderRaftVersion;
  73         this.leaderAddress = leaderAddress;
  74     }
  75
  76     public AppendEntries(final long term, final @NonNull String leaderId, final long prevLogIndex,
  77             final long prevLogTerm, final @NonNull List<ReplicatedLogEntry> entries, final long leaderCommit,
  78             final long replicatedToAllIndex, final short payloadVersion, final short recipientRaftVersion,
  79             final @Nullable String leaderAddress) {
  80         this(term, leaderId, prevLogIndex, prevLogTerm, entries, leaderCommit, replicatedToAllIndex, payloadVersion,
  81                 recipientRaftVersion, RaftVersions.CURRENT_VERSION, leaderAddress);
  82     }
  83
  84     @VisibleForTesting
  85     public AppendEntries(final long term, final @NonNull String leaderId, final long prevLogIndex,
  86             final long prevLogTerm, final @NonNull List<ReplicatedLogEntry> entries, final long leaderCommit,
  87             final long replicatedToAllIndex, final short payloadVersion) {
  88         this(term, leaderId, prevLogIndex, prevLogTerm, entries, leaderCommit, replicatedToAllIndex, payloadVersion,
  89                 RaftVersions.CURRENT_VERSION, null);
  90     }
  91
  92     public @NonNull String getLeaderId() {
  93         return leaderId;
  94     }
  95
  96     public long getPrevLogIndex() {
  97         return prevLogIndex;
  98     }
  99
 100     public long getPrevLogTerm() {
 101         return prevLogTerm;
 102     }
 103
 104     public @NonNull List<ReplicatedLogEntry> getEntries() {
 105         return entries;
 106     }
 107
 108     public long getLeaderCommit() {
 109         return leaderCommit;
 110     }
 111
 112     public long getReplicatedToAllIndex() {
 113         return replicatedToAllIndex;
 114     }
 115
 116     public short getPayloadVersion() {
 117         return payloadVersion;
 118     }
 119
 120     public Optional<String> getLeaderAddress() {
 121         return Optional.ofNullable(leaderAddress);
 122     }
 123
 124     public short getLeaderRaftVersion() {
 125         return leaderRaftVersion;
 126     }
 127
 128     @Override
 129     public String toString() {
 130         return "AppendEntries [leaderId=" + leaderId
 131                 + ", prevLogIndex=" + prevLogIndex
 132                 + ", prevLogTerm=" + prevLogTerm
 133                 + ", leaderCommit=" + leaderCommit
 134                 + ", replicatedToAllIndex=" + replicatedToAllIndex
 135                 + ", payloadVersion=" + payloadVersion
 136                 + ", recipientRaftVersion=" + recipientRaftVersion
 137                 + ", leaderRaftVersion=" + leaderRaftVersion
 138                 + ", leaderAddress=" + leaderAddress
 139                 + ", entries=" + entries + "]";
 140     }
 141
 142     @Override
 143     Object writeReplace() {
 144         return recipientRaftVersion <= RaftVersions.FLUORINE_VERSION ? new ProxyV2(this) : new AE(this);
 145     }
 146
 147     /**
 148      * Fluorine version that adds the leader address.
 149      */
 150     private static class ProxyV2 implements Externalizable {
 151         @java.io.Serial
 152         private static final long serialVersionUID = 1L;
 153
 154         private AppendEntries appendEntries;
 155
 156         // checkstyle flags the public modifier as redundant which really doesn't make sense since it clearly isn't
 157         // redundant. It is explicitly needed for Java serialization to be able to create instances via reflection.
 158         @SuppressWarnings("checkstyle:RedundantModifier")
 159         public ProxyV2() {
 160         }
 161
 162         ProxyV2(final AppendEntries appendEntries) {
 163             this.appendEntries = appendEntries;
 164         }
 165
 166         @Override
 167         public void writeExternal(final ObjectOutput out) throws IOException {
 168             out.writeShort(appendEntries.leaderRaftVersion);
 169             out.writeLong(appendEntries.getTerm());
 170             out.writeObject(appendEntries.leaderId);
 171             out.writeLong(appendEntries.prevLogTerm);
 172             out.writeLong(appendEntries.prevLogIndex);
 173             out.writeLong(appendEntries.leaderCommit);
 174             out.writeLong(appendEntries.replicatedToAllIndex);
 175             out.writeShort(appendEntries.payloadVersion);
 176
 177             out.writeInt(appendEntries.entries.size());
 178             for (ReplicatedLogEntry e: appendEntries.entries) {
 179                 out.writeLong(e.getIndex());
 180                 out.writeLong(e.getTerm());
 181                 out.writeObject(e.getData());
 182             }
 183
 184             out.writeObject(appendEntries.leaderAddress);
 185         }
 186
 187         @Override
 188         public void readExternal(final ObjectInput in) throws IOException, ClassNotFoundException {
 189             short leaderRaftVersion = in.readShort();
 190             long term = in.readLong();
 191             String leaderId = (String) in.readObject();
 192             long prevLogTerm = in.readLong();
 193             long prevLogIndex = in.readLong();
 194             long leaderCommit = in.readLong();
 195             long replicatedToAllIndex = in.readLong();
 196             short payloadVersion = in.readShort();
 197
 198             int size = in.readInt();
 199             var entries = ImmutableList.<ReplicatedLogEntry>builderWithExpectedSize(size);
 200             for (int i = 0; i < size; i++) {
 201                 entries.add(new SimpleReplicatedLogEntry(in.readLong(), in.readLong(), (Payload) in.readObject()));
 202             }
 203
 204             String leaderAddress = (String)in.readObject();
 205
 206             appendEntries = new AppendEntries(term, leaderId, prevLogIndex, prevLogTerm, entries.build(), leaderCommit,
 207                     replicatedToAllIndex, payloadVersion, RaftVersions.CURRENT_VERSION, leaderRaftVersion,
 208                     leaderAddress);
 209         }
 210
 211         @java.io.Serial
 212         private Object readResolve() {
 213             return appendEntries;
 214         }
 215     }
 216 }