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
  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     private static final long serialVersionUID = 1L;
  31
  32     // So that follower can redirect clients
  33     private final @NonNull String leaderId;
  34
  35     // Index of log entry immediately preceding new ones
  36     private final long prevLogIndex;
  37
  38     // term of prevLogIndex entry
  39     private final long prevLogTerm;
  40
  41     // log entries to store (empty for heart beat - may send more than one for efficiency)
  42     private final @NonNull List<ReplicatedLogEntry> entries;
  43
  44     // leader's commitIndex
  45     private final long leaderCommit;
  46
  47     // index which has been replicated successfully to all followers, -1 if none
  48     private final long replicatedToAllIndex;
  49
  50     private final short payloadVersion;
  51
  52     private final short recipientRaftVersion;
  53
  54     private final short leaderRaftVersion;
  55
  56     private final String leaderAddress;
  57
  58     private AppendEntries(final long term, @NonNull final String leaderId, final long prevLogIndex,
  59             final long prevLogTerm, @NonNull final List<ReplicatedLogEntry> entries, final long leaderCommit,
  60             final long replicatedToAllIndex, final short payloadVersion, final short recipientRaftVersion,
  61             final short leaderRaftVersion, @Nullable final String leaderAddress) {
  62         super(term);
  63         this.leaderId = requireNonNull(leaderId);
  64         this.prevLogIndex = prevLogIndex;
  65         this.prevLogTerm = prevLogTerm;
  66         this.entries = requireNonNull(entries);
  67         this.leaderCommit = leaderCommit;
  68         this.replicatedToAllIndex = replicatedToAllIndex;
  69         this.payloadVersion = payloadVersion;
  70         this.recipientRaftVersion = recipientRaftVersion;
  71         this.leaderRaftVersion = leaderRaftVersion;
  72         this.leaderAddress = leaderAddress;
  73     }
  74
  75     public AppendEntries(final long term, final @NonNull String leaderId, final long prevLogIndex,
  76             final long prevLogTerm, final @NonNull List<ReplicatedLogEntry> entries, final long leaderCommit,
  77             final long replicatedToAllIndex, final short payloadVersion, final short recipientRaftVersion,
  78             final @Nullable String leaderAddress) {
  79         this(term, leaderId, prevLogIndex, prevLogTerm, entries, leaderCommit, replicatedToAllIndex, payloadVersion,
  80                 recipientRaftVersion, RaftVersions.CURRENT_VERSION, leaderAddress);
  81     }
  82
  83     @VisibleForTesting
  84     public AppendEntries(final long term, final @NonNull String leaderId, final long prevLogIndex,
  85             final long prevLogTerm, final @NonNull List<ReplicatedLogEntry> entries, final long leaderCommit,
  86             final long replicatedToAllIndex, final short payloadVersion) {
  87         this(term, leaderId, prevLogIndex, prevLogTerm, entries, leaderCommit, replicatedToAllIndex, payloadVersion,
  88                 RaftVersions.CURRENT_VERSION, null);
  89     }
  90
  91     public @NonNull String getLeaderId() {
  92         return leaderId;
  93     }
  94
  95     public long getPrevLogIndex() {
  96         return prevLogIndex;
  97     }
  98
  99     public long getPrevLogTerm() {
 100         return prevLogTerm;
 101     }
 102
 103     public @NonNull List<ReplicatedLogEntry> getEntries() {
 104         return entries;
 105     }
 106
 107     public long getLeaderCommit() {
 108         return leaderCommit;
 109     }
 110
 111     public long getReplicatedToAllIndex() {
 112         return replicatedToAllIndex;
 113     }
 114
 115     public short getPayloadVersion() {
 116         return payloadVersion;
 117     }
 118
 119     public Optional<String> getLeaderAddress() {
 120         return Optional.ofNullable(leaderAddress);
 121     }
 122
 123     public short getLeaderRaftVersion() {
 124         return leaderRaftVersion;
 125     }
 126
 127     @Override
 128     public String toString() {
 129         return "AppendEntries [leaderId=" + leaderId
 130                 + ", prevLogIndex=" + prevLogIndex
 131                 + ", prevLogTerm=" + prevLogTerm
 132                 + ", leaderCommit=" + leaderCommit
 133                 + ", replicatedToAllIndex=" + replicatedToAllIndex
 134                 + ", payloadVersion=" + payloadVersion
 135                 + ", recipientRaftVersion=" + recipientRaftVersion
 136                 + ", leaderRaftVersion=" + leaderRaftVersion
 137                 + ", leaderAddress=" + leaderAddress
 138                 + ", entries=" + entries + "]";
 139     }
 140
 141     @Override
 142     Object writeReplace() {
 143         return recipientRaftVersion > RaftVersions.BORON_VERSION ? new ProxyV2(this) : new Proxy(this);
 144     }
 145
 146     /**
 147      * Fluorine version that adds the leader address.
 148      */
 149     private static class ProxyV2 implements Externalizable {
 150         private static final long serialVersionUID = 1L;
 151
 152         private AppendEntries appendEntries;
 153
 154         // checkstyle flags the public modifier as redundant which really doesn't make sense since it clearly isn't
 155         // redundant. It is explicitly needed for Java serialization to be able to create instances via reflection.
 156         @SuppressWarnings("checkstyle:RedundantModifier")
 157         public ProxyV2() {
 158         }
 159
 160         ProxyV2(final AppendEntries appendEntries) {
 161             this.appendEntries = appendEntries;
 162         }
 163
 164         @Override
 165         public void writeExternal(final ObjectOutput out) throws IOException {
 166             out.writeShort(appendEntries.leaderRaftVersion);
 167             out.writeLong(appendEntries.getTerm());
 168             out.writeObject(appendEntries.leaderId);
 169             out.writeLong(appendEntries.prevLogTerm);
 170             out.writeLong(appendEntries.prevLogIndex);
 171             out.writeLong(appendEntries.leaderCommit);
 172             out.writeLong(appendEntries.replicatedToAllIndex);
 173             out.writeShort(appendEntries.payloadVersion);
 174
 175             out.writeInt(appendEntries.entries.size());
 176             for (ReplicatedLogEntry e: appendEntries.entries) {
 177                 out.writeLong(e.getIndex());
 178                 out.writeLong(e.getTerm());
 179                 out.writeObject(e.getData());
 180             }
 181
 182             out.writeObject(appendEntries.leaderAddress);
 183         }
 184
 185         @Override
 186         public void readExternal(final ObjectInput in) throws IOException, ClassNotFoundException {
 187             short leaderRaftVersion = in.readShort();
 188             long term = in.readLong();
 189             String leaderId = (String) in.readObject();
 190             long prevLogTerm = in.readLong();
 191             long prevLogIndex = in.readLong();
 192             long leaderCommit = in.readLong();
 193             long replicatedToAllIndex = in.readLong();
 194             short payloadVersion = in.readShort();
 195
 196             int size = in.readInt();
 197             List<ReplicatedLogEntry> entries = new ArrayList<>(size);
 198             for (int i = 0; i < size; i++) {
 199                 entries.add(new SimpleReplicatedLogEntry(in.readLong(), in.readLong(), (Payload) in.readObject()));
 200             }
 201
 202             String leaderAddress = (String)in.readObject();
 203
 204             appendEntries = new AppendEntries(term, leaderId, prevLogIndex, prevLogTerm, entries, leaderCommit,
 205                     replicatedToAllIndex, payloadVersion, RaftVersions.CURRENT_VERSION, leaderRaftVersion,
 206                     leaderAddress);
 207         }
 208
 209         private Object readResolve() {
 210             return appendEntries;
 211         }
 212     }
 213
 214     /**
 215      * Pre-Fluorine version.
 216      */
 217     @Deprecated
 218     private static class Proxy implements Externalizable {
 219         private static final long serialVersionUID = 1L;
 220
 221         private AppendEntries appendEntries;
 222
 223         // checkstyle flags the public modifier as redundant which really doesn't make sense since it clearly isn't
 224         // redundant. It is explicitly needed for Java serialization to be able to create instances via reflection.
 225         @SuppressWarnings("checkstyle:RedundantModifier")
 226         public Proxy() {
 227         }
 228
 229         Proxy(final AppendEntries appendEntries) {
 230             this.appendEntries = appendEntries;
 231         }
 232
 233         @Override
 234         public void writeExternal(final ObjectOutput out) throws IOException {
 235             out.writeLong(appendEntries.getTerm());
 236             out.writeObject(appendEntries.leaderId);
 237             out.writeLong(appendEntries.prevLogTerm);
 238             out.writeLong(appendEntries.prevLogIndex);
 239             out.writeLong(appendEntries.leaderCommit);
 240             out.writeLong(appendEntries.replicatedToAllIndex);
 241             out.writeShort(appendEntries.payloadVersion);
 242
 243             out.writeInt(appendEntries.entries.size());
 244             for (ReplicatedLogEntry e: appendEntries.entries) {
 245                 out.writeLong(e.getIndex());
 246                 out.writeLong(e.getTerm());
 247                 out.writeObject(e.getData());
 248             }
 249         }
 250
 251         @Override
 252         public void readExternal(final ObjectInput in) throws IOException, ClassNotFoundException {
 253             long term = in.readLong();
 254             String leaderId = (String) in.readObject();
 255             long prevLogTerm = in.readLong();
 256             long prevLogIndex = in.readLong();
 257             long leaderCommit = in.readLong();
 258             long replicatedToAllIndex = in.readLong();
 259             short payloadVersion = in.readShort();
 260
 261             int size = in.readInt();
 262             List<ReplicatedLogEntry> entries = new ArrayList<>(size);
 263             for (int i = 0; i < size; i++) {
 264                 entries.add(new SimpleReplicatedLogEntry(in.readLong(), in.readLong(), (Payload) in.readObject()));
 265             }
 266
 267             appendEntries = new AppendEntries(term, leaderId, prevLogIndex, prevLogTerm, entries, leaderCommit,
 268                 replicatedToAllIndex, payloadVersion, RaftVersions.CURRENT_VERSION, RaftVersions.BORON_VERSION, null);
 269         }
 270
 271         private Object readResolve() {
 272             return appendEntries;
 273         }
 274     }
 275 }