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 java.io.Externalizable;
  14 import java.io.IOException;
  15 import java.io.ObjectInput;
  16 import java.io.ObjectOutput;
  17 import java.util.ArrayList;
  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 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
  26
  27 /**
  28  * Invoked by leader to replicate log entries (§5.3); also used as heartbeat (§5.2).
  29  */
  30 public final class AppendEntries extends AbstractRaftRPC {
  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     private 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 Proxy(this);
 145     }
 146
 147     /**
 148      * Fluorine version that adds the leader address.
 149      */
 150     private static class ProxyV2 implements Externalizable {
 151         private static final long serialVersionUID = 1L;
 152
 153         private AppendEntries appendEntries;
 154
 155         // checkstyle flags the public modifier as redundant which really doesn't make sense since it clearly isn't
 156         // redundant. It is explicitly needed for Java serialization to be able to create instances via reflection.
 157         @SuppressWarnings("checkstyle:RedundantModifier")
 158         public ProxyV2() {
 159         }
 160
 161         ProxyV2(final AppendEntries appendEntries) {
 162             this.appendEntries = appendEntries;
 163         }
 164
 165         @Override
 166         public void writeExternal(final ObjectOutput out) throws IOException {
 167             out.writeShort(appendEntries.leaderRaftVersion);
 168             out.writeLong(appendEntries.getTerm());
 169             out.writeObject(appendEntries.leaderId);
 170             out.writeLong(appendEntries.prevLogTerm);
 171             out.writeLong(appendEntries.prevLogIndex);
 172             out.writeLong(appendEntries.leaderCommit);
 173             out.writeLong(appendEntries.replicatedToAllIndex);
 174             out.writeShort(appendEntries.payloadVersion);
 175
 176             out.writeInt(appendEntries.entries.size());
 177             for (ReplicatedLogEntry e: appendEntries.entries) {
 178                 out.writeLong(e.getIndex());
 179                 out.writeLong(e.getTerm());
 180                 out.writeObject(e.getData());
 181             }
 182
 183             out.writeObject(appendEntries.leaderAddress);
 184         }
 185
 186         @Override
 187         public void readExternal(final ObjectInput in) throws IOException, ClassNotFoundException {
 188             short leaderRaftVersion = in.readShort();
 189             long term = in.readLong();
 190             String leaderId = (String) in.readObject();
 191             long prevLogTerm = in.readLong();
 192             long prevLogIndex = in.readLong();
 193             long leaderCommit = in.readLong();
 194             long replicatedToAllIndex = in.readLong();
 195             short payloadVersion = in.readShort();
 196
 197             int size = in.readInt();
 198             List<ReplicatedLogEntry> entries = new ArrayList<>(size);
 199             for (int i = 0; i < size; i++) {
 200                 entries.add(new SimpleReplicatedLogEntry(in.readLong(), in.readLong(), (Payload) in.readObject()));
 201             }
 202
 203             String leaderAddress = (String)in.readObject();
 204
 205             appendEntries = new AppendEntries(term, leaderId, prevLogIndex, prevLogTerm, entries, leaderCommit,
 206                     replicatedToAllIndex, payloadVersion, RaftVersions.CURRENT_VERSION, leaderRaftVersion,
 207                     leaderAddress);
 208         }
 209
 210         private Object readResolve() {
 211             return appendEntries;
 212         }
 213     }
 214
 215     /**
 216      * Pre-Fluorine version.
 217      */
 218     @Deprecated
 219     private static class Proxy implements Externalizable {
 220         private static final long serialVersionUID = 1L;
 221
 222         private AppendEntries appendEntries;
 223
 224         // checkstyle flags the public modifier as redundant which really doesn't make sense since it clearly isn't
 225         // redundant. It is explicitly needed for Java serialization to be able to create instances via reflection.
 226         @SuppressWarnings("checkstyle:RedundantModifier")
 227         public Proxy() {
 228         }
 229
 230         Proxy(final AppendEntries appendEntries) {
 231             this.appendEntries = appendEntries;
 232         }
 233
 234         @Override
 235         public void writeExternal(final ObjectOutput out) throws IOException {
 236             out.writeLong(appendEntries.getTerm());
 237             out.writeObject(appendEntries.leaderId);
 238             out.writeLong(appendEntries.prevLogTerm);
 239             out.writeLong(appendEntries.prevLogIndex);
 240             out.writeLong(appendEntries.leaderCommit);
 241             out.writeLong(appendEntries.replicatedToAllIndex);
 242             out.writeShort(appendEntries.payloadVersion);
 243
 244             out.writeInt(appendEntries.entries.size());
 245             for (ReplicatedLogEntry e: appendEntries.entries) {
 246                 out.writeLong(e.getIndex());
 247                 out.writeLong(e.getTerm());
 248                 out.writeObject(e.getData());
 249             }
 250         }
 251
 252         @Override
 253         public void readExternal(final ObjectInput in) throws IOException, ClassNotFoundException {
 254             long term = in.readLong();
 255             String leaderId = (String) in.readObject();
 256             long prevLogTerm = in.readLong();
 257             long prevLogIndex = in.readLong();
 258             long leaderCommit = in.readLong();
 259             long replicatedToAllIndex = in.readLong();
 260             short payloadVersion = in.readShort();
 261
 262             int size = in.readInt();
 263             List<ReplicatedLogEntry> entries = new ArrayList<>(size);
 264             for (int i = 0; i < size; i++) {
 265                 entries.add(new SimpleReplicatedLogEntry(in.readLong(), in.readLong(), (Payload) in.readObject()));
 266             }
 267
 268             appendEntries = new AppendEntries(term, leaderId, prevLogIndex, prevLogTerm, entries, leaderCommit,
 269                 replicatedToAllIndex, payloadVersion, RaftVersions.CURRENT_VERSION, RaftVersions.BORON_VERSION, null);
 270         }
 271
 272         private Object readResolve() {
 273             return appendEntries;
 274         }
 275     }
 276 }