rep/stream/Protocol.java

/*-
 * See the file LICENSE for redistribution information.
 *
 * Copyright (c) 2002, 2014 Oracle and/or its affiliates.  All rights reserved.
 *
 */

package com.sleepycat.je.rep.stream;

import static com.sleepycat.je.rep.utilint.BinaryProtocolStatDefinition.N_ACK_MESSAGES;
import static com.sleepycat.je.rep.utilint.BinaryProtocolStatDefinition.N_GROUPED_ACKS;
import static com.sleepycat.je.rep.utilint.BinaryProtocolStatDefinition.N_GROUP_ACK_MESSAGES;
import static com.sleepycat.je.rep.utilint.BinaryProtocolStatDefinition.N_MAX_GROUPED_ACKS;

import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.UUID;

import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.Durability.SyncPolicy;
import com.sleepycat.je.EnvironmentFailureException;
import com.sleepycat.je.JEVersion;
import com.sleepycat.je.log.LogEntryType;
import com.sleepycat.je.log.LogUtils;
import com.sleepycat.je.rep.NodeType;
import com.sleepycat.je.rep.impl.RepGroupImpl;
import com.sleepycat.je.rep.impl.RepNodeImpl;
import com.sleepycat.je.rep.impl.node.NameIdPair;
import com.sleepycat.je.rep.impl.node.RepNode;
import com.sleepycat.je.rep.utilint.BinaryProtocol;
import com.sleepycat.je.rep.utilint.LongMaxZeroStat;
import com.sleepycat.je.utilint.LongMaxStat;
import com.sleepycat.je.utilint.LongStat;
import com.sleepycat.je.utilint.VLSN;


/**
 * Defines the messages used to set up a feeder-replica replication stream.
 *
 * From Feeder to Replica
 *
 *    Heartbeat -> HeartbeatResponse
 *    Commit -> Ack
 *    Commit+ -> GroupAck
 *    Entry
 *    ShutdownRequest -> ShutdownResponse
 *
 * Note: in the future, we may want to support bulk entry messages
 *
 * From Replica to Feeder
 *
 * The following subset of messages represents the handshake protocol that
 * precedes the transmission of replication log entries.
 *
 *    ReplicaProtocolVersion -> FeederProtocolVersion | DuplicateNodeReject
 *    ReplicaJEVersions -> FeederJEVersions | JEVersionsReject
 *    NodeGroupInfo -> NodeGroupInfoOK | NodeGroupInfoReject
 *    SNTPRequest -> SNTPResponse
 *    -> HeartbeatResponse
 *
 * A HeartbeatResponse is not strictly a response message and may also be sent
 * spontaneously if there is no output activity in a heartbeat interval. This
 * spontaneous generation of a HeartbeatReponse ensures that a socket is not
 * timed out if the feeder or the replica replay are otherwise busy.
 *
 * Note that there may be multiple SNTPRequest/SNTPResponse message pairs that
 * are exchanged as part of a single handshake. So a successful handshake
 * requested sequence generated by the Replica looks like:
 *
 * ReplicaProtocolVersion ReplicaJEVersions MembershipInfo [SNTPRequest]+
 *
 * The following messages constitute the syncup and the transmission of log
 * entries.
 *
 *    EntryRequest -> Entry | EntryNotFound | AlternateMatchpoint
 *    RestoreRequest -> RestoreResponse
 *    StartStream
 *
 * The Protocol instance has local state in terms of buffers that are reused
 * across multiple messages. A Protocol instance is expected to be used in
 * strictly serial fashion. Consequently, there is an instance for each Replica
 * to Feeder connection, and two instances per Feeder to Replica connection:
 * one for the InputThread and one for the OutputThread.
 */
public class Protocol extends BinaryProtocol {

    /*
     * Note that the GROUP_ACK response message was introduced in version 5,
     * but is disabled by default via RepParams.REPLICA_ENABLE_GROUP_ACKS.
     *
     * It can be enabled when we can increase the protocol version number.
     */

    /* The default (highest) version supported by the Protocol code. */
    public static final int MAX_VERSION = 5;

    /* The minimum version we're willing to interact with. */
    static final int MIN_VERSION = 3;

    /* Version added in JE 6.0.1 to support RepGroupImpl version 3. */
    public static final int VERSION_5 = 5;
    public static final JEVersion VERSION_5_JE_VERSION =
        new JEVersion("6.0.1");

    /*
     * Version in which HEARTBEAT_RESPONSE added a second field.  We can manage
     * without this optional additional information if we have to, we we can
     * still interact with the previous protocol version.  (JE 5.0.58)
     */
    static final int VERSION_4 = 4;
    public static final JEVersion VERSION_4_JE_VERSION =
        new JEVersion("5.0.58");

    /* Version added in JE 4.0.50 to address byte order issues. */
    static final int VERSION_3 = 3;
    public static final JEVersion VERSION_3_JE_VERSION =
        new JEVersion("4.0.50");

    /* The replication node that's communicating via this protocol. */
    private final RepNode repNode;

    /** The log version of the format used to write log entries. */
    private final int writeLogVersion;

    /* Count of all singleton ACK messages. */
    private final LongStat nAckMessages;

    /* Count of all group ACK messages. */
    private final LongStat nGroupAckMessages;

    /* Sum of all acks sent via group ACK messages. */
    private final LongStat nGroupedAcks;

    /* Max number of acks sent via a single group ACK message. */
    private final LongMaxStat nMaxGroupedAcks;


    /**
     * Returns a Protocol object configured that implements the specified
     * (supported) protocol version.
     *
     * @param repNode the node using the protocol
     *
     * @param protocolVersion the version of the protocol that must be
     *        implemented by this object
     *
     * @param maxProtocolVersion the highest supported protocol version, which
     *        may be lower than the code version, for testing purposes
     *
     * @param writeLogVersion the log version of the format used to write log
     *        entries
     */
    private Protocol(final RepNode repNode,
                     final int protocolVersion,
                     final int maxProtocolVersion,
                     final int writeLogVersion) {
        super((repNode != null) ? repNode.getNameIdPair() : NameIdPair.NULL,
              maxProtocolVersion,
              protocolVersion,
              (repNode != null) ? repNode.getRepImpl() : null);

        /* repNode is only null during test usage. */
        this.repNode = repNode;
        this.configuredVersion = protocolVersion;
        this.writeLogVersion = writeLogVersion;

        nAckMessages = new LongStat(stats, N_ACK_MESSAGES);
        nGroupAckMessages = new LongStat(stats, N_GROUP_ACK_MESSAGES);
        nGroupedAcks = new LongStat(stats, N_GROUPED_ACKS);
        nMaxGroupedAcks = new LongMaxZeroStat(stats, N_MAX_GROUPED_ACKS);

        initializeMessageOps(new MessageOp[] {
            REPLICA_PROTOCOL_VERSION,
            FEEDER_PROTOCOL_VERSION,
            DUP_NODE_REJECT,
            REPLICA_JE_VERSIONS,
            FEEDER_JE_VERSIONS,
            JE_VERSIONS_REJECT,
            MEMBERSHIP_INFO,
            MEMBERSHIP_INFO_OK,
            MEMBERSHIP_INFO_REJECT,
            SNTP_REQUEST,
            SNTP_RESPONSE,
            ENTRY,
            START_STREAM,
            HEARTBEAT,
            HEARTBEAT_RESPONSE,
            COMMIT,
            ACK,
            ENTRY_REQUEST,
            ENTRY_NOTFOUND,
            RESTORE_REQUEST,
            RESTORE_RESPONSE,
            ALT_MATCHPOINT,
            SHUTDOWN_REQUEST,
            SHUTDOWN_RESPONSE,
            GROUP_ACK
        });
    }

    /**
     * Returns a protocol object that supports the specific requested protocol
     * version, or null if the version is not supported.
     */
    public static Protocol get(final RepNode repNode,
                               final int protocolVersion) {
        return get(repNode, protocolVersion, protocolVersion);
    }

    /**
     * Returns a protocol object that supports the specific requested protocol
     * version, which must not be higher than the specified maximum version, or
     * null if no such version is supported.
     */
    public static Protocol get(final RepNode repNode,
                               final int protocolVersion,
                               final int maxProtocolVersion) {
        return get(repNode, protocolVersion, maxProtocolVersion,
                   LogEntryType.LOG_VERSION);
    }

    /**
     * Returns a protocol object that supports the specified protocol, which
     * must be less than the specified maximum version, and writes log entries
     * in the specified log version format.  Returns null if no such version is
     * supported.
     */
    public static Protocol get(final RepNode repNode,
                               final int protocolVersion,
                               final int maxProtocolVersion,
                               final int writeLogVersion) {
        assert repNode != null;

        /*
         * If the RepGroupImpl has been upgraded to version 3, then require
         * protocol version 5, which is required to support that RepGroupImpl
         * version.  This check prevents new facilities that depend on
         * RepGroupImpl version 3 from being seen by non-upgraded replicas.
         */
        int minProtocolVersion = MIN_VERSION;
        final RepGroupImpl group = repNode.getGroup();
        if (group == null) {
            throw EnvironmentFailureException.unexpectedState(
                "Group is null");
        }
        final int groupFormatVersion = group.getFormatVersion();
        if (groupFormatVersion >= RepGroupImpl.FORMAT_VERSION_3) {
            minProtocolVersion = VERSION_5;
        }

        return get(repNode, protocolVersion, minProtocolVersion,
                   maxProtocolVersion, writeLogVersion);
    }

    /**
     * Returns a protocol object using the specified minimum and maximum
     * values, returning null if no supported version is found.  Use this
     * method for testing when the RepGroupImpl object is not available.
     */
    static Protocol get(final RepNode repNode,
                        final int protocolVersion,
                        final int minProtocolVersion,
                        final int maxProtocolVersion,
                        final int writeLogVersion) {

        if (!isSupportedVersion(protocolVersion, minProtocolVersion,
                                maxProtocolVersion)) {
            return null;
        }

        /*
         * Future code will do what is appropriate in support of the version
         * depending on the nature of the incompatibility.
         */
        return new Protocol(repNode, protocolVersion, maxProtocolVersion,
                            writeLogVersion);
    }

    /**
     * Returns a protocol object using the specified protocol version.
     */
    static Protocol getProtocol(final RepNode repNode,
                                final int protocolVersion) {

        assert repNode != null;
        return new Protocol(repNode, protocolVersion, protocolVersion,
                            LogEntryType.LOG_VERSION);
    }

    /**
     * Returns true if the code can support the version.
     *
     * @param protocolVersion protocol version being queried
     * @param minProtocolVersion minimum protocol version supported
     * @param maxProtocolVersion maximum protocol version supported
     *
     * @return true if the protocol version is supported by this implementation
     *         of the protocol
     */
    private static boolean isSupportedVersion(final int protocolVersion,
                                              final int minProtocolVersion,
                                              final int maxProtocolVersion) {
        if (protocolVersion == Integer.MIN_VALUE) {
            /* For testing purposes. */
            return false;
        }

        /*
         * Version compatibility check: for now, a simple range check.  We can
         * make this fancier in the future if necessary.
         */
        return minProtocolVersion <= protocolVersion &&
            protocolVersion <= maxProtocolVersion;
    }

    /**
     * Gets the JE version that corresponds to the specified protocol version,
     * for use in creating error messages that explain protocol version errors
     * in terms of JE versions.  Returns null if the associated version is not
     * known.
     */
    static JEVersion getProtocolJEVersion(final int protocolVersion) {
        switch (protocolVersion) {
        case VERSION_5:
            return VERSION_5_JE_VERSION;
        case VERSION_4:
            return VERSION_4_JE_VERSION;
        case VERSION_3:
            return VERSION_3_JE_VERSION;
        default:
            return null;
        }
    }

    /**
     * Gets the protocol version that corresponds to the specified JE version,
     * throwing an IllegalArgumentException if the version is not supported.
     */
    static int getJEVersionProtocolVersion(final JEVersion jeVersion) {
        if (jeVersion == null) {
            return VERSION_5;
        } else if (jeVersion.compareTo(VERSION_5_JE_VERSION) >= 0) {
            return VERSION_5;
        } else if (jeVersion.compareTo(VERSION_4_JE_VERSION) >= 0) {
            return VERSION_4;
        } else if (jeVersion.compareTo(VERSION_3_JE_VERSION) >= 0) {
            return VERSION_3;
        } else {
            throw new IllegalArgumentException(
                "JE version not supported: " + jeVersion);
        }
    }

    /**
     * Write an entry output wire record to the message buffer using the write
     * log version format and increment nEntriesWrittenOldVersion if the entry
     * format was changed.
     */
    void writeOutputWireRecord(final OutputWireRecord record,
                               final ByteBuffer messageBuffer) {
        final boolean changedFormat =
            record.writeToWire(messageBuffer, writeLogVersion);
        if (changedFormat) {
            nEntriesWrittenOldVersion.increment();
        }
    }

    public final static MessageOp REPLICA_PROTOCOL_VERSION =
        new MessageOp((short) 1, ReplicaProtocolVersion.class);

    public final static MessageOp FEEDER_PROTOCOL_VERSION =
        new MessageOp((short) 2, FeederProtocolVersion.class);

    public final static MessageOp DUP_NODE_REJECT =
        new MessageOp((short) 3, DuplicateNodeReject.class);

    public final static MessageOp REPLICA_JE_VERSIONS =
        new MessageOp((short) 4, ReplicaJEVersions.class);

    public final static MessageOp FEEDER_JE_VERSIONS =
        new MessageOp((short) 5, FeederJEVersions.class);

    public final static MessageOp JE_VERSIONS_REJECT =
        new MessageOp((short) 6, JEVersionsReject.class);

    public final static MessageOp MEMBERSHIP_INFO =
        new MessageOp((short) 7, NodeGroupInfo.class);

    public final static MessageOp MEMBERSHIP_INFO_OK =
        new MessageOp((short) 8, NodeGroupInfoOK.class);

    public final static MessageOp MEMBERSHIP_INFO_REJECT =
        new MessageOp((short) 9, NodeGroupInfoReject.class);

    public final static MessageOp SNTP_REQUEST =
        new MessageOp((short)10, SNTPRequest.class);

    public final static MessageOp SNTP_RESPONSE =
        new MessageOp((short)11, SNTPResponse.class);

        /* Core Replication Stream post-handshake messages */
    public final static MessageOp ENTRY =
        new MessageOp((short) 101, Entry.class);

    public final static MessageOp START_STREAM =
        new MessageOp((short) 102, StartStream.class);

    public final static MessageOp HEARTBEAT =
        new MessageOp((short) 103, Heartbeat.class);

    public final static MessageOp HEARTBEAT_RESPONSE =
        new MessageOp((short) 104, HeartbeatResponse.class);

    public final static MessageOp COMMIT =
        new MessageOp((short) 105, Commit.class);

    public final static MessageOp ACK =
        new MessageOp((short) 106, Ack.class);

    public final static MessageOp ENTRY_REQUEST =
        new MessageOp((short) 107, EntryRequest.class);

    public final static MessageOp ENTRY_NOTFOUND =
        new MessageOp((short) 108, EntryNotFound.class);

    public final static MessageOp ALT_MATCHPOINT =
        new MessageOp((short) 109, AlternateMatchpoint.class);

    public final static MessageOp RESTORE_REQUEST =
        new MessageOp((short) 110, RestoreRequest.class);

    public final static MessageOp RESTORE_RESPONSE =
        new MessageOp((short) 111, RestoreResponse.class);

    public final static MessageOp SHUTDOWN_REQUEST =
        new MessageOp((short) 112, ShutdownRequest.class);

    public final static MessageOp SHUTDOWN_RESPONSE =
        new MessageOp((short) 113, ShutdownResponse.class);

    public final static MessageOp GROUP_ACK =
        new MessageOp((short) 114, GroupAck.class);

    /**
     * Base class for all protocol handshake messages.
     */
    abstract class HandshakeMessage extends SimpleMessage {
    }

    /**
     * Version broadcasts the sending node's protocol version.
     */
    abstract class ProtocolVersion extends HandshakeMessage {
        private final int version;

        @SuppressWarnings("hiding")
        private final NameIdPair nameIdPair;

        public ProtocolVersion(int version) {
            super();
            this.version = version;
            this.nameIdPair = Protocol.this.nameIdPair;
        }

        @Override
        public ByteBuffer wireFormat() {
            return wireFormat(version, nameIdPair);
        }

        public ProtocolVersion(ByteBuffer buffer) {
            version = LogUtils.readInt(buffer);
            nameIdPair = getNameIdPair(buffer);
        }

        /**
         * @return the version
         */
        int getVersion() {
            return version;
        }

        /**
         * The nodeName of the sender
         *
         * @return nodeName
         */
        NameIdPair getNameIdPair() {
            return nameIdPair;
        }
    }

    /**
     * The replica sends the feeder its protocol version.
     *
     * IMPORTANT: This message must not change.
     */
    public class ReplicaProtocolVersion extends ProtocolVersion {

        public ReplicaProtocolVersion() {
            super(configuredVersion);
        }

        public ReplicaProtocolVersion(ByteBuffer buffer) {
            super(buffer);
        }

        @Override
        public MessageOp getOp() {
            return REPLICA_PROTOCOL_VERSION;
        }
    }

    /**
     * The feeder sends the replica its proposed version.
     *
     * IMPORTANT: This message must not change.
     */
    public class FeederProtocolVersion extends ProtocolVersion {

        public FeederProtocolVersion(int proposedVersion) {
            super(proposedVersion);
        }

        public FeederProtocolVersion(ByteBuffer buffer) {
            super(buffer);
        }

        @Override
        public MessageOp getOp() {
            return FEEDER_PROTOCOL_VERSION;
        }
    }

    /* Reject response to a ReplicaProtocolVersion request */
    public class DuplicateNodeReject extends RejectMessage {

        DuplicateNodeReject(String errorMessage) {
            super(errorMessage);
        }

        public DuplicateNodeReject(ByteBuffer buffer) {
            super(buffer);
        }

        @Override
        public MessageOp getOp() {
            return DUP_NODE_REJECT;
        }
    }

    public class SNTPRequest extends HandshakeMessage {

        private final long originateTimestamp;

        /* Set by the receiver at the time the message is recreated. */
        private long receiveTimestamp = -1;

        /*
         * Determines whether this is the last in a consecutive stream of
         * requests to determine the skew.
         */
        private boolean isLast = true;

        public SNTPRequest(boolean isLast) {
            super();
            this.isLast = isLast;
            originateTimestamp = repNode.getClock().currentTimeMillis();
        }

        @Override
        public ByteBuffer wireFormat() {
            return wireFormat(originateTimestamp, isLast);
        }

        public SNTPRequest(ByteBuffer buffer) {
            this.originateTimestamp = LogUtils.readLong(buffer);
            this.isLast = getBoolean(buffer);
            this.receiveTimestamp = repNode.getClock().currentTimeMillis();
        }

        @Override
        public MessageOp getOp() {
            return SNTP_REQUEST;
        }

        public long getOriginateTimestamp() {
            return originateTimestamp;
        }

        public long getReceiveTimestamp() {
            return receiveTimestamp;
        }

        public boolean isLast() {
            return isLast;
        }
    }

    public class SNTPResponse extends HandshakeMessage {

        /* These fields have the standard SNTP interpretation */
        private final long originateTimestamp; // time request sent by client
        private final long receiveTimestamp; // time request received by server

        /*
         * Initialized when the message is serialized to ensure it's as
         * accurate as possible.
         */
        private long transmitTimestamp = -1; // time reply sent by server

        /* Initialized at de-serialization for similar reasons. */
        private long destinationTimestamp = -1; //time reply received by client

        public SNTPResponse(SNTPRequest request) {
            this.originateTimestamp = request.originateTimestamp;
            this.receiveTimestamp = request.receiveTimestamp;
        }

        @Override
        public ByteBuffer wireFormat() {
            transmitTimestamp = repNode.getClock().currentTimeMillis();
            return wireFormat(originateTimestamp,
                              receiveTimestamp,
                              transmitTimestamp);
        }

        public SNTPResponse(ByteBuffer buffer) {
            originateTimestamp = LogUtils.readLong(buffer);
            receiveTimestamp = LogUtils.readLong(buffer);
            transmitTimestamp = LogUtils.readLong(buffer);
            destinationTimestamp = repNode.getClock().currentTimeMillis();
        }

        @Override
        public MessageOp getOp() {
            return SNTP_RESPONSE;
        }

        public long getOriginateTimestamp() {
            return originateTimestamp;
        }

        public long getReceiveTimestamp() {
            return receiveTimestamp;
        }

        public long getTransmitTimestamp() {
            return transmitTimestamp;
        }

        public long getDestinationTimestamp() {
            return destinationTimestamp;
        }

        public long getDelay() {
            assert(destinationTimestamp != -1);
            return (destinationTimestamp - originateTimestamp) -
                    (transmitTimestamp - receiveTimestamp);
        }

        public long getDelta() {
            assert(destinationTimestamp != -1);
            return ((receiveTimestamp - originateTimestamp) +
                    (transmitTimestamp - destinationTimestamp))/2;
        }
    }

    /**
     * Abstract message used as the basis for the exchange of software versions
     * between replicated nodes
     */
    abstract class JEVersions extends HandshakeMessage {
        private final JEVersion version;

        private final int logVersion;

        public JEVersions(JEVersion version, int logVersion) {
            this.version = version;
            this.logVersion = logVersion;
        }

        @Override
        public ByteBuffer wireFormat() {
            return wireFormat(version.getVersionString(), logVersion);
        }

        public JEVersions(ByteBuffer buffer) {
            this.version = new JEVersion(getString(buffer));
            this.logVersion = LogUtils.readInt(buffer);
        }

        public JEVersion getVersion() {
            return version;
        }

        public byte getLogVersion() {
            return (byte)logVersion;
        }
    }

    public class ReplicaJEVersions extends JEVersions {

        ReplicaJEVersions(JEVersion version, int logVersion) {
            super(version, logVersion);
        }

        public ReplicaJEVersions(ByteBuffer buffer) {
            super(buffer);
        }

        @Override
        public MessageOp getOp() {
            return REPLICA_JE_VERSIONS;
        }

    }

    public class FeederJEVersions extends JEVersions {

        FeederJEVersions(JEVersion version, int logVersion) {
            super(version, logVersion);
        }

        public FeederJEVersions(ByteBuffer buffer) {
            super(buffer);
        }

        @Override
        public MessageOp getOp() {
            return FEEDER_JE_VERSIONS;
        }
    }

    /* Reject response to a ReplicaJEVersions request */
    public class JEVersionsReject extends RejectMessage {

        public JEVersionsReject(String errorMessage) {
            super(errorMessage);
        }

        public JEVersionsReject(ByteBuffer buffer) {
            super(buffer);
        }

        @Override
        public MessageOp getOp() {
            return JE_VERSIONS_REJECT;
        }
    }

    public class NodeGroupInfo extends HandshakeMessage {
        private final String groupName;
        private final UUID uuid;

        @SuppressWarnings("hiding")
        private final NameIdPair nameIdPair;
        private final String hostName;
        private final int port;
        private final NodeType nodeType;
        private final boolean designatedPrimary;

        /**
         * A string version of the JE version running on this node, or the
         * empty string if not known.
         */
        private final String jeVersion;

        NodeGroupInfo(final String groupName,
                      final UUID uuid,
                      final NameIdPair nameIdPair,
                      final String hostName,
                      final int port,
                      final NodeType nodeType,
                      final boolean designatedPrimary,
                      final JEVersion jeVersion) {

            this.groupName = groupName;
            this.uuid = uuid;
            this.nameIdPair = nameIdPair;
            this.hostName = hostName;
            this.port = port;
            this.nodeType = nodeType;
            this.designatedPrimary = designatedPrimary;
            this.jeVersion = (jeVersion != null) ?
                jeVersion.getNumericVersionString() :
                "";
        }

        @Override
        public MessageOp getOp() {
            return MEMBERSHIP_INFO;
        }

        @Override
        public ByteBuffer wireFormat() {
            final boolean repGroupV3 = (getVersion() >= VERSION_5);
            if (!repGroupV3 && nodeType.compareTo(NodeType.ELECTABLE) > 0) {
                throw new IllegalStateException(
                    "Node type not supported before group version 3: " +
                    nodeType);
            }
            final Object[] args = new Object[repGroupV3 ? 9 : 8];
            args[0] = groupName;
            args[1] = uuid.getMostSignificantBits();
            args[2] = uuid.getLeastSignificantBits();
            args[3] = nameIdPair;
            args[4] = hostName;
            args[5] = port;
            args[6] = nodeType;
            args[7] = designatedPrimary;
            if (repGroupV3) {
                args[8] = jeVersion;
            }
            return wireFormat(args);
        }

        public NodeGroupInfo(ByteBuffer buffer) {
            this.groupName = getString(buffer);
            this.uuid = new UUID(LogUtils.readLong(buffer),
                                 LogUtils.readLong(buffer));
            this.nameIdPair = getNameIdPair(buffer);
            this.hostName = getString(buffer);
            this.port = LogUtils.readInt(buffer);
            this.nodeType = getEnum(NodeType.class, buffer);
            this.designatedPrimary = getBoolean(buffer);
            jeVersion = (getVersion() >= VERSION_5) ? getString(buffer) : "";
        }

        public String getGroupName() {
            return groupName;
        }

        public UUID getUUID() {
            return uuid;
        }

        public String getNodeName() {
            return nameIdPair.getName();
        }

        public int getNodeId() {
            return nameIdPair.getId();
        }

        public String getHostName() {
            return hostName;
        }

        public NameIdPair getNameIdPair() {
            return nameIdPair;
        }

        public int port() {
            return port;
        }
        public NodeType getNodeType() {
            return nodeType;
        }

        public boolean isDesignatedPrimary() {
            return designatedPrimary;
        }

        /**
         * Returns the JE version most recently noted running on the associated
         * node, or null if not known.
         */
        public JEVersion getJEVersion() {
            return !jeVersion.isEmpty() ? new JEVersion(jeVersion) : null;
        }
    }

    /**
     * Response to a NodeGroupInfo request that was successful.  The object
     * contains the group's UUID and the replica's NameIdPair.  The group UUID
     * is used to update the replica's notion of the group UUID on first
     * joining.  The NameIdPair is used to update the replica's node ID for a
     * secondary node, which is not available in the RepGroupDB.
     */
    public class NodeGroupInfoOK extends HandshakeMessage {

        private final UUID uuid;
        @SuppressWarnings("hiding")
        private final NameIdPair nameIdPair;

        public NodeGroupInfoOK(UUID uuid, NameIdPair nameIdPair) {
            super();
            this.uuid = uuid;
            this.nameIdPair = nameIdPair;
        }

        public NodeGroupInfoOK(ByteBuffer buffer) {
            uuid = new UUID(LogUtils.readLong(buffer),
                            LogUtils.readLong(buffer));
            nameIdPair = getNameIdPair(buffer);
        }

        @Override
        public ByteBuffer wireFormat() {
            return wireFormat(uuid.getMostSignificantBits(),
                              uuid.getLeastSignificantBits(),
                              nameIdPair);
        }

        @Override
        public MessageOp getOp() {
            return MEMBERSHIP_INFO_OK;
        }

        public NameIdPair getNameIdPair() {
            return nameIdPair;
        }

        public UUID getUUID() {
            return uuid;
        }
    }

    public class NodeGroupInfoReject extends RejectMessage {

        NodeGroupInfoReject(String errorMessage) {
            super(errorMessage);
        }

        @Override
        public MessageOp getOp() {
            return MEMBERSHIP_INFO_REJECT;
        }

        @Override
        public ByteBuffer wireFormat() {
            return wireFormat(errorMessage);
        }

        public NodeGroupInfoReject(ByteBuffer buffer) {
            super(buffer);
        }
    }

    /**
     * Base class for messages which contain only a VLSN
     */
    abstract class VLSNMessage extends Message {
        protected final VLSN vlsn;

        VLSNMessage(VLSN vlsn) {
            super();
            this.vlsn = vlsn;
        }

        public VLSNMessage(ByteBuffer buffer) {
            long vlsnSequence = LogUtils.readLong(buffer);
            vlsn = new VLSN(vlsnSequence);
        }

        @Override
        public ByteBuffer wireFormat() {
            int bodySize = 8;
            ByteBuffer messageBuffer = allocateInitializedBuffer(bodySize);
            LogUtils.writeLong(messageBuffer, vlsn.getSequence());
            messageBuffer.flip();
            return messageBuffer;
        }

        VLSN getVLSN() {
            return vlsn;
        }

        @Override
        public String toString() {
            return super.toString() + " " + vlsn;
        }
    }

    /**
     * A message containing a log entry in the replication stream.
     */
    public class Entry extends Message {

        /*
         * InputWireRecord is set when this Message had been received at this
         * node. OutputWireRecord is set when this message is created for
         * sending from this node.
         */
        final protected InputWireRecord inputWireRecord;
        protected OutputWireRecord outputWireRecord;

        public Entry(final OutputWireRecord outputWireRecord) {
            inputWireRecord = null;
            this.outputWireRecord = outputWireRecord;
        }

        @Override
        public MessageOp getOp() {
            return ENTRY;
        }

        @Override
        public ByteBuffer wireFormat() {
            final int bodySize = getWireSize();
            final ByteBuffer messageBuffer =
                allocateInitializedBuffer(bodySize);
            writeOutputWireRecord(outputWireRecord, messageBuffer);
            messageBuffer.flip();
            return messageBuffer;
        }

        protected int getWireSize() {
            return outputWireRecord.getWireSize(writeLogVersion);
        }

        public Entry(final ByteBuffer buffer)
            throws DatabaseException {

            inputWireRecord =
                new InputWireRecord(repNode.getRepImpl(), buffer);
        }

        public InputWireRecord getWireRecord() {
            return inputWireRecord;
        }

        @Override
        public String toString() {

            final StringBuilder sb = new StringBuilder();
            sb.append(super.toString());

            if (inputWireRecord != null) {
                sb.append(" ");
                sb.append(inputWireRecord);
            }

            if (outputWireRecord != null) {
                sb.append(" ");
                sb.append(outputWireRecord);
            }

            return sb.toString();
        }

        /* For unit test support */
        @Override
        public boolean match(Message other) {

            /*
             * This message was read in, but we need to compare it to a message
             * that was sent out.
             */
            if (outputWireRecord == null) {
                outputWireRecord = new OutputWireRecord(repNode.getRepImpl(),
                                                        inputWireRecord);
            }
            return super.match(other);
        }

        /* True if the log entry is a TxnAbort or TxnCommit. */
        public boolean isTxnEnd() {
            final byte entryType = getWireRecord().getEntryType();
            if (LogEntryType.LOG_TXN_COMMIT.equalsType(entryType) ||
                LogEntryType.LOG_TXN_ABORT.equalsType(entryType)) {
                return true;
            }

            return false;
        }
    }

    /**
     * StartStream indicates that the replica would like the feeder to start
     * the replication stream at the proposed vlsn.
     */
    public class StartStream extends VLSNMessage {

        StartStream(VLSN startVLSN) {
            super(startVLSN);
        }

        public StartStream(ByteBuffer buffer) {
            super(buffer);
        }

        @Override
        public MessageOp getOp() {
            return START_STREAM;
        }
    }

    public class Heartbeat extends Message {

        private final long masterNow;
        private final long currentTxnEndVLSN;

        public Heartbeat(long masterNow, long currentTxnEndVLSN) {
            this.masterNow = masterNow;
            this.currentTxnEndVLSN = currentTxnEndVLSN;
        }

        @Override
        public MessageOp getOp() {
            return HEARTBEAT;
        }

        @Override
        public ByteBuffer wireFormat() {
            int bodySize = 8 * 2 /* masterNow + currentTxnEndVLSN */;
            ByteBuffer messageBuffer = allocateInitializedBuffer(bodySize);
            LogUtils.writeLong(messageBuffer, masterNow);
            LogUtils.writeLong(messageBuffer, currentTxnEndVLSN);
            messageBuffer.flip();
            return messageBuffer;
        }

        public Heartbeat(ByteBuffer buffer) {
            masterNow = LogUtils.readLong(buffer);
            currentTxnEndVLSN = LogUtils.readLong(buffer);
        }

        public long getMasterNow() {
            return masterNow;
        }

        public long getCurrentTxnEndVLSN() {
            return currentTxnEndVLSN;
        }

        @Override
        public String toString() {
            return super.toString() + " masterNow=" + masterNow +
                " currentCommit=" + currentTxnEndVLSN;
        }
    }

    public class HeartbeatResponse extends Message {
        /* The latest syncupVLSN */
        private final VLSN syncupVLSN;
        private final VLSN txnEndVLSN;

        public HeartbeatResponse(VLSN syncupVLSN, VLSN ackedVLSN) {
            super();
            this.syncupVLSN = syncupVLSN;
            this.txnEndVLSN = ackedVLSN;
        }

        public HeartbeatResponse(ByteBuffer buffer) {
            syncupVLSN = new VLSN(LogUtils.readLong(buffer));
            txnEndVLSN =
                getVersion() >= VERSION_4 ?
                new VLSN(LogUtils.readLong(buffer)) :
                null;
        }

        @Override
        public MessageOp getOp() {
            return HEARTBEAT_RESPONSE;
        }

        @Override
        public ByteBuffer wireFormat() {
            boolean includeTxnEndVLSN = getVersion() >= VERSION_4;
            int bodySize = includeTxnEndVLSN ?
                           8 * 2 :
                           8;
            ByteBuffer messageBuffer = allocateInitializedBuffer(bodySize);
            LogUtils.writeLong(messageBuffer, syncupVLSN.getSequence());
            if (includeTxnEndVLSN) {
                LogUtils.writeLong(messageBuffer, txnEndVLSN.getSequence());
            }
            messageBuffer.flip();
            return messageBuffer;
        }

        public VLSN getSyncupVLSN() {
            return syncupVLSN;
        }

        public VLSN getTxnEndVLSN() {
            return txnEndVLSN;
        }

        @Override
        public String toString() {
            return super.toString() + " syncupVLSN=" + syncupVLSN;
        }
    }

    /**
     * Message used to shutdown a node
     */
    public class ShutdownRequest extends SimpleMessage {
        /* The time that the shutdown was initiated on the master. */
        private final long shutdownTimeMs;

        public ShutdownRequest(long shutdownTimeMs) {
            super();
            this.shutdownTimeMs = shutdownTimeMs;
        }

        @Override
        public MessageOp getOp() {
            return SHUTDOWN_REQUEST;
        }

        public ShutdownRequest(ByteBuffer buffer) {
            shutdownTimeMs = LogUtils.readLong(buffer);
        }

        @Override
        public ByteBuffer wireFormat() {
            return wireFormat(shutdownTimeMs);
        }

        public long getShutdownTimeMs() {
            return shutdownTimeMs;
        }
    }

    /**
     * Message in response to a shutdown request.
     */
    public class ShutdownResponse extends Message {

        public ShutdownResponse() {
            super();
        }

        @Override
        public MessageOp getOp() {
            return SHUTDOWN_RESPONSE;
        }

        public ShutdownResponse(@SuppressWarnings("unused") ByteBuffer buffer) {
        }
    }

    public class Commit extends Entry {
        private final boolean needsAck;
        private final SyncPolicy replicaSyncPolicy;

        public Commit(final boolean needsAck,
                      final SyncPolicy replicaSyncPolicy,
                      final OutputWireRecord wireRecord) {
            super(wireRecord);
            this.needsAck = needsAck;
            this.replicaSyncPolicy = replicaSyncPolicy;
        }

        @Override
        public MessageOp getOp() {
            return COMMIT;
        }

        @Override
        public ByteBuffer wireFormat() {
            final int bodySize = super.getWireSize() +
                1 /* needsAck */ +
                1 /* replica sync policy */;
            final ByteBuffer messageBuffer =
                allocateInitializedBuffer(bodySize);
            messageBuffer.put((byte) (needsAck ? 1 : 0));
            messageBuffer.put((byte) replicaSyncPolicy.ordinal());
            writeOutputWireRecord(outputWireRecord, messageBuffer);
            messageBuffer.flip();
            return messageBuffer;
        }

        public Commit(final ByteBuffer buffer)
            throws DatabaseException {

            this(getByteNeedsAck(buffer.get()),
                 getByteReplicaSyncPolicy(buffer.get()),
                 buffer);
        }

        private Commit(final boolean needsAck,
                       final SyncPolicy replicaSyncPolicy,
                       final ByteBuffer buffer)
            throws DatabaseException {

            super(buffer);
            this.needsAck = needsAck;
            this.replicaSyncPolicy = replicaSyncPolicy;
        }

        public boolean getNeedsAck() {
            return needsAck;
        }

        public SyncPolicy getReplicaSyncPolicy() {
            return replicaSyncPolicy;
        }
    }

    /**
     * Returns whether the byte value specifies that an acknowledgment is
     * needed.
     */
    private static boolean getByteNeedsAck(final byte needsAckByte) {
        switch (needsAckByte) {
        case 0:
            return false;
        case 1:
            return true;
        default:
            throw EnvironmentFailureException.unexpectedState(
                "Invalid bool ordinal: " + needsAckByte);
        }
    }

    /** Returns the sync policy specified by the argument. */
    private static SyncPolicy getByteReplicaSyncPolicy(
        final byte syncPolicyByte) {

        for (final SyncPolicy p : SyncPolicy.values()) {
            if (p.ordinal() == syncPolicyByte) {
                return p;
            }
        }
        throw EnvironmentFailureException.unexpectedState(
            "Invalid sync policy ordinal: " + syncPolicyByte);
    }

    public class Ack extends Message {

        private final long txnId;

        public Ack(long txnId) {
            super();
            this.txnId = txnId;
            nAckMessages.increment();
        }

        @Override
        public MessageOp getOp() {
            return ACK;
        }

        @Override
        public ByteBuffer wireFormat() {
            int bodySize = 8;
            ByteBuffer messageBuffer = allocateInitializedBuffer(bodySize);
            LogUtils.writeLong(messageBuffer, txnId);
            messageBuffer.flip();
            return messageBuffer;
        }

        public Ack(ByteBuffer buffer) {
            txnId = LogUtils.readLong(buffer);
        }

        public long getTxnId() {
            return txnId;
        }

        @Override
        public String toString() {
            return super.toString() + " txn " + txnId;
        }
    }

    public class GroupAck extends Message {

        private final long txnIds[];

        public GroupAck(long txnIds[]) {
            super();
            this.txnIds = txnIds;
            nGroupAckMessages.increment();
            nGroupedAcks.add(txnIds.length);
            nMaxGroupedAcks.setMax(txnIds.length);
        }

        @Override
        public MessageOp getOp() {
            return GROUP_ACK;
        }

        @Override
        public ByteBuffer wireFormat() {

            final int bodySize = 4 + 8 *  txnIds.length;
            final ByteBuffer messageBuffer =
                allocateInitializedBuffer(bodySize);

            putLongArray(messageBuffer, txnIds);
            messageBuffer.flip();

            return messageBuffer;
        }

        public GroupAck(ByteBuffer buffer) {
            txnIds = readLongArray(buffer);
        }

        public long[] getTxnIds() {
            return txnIds;
        }

        @Override
        public String toString() {
            return super.toString() + " txn " + Arrays.toString(txnIds);
        }
    }


    private void putLongArray(ByteBuffer buffer, long[] la) {
        LogUtils.writeInt(buffer, la.length);

        for (long l : la) {
            LogUtils.writeLong(buffer, l);
        }
    }

    private long[] readLongArray(ByteBuffer buffer) {
        final long la[] = new long[LogUtils.readInt(buffer)];

        for (int i=0;  i < la.length; i++) {
            la[i] = LogUtils.readLong(buffer);
        }

        return la;
    }

    /**
     * A replica node asks a feeder for the log entry at this VLSN.
     */
    public class EntryRequest extends VLSNMessage {

        EntryRequest(VLSN matchpoint) {
            super(matchpoint);
        }

        public EntryRequest(ByteBuffer buffer) {
            super(buffer);
        }

        @Override
        public MessageOp getOp() {
            return ENTRY_REQUEST;
        }
    }

    /**
     * Response when the EntryRequest asks for a VLSN that is below the VLSN
     * range covered by the Feeder.
     */
    public class EntryNotFound extends Message {

        public EntryNotFound() {
        }

        public EntryNotFound(@SuppressWarnings("unused") ByteBuffer buffer) {
            super();
        }

        @Override
        public MessageOp getOp() {
            return ENTRY_NOTFOUND;
        }
    }

    public class AlternateMatchpoint extends Message {

        private final InputWireRecord alternateInput;
        private OutputWireRecord alternateOutput = null;

        AlternateMatchpoint(final OutputWireRecord alternate) {
            alternateInput = null;
            this.alternateOutput = alternate;
        }

        @Override
        public MessageOp getOp() {
            return ALT_MATCHPOINT;
        }

        @Override
        public ByteBuffer wireFormat() {
            final int bodySize = alternateOutput.getWireSize(writeLogVersion);
            final ByteBuffer messageBuffer =
                allocateInitializedBuffer(bodySize);
            writeOutputWireRecord(alternateOutput, messageBuffer);
            messageBuffer.flip();
            return messageBuffer;
        }

        public AlternateMatchpoint(final ByteBuffer buffer)
            throws DatabaseException {
            alternateInput = new InputWireRecord(repNode.getRepImpl(), buffer);
        }

        public InputWireRecord getAlternateWireRecord() {
            return alternateInput;
        }

        /* For unit test support */
        @Override
        public boolean match(Message other) {

            /*
             * This message was read in, but we need to compare it to a message
             * that was sent out.
             */
            if (alternateOutput == null) {
                alternateOutput =
                    new OutputWireRecord(repNode.getRepImpl(), alternateInput);
            }
            return super.match(other);
        }
    }

    /**
     * Request from the replica to the feeder for sufficient information to
     * start a network restore.
     */
    public class RestoreRequest extends VLSNMessage {

        RestoreRequest(VLSN failedMatchpoint) {
            super(failedMatchpoint);
        }

        public RestoreRequest(ByteBuffer buffer) {
            super(buffer);
        }

        @Override
        public MessageOp getOp() {
            return RESTORE_REQUEST;
        }
    }

    /**
     * Response when the replica needs information to instigate a network
     * restore. The message contains two pieces of information. One is a set of
     * nodes that could be used as the basis for a NetworkBackup so that the
     * request node can become current again. The second is a suitable low vlsn
     * for the replica, which will be registered as this replica's local
     * CBVLSN. This will contribute to the global CBVLSN calculation.
     *
     * The feeder when sending this response must, if it is also the master,
     * update the membership table to set the local CBVLSN for the requesting
     * node thus ensuring that it can continue the replication stream at this
     * VLSN (or higher) when it retries the syncup operation.
     */
    public class RestoreResponse extends SimpleMessage {
        private final VLSN cbvlsn;

        private final RepNodeImpl[] logProviders;

        public RestoreResponse(VLSN cbvlsn, RepNodeImpl[] logProviders) {
            this.cbvlsn = cbvlsn;
            this.logProviders = logProviders;
        }

        public RestoreResponse(ByteBuffer buffer) {
            long vlsnSequence = LogUtils.readLong(buffer);
            cbvlsn = new VLSN(vlsnSequence);
            logProviders = getRepNodeImplArray(buffer);
        }

        @Override
        public ByteBuffer wireFormat() {
            return wireFormat(cbvlsn.getSequence(), logProviders);
        }

        /* Add support for RepNodeImpl arrays. */

        @Override
        protected void putWireFormat(final ByteBuffer buffer,
                                     final Object obj) {
            if (obj.getClass() == RepNodeImpl[].class) {
                putRepNodeImplArray(buffer, (RepNodeImpl[]) obj);
            } else {
                super.putWireFormat(buffer, obj);
            }
        }

        @Override
        protected int wireFormatSize(final Object obj) {
            if (obj.getClass() == RepNodeImpl[].class) {
                return getRepNodeImplArraySize((RepNodeImpl[]) obj);
            }
            return super.wireFormatSize(obj);
        }

        private void putRepNodeImplArray(final ByteBuffer buffer,
                                         final RepNodeImpl[] ra) {
            LogUtils.writeInt(buffer, ra.length);
            final int groupFormatVersion = getGroupFormatVersion();
            for (final RepNodeImpl node : ra) {
                putByteArray(
                    buffer,
                    RepGroupImpl.serializeBytes(node, groupFormatVersion));
            }
        }

        private RepNodeImpl[] getRepNodeImplArray(final ByteBuffer buffer) {
            final RepNodeImpl[] ra = new RepNodeImpl[LogUtils.readInt(buffer)];
            final int groupFormatVersion = getGroupFormatVersion();
            for (int i = 0; i < ra.length; i++) {
                ra[i] = RepGroupImpl.deserializeNode(
                    getByteArray(buffer), groupFormatVersion);
            }
            return ra;
        }

        private int getRepNodeImplArraySize(RepNodeImpl[] ra) {
            int size = 4; /* array length */
            final int groupFormatVersion = getGroupFormatVersion();
            for (final RepNodeImpl node : ra) {
                size += (4 /* Node size */ +
                         RepGroupImpl.serializeBytes(node, groupFormatVersion)
                         .length);
            }
            return size;
        }

        /**
         * Returns the RepGroupImpl version to use for the currently configured
         * protocol version.
         */
        private int getGroupFormatVersion() {
            return (getVersion() < VERSION_5) ?
                RepGroupImpl.FORMAT_VERSION_2 :
                RepGroupImpl.MAX_FORMAT_VERSION;
        }

        @Override
        public MessageOp getOp() {
            return RESTORE_RESPONSE;
        }

        RepNodeImpl[] getLogProviders() {
            return logProviders;
        }

        VLSN getCBVLSN() {
            return cbvlsn;
        }
    }
}