Motivation

Ignite use a few protocols of inter-node message exchange:

1. Communication protocol
2. Discovery protocol

All protocols have own serialization mechanisms and Currently Communication protocol doesn't support message exchange with node of another version. For making Rolling upgrade feature possible we must make the protocol these protocols compatible between Ignite versions. 

Prerequisites

...

1. Communication protocol is peer-to-peer protocol: every message is sent between one node-sender and one node-receiver.
2. Communication protocol is an internal protocol used for system messages exchange between Ignite nodes:
   1. There is a limited amount of peers, and all peers are aware of versions of each other.
   2. All peers are aware of all possible messages and their schemas.
   3. At most time peers are of same version. In short period of time (during RU) messags schemas might differ, but not much (few messages might differ with few fields only).
3. It is proposed to provide compatibility between versions that differ by 1 minor version, for example between 2.20.X and 2.19.X (but not between 2.20.X and 2.18.X).
4. Messages can be pretty big, contains cache entries (putAll, historic rebalance).Code changes should be minimal.
5. Users applications deployed on cluster (services, etc) must support compatibility by itself.

Current implementation

...

Communication protocol

...

• All peers hold predefined messages schemas (described with java classes).
• Messages follow schema in strict way - no fields are skipped and order of fields is guaranteed.
• Peers send messages as byte stream serializing fields one by one, no delimiters are used between fields and messages.
• For serializing POJO fields different techniques are used:
  • Message consumes already serialized byte array (e.g. JdkMarshaller for IgniteDiagnosticMessage, SchemaOperationStatusMessage... BinaryMarshaller for GridJobExecuteRequest#jobBytes).
  • User's BinaryObjects, Binarylizable are marshalled to byte array prior to serializing a message (the array is stored in cache structures).
  • Some messages contain POJO field annotated with GridDirectTransient, and additional byte array field fill with GridCacheMessage#prepareMarshal (e.g. BinaryMarshaller for GridChangeGlobalStateMessageResponse).
  • Some messages contains POJO that implements Message interface. 
  GridCacheMessage holds additional field depInfo
  • .
   

Discovery protocol

...

• All peers hold predefined messages schemas (described with java classes).
• Peers use JdkMarshaller (java serialization) to serialize and deserialize a message.

Proposed changes

Message DTO

• full message. Borders between messages are controlled by java serialization.
• Responses with status code or ResponseMessage.

Management command protocol

There is an additional serialization protocol between control.sh (thin client) and server nodes

• Management operates compute tasks for sending commands.
• Compute args are serialized with IgniteDataTransferObject (that is actually simple compacted Externalizable - fields without java staff). 

This protocol is out of scope of this proposal.

Proposed changes

Validate joining node version

The joining node version must be checked versus actual Ignite versions in a cluster (within OnDiscoveryNodeValidationProcessor):

• If a cluster contains nodes of only version, then the joining node can be greater/less than this version by up to 1 minor version.
• If a cluster contains nodes of two versions, then joining node version must be one of them.

Message serialization framework

There should be one serialization framework for communication and discovery protocol.

1. MessageWriter, MessageReader Message#writeTo, Message#readFrom logic is depends on a remote IgniteProductVersion.
2. Message#writeTo, Message#readFrom is auto-generated and stored separately from Message DTO classes classes.
3. Add Message#marshal(Marshaller), Message#unmarshal(Marshaller) methods - to marshal POJO fields if needed. 
4. Add @Since, @Until  annotations for Message classes and fields.
5. Message#writeObject - for serializing Objects (java functions, and user objects)  with  BinaryMarshaller.
6. Message fields may contain:
   1. primitive classes.
   2. known collections.
   3. others Message.
   4. Users classes and java functions (e.g. ComputeJob)
   Marshalling must be done within Message#writeTo with specifying remote version to marshaller
   1. .
   2. byte[] fields in Message must be forbidden.

Marshaller 

1. Marshaller#marshal, Marshaller#unmarshal is depends on remote IgniteProductVersion.

Version check

1. 1. (objects serialized externally) must be avoided as much as possible, because their compatibility can't be guaranteed.
2. Order of fields in Message is fixed with @Order annotation.
3. Adding and removing fields must be go along with @Since, @Until annotations for Message classes and fields.

Code checks

1. Ignite CI notifies Ignite CI must notify for IF-clauses with condition based on IgniteVersion older than (curVer - 1).
2. Ignite CI must forbid forbids code changes if Message DTO fields changed without corresponding @Since, @Until annotations.
3. Ignite CI must forbid forbids code changes if Message DTO contains byte[] fields.
4. Ignite CI forbids code changes if Message field changes type.
5. Ignite CI checks @Order annotation of fields - starts with 0, no lags.

Nice to have (for later research)

...

1. Optional tagged fields. Such fields is not part of Message schema. The fields can be attached to any message. Cases: securityId, traceId, incremenalIndex, sessionAttributes, depInfo, etc.
   Current approach - is creating a new message-wrapper (IncrementalSnapshotAwareMessage, TransactionAttributesAwareRequest) that wraps original Message with extra data.
2. Lazy deserialization/unmarshalling of specific fields. Cases: skip deserializing optional fields, transfer cache entries as byte arrays. It can be achieved by storing these fields as byte array.
3. Compact length of varlen/collections - currently we use 4 bytes (int) to write length of collection or varlen type. In most cases this too much and the length can be encoded with less data (using 1-2 bytes instead).

Communication protocol

Communication protocol consist of 2 parts:

1. Data - set of declared Message classes, including ser/des algorithm.
2. Transport - algorithm of transport the Messages to remote node.

Data compatibility

Message - is base class for all messages transported between nodes. Proposed changes:

1. Deprecate Message#writeTo and Message#readFrom in favor generated MessageSerializer. 
2. MessageSerializer#writeTo consumes MessageWriter that stores IgniteProductVersion of a receiver Message#writeTo consumes IgniteProductVersion of destination node and use it for serializing data for this version (mostly, for ignoring some fields).

3. Message#readFrom MessageSerializer#readFrom consumes MessageReader that stores IgniteProductVersion of source node and use it for deserializing data (mostly, for setting default values of new fields).

   Code Block
   language java title Message
   public interface MessageMessageSerializer {    public boolean writeTo(Message msg, ByteBuffer buf, MessageWriter writer, IgniteProductVersion destVer);     public boolean readFrom(Message msg, ByteBuffer buf, MessageReader reader,); } public interface MessageWriter { public IgniteProductVersion srcVerreceiverVersion();    } public interface MessageReader { public shortIgniteProductVersion directTypesenderVersion(); }

   
   

4. Introduce annotations @Since and @Until for Message classes and Message fields, to use it for generating code for Message#writeTo and Message#readFrom:
   
   

// Package thatwhere the storeMessage allis schemasdefined.
package org.apache.ignite.internal.messagesmy.schemamessage;

// package private class.
@Since(version = "2.19.0")
public class MyMessage implements Message {
    @Order(0)
    private int id;
    
 	/** Remove field. */
    @Until(version = "2.20.0")
    @Order(1)
    private String rmFld;

    /** New field. */
    @Since(version = "2.20.0")
 	@Order(2)
    private String newFld;
}

	// Message must Generatedhave codesetters/getters fromfor theall schema@Ordered ^fields. 
package org.apache.ignite.internal.messages;

// public class.
public class MyMessage {
    private int id;
    
	/**
     * Remove field. 
     * @deprecated since 2.20.0. 
     */
    @Deprecated
    private String rmFld;

    /** 
     * New field.
     * @since 2.20.0
     */
    private String newFld;

    @Override public boolean writeTo(Methods names are equal to a corresponding field name. 
	public void id(int id) {
		this.id = id;
	}

  	public int id() {
		return id;
	}

  	public void newFld(String newFld) {
		this.newFld = newFld;
	}
 
 	public String newFld() {
		return newFld;
	}
}

// Generated code from the message ^ for Ignite version 2.20.0.
// Use the same package as corresponding message.
package org.apache.ignite.internal.my.message;

class MyMessageSerializer {
    public boolean writeTo(Message m, ByteBuffer buf, MessageWriter writer, IgniteProductVersion destVer) {
		MyMessage msg =      if (destVer.lessThan(2, 19, 0))
            throw new IgniteException("Must not send the message to destination node"(MyMessage)m;

		IgniteProductVersion rcvVer = writer.receiverVersion();

        if (!writer.writeString(msg.id())
            return false;

        if (destVerrcvVer.lessThan(2, 20, 0)) {
            if (!writer.writeString(msg.rmFld())
                return false;
        }

        if (destVerrcvVer.greaterThanEqual(2, 20, 0)) {
            if (!writer.writeString(msg.newFld())
                return false;
        }

        return true;
    }

    @Overridepublic publicstatic boolean readFrom(Message m, ByteBuffer buf, MessageReader reader, IgniteProductVersion srcVer) {) {
 		MyMessage msg = (MyMessage)m;
 
        idIgniteProductVersion srcVer = reader.readStringsenderVersion();

        msg.id(reader.readString());

		if (srcVer.greaterThanEquallessThan(2, 20, 0))
            newFld = 			msg.rmFld(reader.readString());

        if (srcVer.greaterThanEqual(2, 20, else0))
            newFld = nullmsg.newFld(reader.readString());

        return true;
    }
}

Rules to describe Message (must be automated and validated):

1. Do not remove Message class or Message fields, but annotate it with @Until. 
2. Do not change types or order @Order of fields.
3. New fields must be annotated with @Since. All such fields must be optional, default value is null. Handling the nulls is care of Message consumer on the reader side.

...

1. Setters and getters must follow name of the field.

Transport compatibility 

Communication handshake

...