diff --git a/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/jmh/direct/JmhDirectMessageReaderBenchmark.java b/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/jmh/direct/JmhDirectMessageReaderBenchmark.java
index 7c93e2be90b60..6735e9514b407 100644
--- a/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/jmh/direct/JmhDirectMessageReaderBenchmark.java
+++ b/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/jmh/direct/JmhDirectMessageReaderBenchmark.java
@@ -44,6 +44,7 @@
import org.openjdk.jmh.annotations.Warmup;
import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.MessageSerializationContext.IGNORED;
import static org.openjdk.jmh.annotations.Mode.Throughput;
/** Benchmarks the {@link DirectMessageReader} compressed-field hot path. */
@@ -96,7 +97,7 @@ public void setup() {
writer.setBuffer(buf);
- boolean finished = writer.writeMessage(msg, true);
+ boolean finished = writer.writeMessage(msg, true, IGNORED);
if (!finished)
throw new IllegalStateException("Message does not fit into the buffer.");
@@ -111,7 +112,7 @@ public Message compressedMessage() {
reader.setBuffer(buf);
- Message msg = reader.readMessage(true);
+ Message msg = reader.readMessage(true, IGNORED);
reader.reset();
diff --git a/modules/codegen/src/main/java/org/apache/ignite/internal/FeatureRegistry.java b/modules/codegen/src/main/java/org/apache/ignite/internal/FeatureRegistry.java
new file mode 100644
index 0000000000000..144d6e87324b2
--- /dev/null
+++ b/modules/codegen/src/main/java/org/apache/ignite/internal/FeatureRegistry.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeature;
+
+/**
+ * Links the annotated class to the specified {@link IgniteFeature} registry. The registry
+ * is used to resolve fully qualified names of features that introduced or deprecated fields
+ * (see {@link Order#introducedBy()} and {@link Order#deprecatedBy()}).
+ *
+ *
If this annotation is absent, the Ignite Core Feature Registry is used.
+ *
+ * @see Order
+ * @see IgniteFeature
+ */
+@Retention(RetentionPolicy.CLASS)
+@Target(ElementType.TYPE)
+public @interface FeatureRegistry {
+ /** @return Class of the feature registry. */
+ Class value();
+}
diff --git a/modules/codegen/src/main/java/org/apache/ignite/internal/MessageProcessor.java b/modules/codegen/src/main/java/org/apache/ignite/internal/MessageProcessor.java
index 8b4ead738ee22..4f30a67f1bf1d 100644
--- a/modules/codegen/src/main/java/org/apache/ignite/internal/MessageProcessor.java
+++ b/modules/codegen/src/main/java/org/apache/ignite/internal/MessageProcessor.java
@@ -41,6 +41,7 @@
import javax.lang.model.element.TypeElement;
import javax.lang.model.element.VariableElement;
import javax.lang.model.type.DeclaredType;
+import javax.lang.model.type.MirroredTypeException;
import javax.lang.model.type.TypeMirror;
import javax.lang.model.util.ElementFilter;
import javax.lang.model.util.Elements;
@@ -48,9 +49,11 @@
import org.apache.ignite.internal.systemview.SystemViewRowAttributeWalkerProcessor;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.lang.IgniteBiTuple;
+import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.MessageSerializerGenerator.DLFT_ENUM_MAPPER_CLS;
import static org.apache.ignite.internal.MessageSerializerGenerator.enumType;
+import static org.apache.ignite.internal.MessageSerializerGenerator.qualifiedClassName;
/**
* Annotation processor that generates serialization and deserialization code for classes implementing the {@code Message} interface.
@@ -96,6 +99,13 @@ public class MessageProcessor extends AbstractProcessor {
/** Checked exception declared by the generated methods. */
static final String IGNITE_CHECKED_EXCEPTION_CLS = "org.apache.ignite.IgniteCheckedException";
+ /** Feature registry a message resolves its guards against unless it declares one with {@link FeatureRegistry}. */
+ static final String DFLT_FEATURE_REG_CLS =
+ "org.apache.ignite.internal.processors.rollingupgrade.feature.SupportedFeatureRegistry";
+
+ /** */
+ static final String IGNITE_FEATURE_CLS = "org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeature";
+
/** */
public static final String GRID_H2_NULL = "org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2Null";
@@ -105,10 +115,14 @@ public class MessageProcessor extends AbstractProcessor {
/** */
public static final Set NO_PUBLIC_CTOR_MSGS = Set.of(GRID_H2_NULL, ZK_NO_SERVERS_MESSAGE);
- /** Messages with no fields. A serializer generation intentionally skipped. */
+ /** */
+ static final String OP_CTX_SNAPSHOT_MESSAGE_CLASS = "org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage";
+
+ /** Messages with no fields, or with a hand-written serializer. A serializer generation intentionally skipped. */
static final String[] SKIP_MESSAGES = {
"org.apache.ignite.internal.processors.odbc.ClientMessage",
COMPRESSED_MESSAGE_CLASS,
+ OP_CTX_SNAPSHOT_MESSAGE_CLASS,
"org.apache.ignite.loadtests.communication.GridTestMessage",
"org.apache.ignite.spi.communication.tcp.TestDelayMessage"
};
@@ -425,4 +439,99 @@ TypeMirror type(String clazz) {
TypeElement typeElement = elementUtils.getTypeElement(clazz);
return typeElement != null ? typeElement.asType() : null;
}
+
+ /** */
+ @Nullable public static FieldFeatureGuard buildFieldFeatureGuard(ProcessingEnvironment env, VariableElement field) {
+ Order ann = field.getAnnotation(Order.class);
+
+ String introducingFeature = ann.introducedBy();
+ String deprecatingFeature = ann.deprecatedBy();
+
+ if (introducingFeature.isEmpty() && deprecatingFeature.isEmpty())
+ return null;
+
+ if (introducingFeature.equals(deprecatingFeature)) {
+ printError(env, field, "Elements introducedBy and deprecatedBy of the @Order annotation must not reference the same feature.");
+
+ return null;
+ }
+
+ String regCls = resolveFeatureRegistry(field.getEnclosingElement());
+
+ String regName = regCls.substring(regCls.lastIndexOf('.') + 1);
+
+ List conditions = new ArrayList<>();
+
+ if (!introducingFeature.isEmpty()) {
+ validateFeature(env, field, introducingFeature, regCls);
+
+ conditions.add("ctx.includeFieldIntroducedBy(" + regName + '.' + introducingFeature + ")");
+ }
+
+ if (!deprecatingFeature.isEmpty()) {
+ validateFeature(env, field, deprecatingFeature, regCls);
+
+ conditions.add("ctx.includeFieldDeprecatedBy(" + regName + '.' + deprecatingFeature + ")");
+ }
+
+ return new FieldFeatureGuard(regCls, String.join(" && ", conditions));
+ }
+
+ /** */
+ private static void validateFeature(ProcessingEnvironment env, VariableElement field, String featureName, String regCls) {
+ TypeElement regElem = env.getElementUtils().getTypeElement(regCls);
+
+ if (regElem == null) {
+ printError(env, field, "Cannot resolve the feature registry class [reg=" + regCls + ']');
+
+ return;
+ }
+
+ for (Element featureElem : regElem.getEnclosedElements()) {
+ if (featureElem.getKind() != ElementKind.FIELD || !featureElem.getSimpleName().contentEquals(featureName))
+ continue;
+
+ Set mods = featureElem.getModifiers();
+
+ if (!mods.contains(Modifier.PUBLIC) || !mods.contains(Modifier.STATIC) || !mods.contains(Modifier.FINAL))
+ printError(env, field, "Feature constant must be public static final [reg=" + regCls + ", feature=" + featureName + ']');
+ else if (!isIgniteFeature(env, featureElem))
+ printError(env, field, "Feature constant must be of type IgniteFeature [reg=" + regCls + ", feature=" + featureName + ']');
+
+ return;
+ }
+
+ printError(env, field,
+ "Failed to resolve feature in the registry by its name [reg=" + regCls + ", feature=" + featureName + ']');
+ }
+
+ /** */
+ private static boolean isIgniteFeature(ProcessingEnvironment env, Element featureElem) {
+ TypeElement igniteFeatureType = env.getElementUtils().getTypeElement(IGNITE_FEATURE_CLS);
+
+ return igniteFeatureType != null && env.getTypeUtils().isAssignable(featureElem.asType(), igniteFeatureType.asType());
+ }
+
+ /** */
+ private static void printError(ProcessingEnvironment env, Element el, String msg) {
+ env.getMessager().printMessage(Diagnostic.Kind.ERROR, msg, el);
+ }
+
+ /** */
+ private static String resolveFeatureRegistry(Element cls) {
+ FeatureRegistry ann = cls.getAnnotation(FeatureRegistry.class);
+
+ if (ann == null)
+ return DFLT_FEATURE_REG_CLS;
+
+ try {
+ return ann.value().getName();
+ }
+ catch (MirroredTypeException e) {
+ return qualifiedClassName(e.getTypeMirror());
+ }
+ }
+
+ /** */
+ public record FieldFeatureGuard(String registry, String expression) { }
}
diff --git a/modules/codegen/src/main/java/org/apache/ignite/internal/MessageSerializerGenerator.java b/modules/codegen/src/main/java/org/apache/ignite/internal/MessageSerializerGenerator.java
index 81680f1b1088c..c3b483c017381 100644
--- a/modules/codegen/src/main/java/org/apache/ignite/internal/MessageSerializerGenerator.java
+++ b/modules/codegen/src/main/java/org/apache/ignite/internal/MessageSerializerGenerator.java
@@ -41,6 +41,7 @@
import javax.lang.model.type.PrimitiveType;
import javax.lang.model.type.TypeKind;
import javax.lang.model.type.TypeMirror;
+import org.apache.ignite.internal.MessageProcessor.FieldFeatureGuard;
import org.apache.ignite.internal.systemview.SystemViewRowAttributeWalkerProcessor;
import org.apache.ignite.internal.util.typedef.F;
import org.jetbrains.annotations.Nullable;
@@ -50,6 +51,7 @@
import static org.apache.ignite.internal.MessageProcessor.GRID_H2_NULL;
import static org.apache.ignite.internal.MessageProcessor.KEY_CACHE_OBJECT_CLS;
import static org.apache.ignite.internal.MessageProcessor.MESSAGE_INTERFACE;
+import static org.apache.ignite.internal.MessageProcessor.buildFieldFeatureGuard;
/** Generates {@code *Serializer} classes for {@code Message} types. */
public class MessageSerializerGenerator extends MessageCompanionGenerator {
@@ -68,6 +70,9 @@ public class MessageSerializerGenerator extends MessageCompanionGenerator {
/** */
private static final String MESSAGE_READER_CLS = "org.apache.ignite.plugin.extensions.communication.MessageReader";
+ /** */
+ private static final String MESSAGE_SER_CTX_CLS = "org.apache.ignite.internal.MessageSerializationContext";
+
/** */
private static final String ENUM_MAPPER_CLS = "org.apache.ignite.plugin.extensions.communication.mappers.EnumMapper";
@@ -145,6 +150,7 @@ public class MessageSerializerGenerator extends MessageCompanionGenerator {
imports.add(MESSAGE_SERIALIZER_CLS);
imports.add(MESSAGE_WRITER_CLS);
imports.add(MESSAGE_READER_CLS);
+ imports.add(MESSAGE_SER_CTX_CLS);
writeClassHeader(writer, "MessageSerializer", serClsName);
@@ -195,8 +201,10 @@ private void generateMethods(List fields) throws Exception {
private void generateMethod(List code, List fields, boolean write) throws Exception {
code.add(indentedLine(METHOD_JAVADOC));
- code.add(indentedLine("@Override public final boolean %s(" + simpleNameWithGeneric(type) + " msg, %s) {",
- write ? "writeTo" : "readFrom", write ? "MessageWriter writer" : "MessageReader reader"));
+ code.add(indentedLine(
+ "@Override public final boolean %s(" + simpleNameWithGeneric(type) + " msg, %s, MessageSerializationContext ctx) {",
+ write ? "writeTo" : "readFrom",
+ write ? "MessageWriter writer" : "MessageReader reader"));
indent++;
@@ -261,7 +269,7 @@ private void processField(VariableElement field, int opt, boolean write) throws
throw new UnsupportedOperationException("You should use ErrorMessage for serialization of throwables.");
if (write)
- writeField(opt, callExpr(field, true));
+ writeField(field, opt, callExpr(field, true));
else
readField(field, opt, callExpr(field, false));
}
@@ -301,13 +309,29 @@ private String callExpr(VariableElement field, boolean write) throws Exception {
* @param opt Case option.
* @param writeExpr Writer call expression.
*/
- private void writeField(int opt, String writeExpr) {
+ private void writeField(VariableElement field, int opt, String writeExpr) {
write.add(indentedLine("case %d:", opt));
indent++;
+ FieldFeatureGuard guard = buildFieldFeatureGuard(env, field);
+
+ if (guard != null) {
+ imports.add(guard.registry());
+
+ write.add(indentedLine("if (%s) {", guard.expression()));
+
+ indent++;
+ }
+
returnFalseIf(write, "!" + writeExpr);
+ if (guard != null) {
+ indent--;
+
+ write.add(indentedLine("}"));
+ }
+
write.add(EMPTY);
write.add(indentedLine("writer.incrementState();"));
write.add(EMPTY);
@@ -335,11 +359,27 @@ private void readField(VariableElement field, int opt, String readExpr) {
indent++;
+ FieldFeatureGuard guard = buildFieldFeatureGuard(env, field);
+
+ if (guard != null) {
+ imports.add(guard.registry());
+
+ read.add(indentedLine("if (%s) {", guard.expression()));
+
+ indent++;
+ }
+
read.add(indentedLine("%s = %s;", fieldRef(field), readExpr));
read.add(EMPTY);
returnFalseIf(read, "!reader.isLastRead()");
+ if (guard != null) {
+ indent--;
+
+ read.add(indentedLine("}"));
+ }
+
read.add(EMPTY);
read.add(indentedLine("reader.incrementState();"));
read.add(EMPTY);
@@ -360,13 +400,13 @@ private FieldCall fieldCall(VariableElement field) throws Exception {
checkTypeForCompress(type);
if (type.getKind().isPrimitive())
- return new FieldCall(capitalizeOnlyFirst(type.getKind().name()), null, false);
+ return FieldCall.scalar(capitalizeOnlyFirst(type.getKind().name()));
if (type.getKind() == TypeKind.ARRAY) {
TypeMirror compType = ((ArrayType)type).getComponentType();
if (compType.getKind().isPrimitive())
- return new FieldCall(capitalizeOnlyFirst(compType.getKind().name()) + "Array", null, false);
+ return FieldCall.scalar(capitalizeOnlyFirst(compType.getKind().name()) + "Array");
if (compType.getKind() == TypeKind.DECLARED) {
Element compElem = ((DeclaredType)compType).asElement();
@@ -375,52 +415,52 @@ private FieldCall fieldCall(VariableElement field) throws Exception {
imports.add(((QualifiedNameable)compElem).getQualifiedName().toString());
}
- return new FieldCall("ObjectArray", messageCollectionItemTypes(field, type), false);
+ return FieldCall.collection("ObjectArray", messageCollectionItemTypes(field, type), false);
}
if (type.getKind() == TypeKind.DECLARED) {
if (sameType(type, String.class))
- return new FieldCall("String", null, false);
+ return FieldCall.scalar("String");
if (sameType(type, BitSet.class))
- return new FieldCall("BitSet", null, false);
+ return FieldCall.scalar("BitSet");
if (sameType(type, UUID.class))
- return new FieldCall("Uuid", null, false);
+ return FieldCall.scalar("Uuid");
if (sameType(type, IGNITE_UUID_CLS))
- return new FieldCall("IgniteUuid", null, false);
+ return FieldCall.scalar("IgniteUuid");
if (sameType(type, AFFINITY_TOPOLOGY_VERSION_CLS))
- return new FieldCall("AffinityTopologyVersion", null, false);
+ return FieldCall.scalar("AffinityTopologyVersion");
if (assignableFrom(erasedType(type), type(Map.class.getName())))
- return new FieldCall("Map", messageCollectionItemTypes(field, type), compress);
+ return FieldCall.collection("Map", messageCollectionItemTypes(field, type), compress);
if (assignableFrom(type, type(KEY_CACHE_OBJECT_CLS)))
- return new FieldCall("KeyCacheObject", null, false);
+ return FieldCall.scalar("KeyCacheObject");
if (assignableFrom(type, type(CACHE_OBJECT_CLS)))
- return new FieldCall("CacheObject", null, false);
+ return FieldCall.scalar("CacheObject");
if (assignableFrom(type, type(GRID_LONG_LIST_CLS)))
- return new FieldCall("GridLongList", null, false);
+ return FieldCall.scalar("GridLongList");
if (assignableFrom(type, type(IGNITE_PRODUCT_VERSION_CLS)))
- return new FieldCall("IgniteProductVersion", null, false);
+ return FieldCall.scalar("IgniteProductVersion");
if (assignableFrom(type, type(GRID_CACHE_VERSION_CLS)))
- return new FieldCall("GridCacheVersion", null, false);
+ return FieldCall.scalar("GridCacheVersion");
if (assignableFrom(type, type(MESSAGE_INTERFACE))) {
if (sameType(type, COMPRESSED_MESSAGE_CLASS))
throw new IllegalArgumentException(COMPRESSED_MSG_ERROR);
- return new FieldCall("Message", null, compress);
+ return FieldCall.message(compress);
}
if (assignableFrom(erasedType(type), type(Collection.class.getName())))
- return new FieldCall("Collection", messageCollectionItemTypes(field, type), false);
+ return FieldCall.collection("Collection", messageCollectionItemTypes(field, type), false);
throw new IllegalArgumentException("Unsupported declared type: " + type);
}
@@ -756,10 +796,14 @@ private static final class FieldCall {
private final boolean compress;
/** */
- private FieldCall(String mtd, @Nullable String collDesc, boolean compress) {
+ private final boolean isSerCtxRequired;
+
+ /** */
+ private FieldCall(String mtd, @Nullable String collDesc, boolean compress, boolean isSerCtxRequired) {
this.mtd = mtd;
this.collDesc = collDesc;
this.compress = compress;
+ this.isSerCtxRequired = isSerCtxRequired;
}
/** @return Full call expression; {@code valArg}, when given, is passed as the first argument (write side). */
@@ -775,8 +819,26 @@ private String expr(String mtdPrefix, @Nullable String valArg) {
if (compress)
args.add("true");
+ if (isSerCtxRequired)
+ args.add("ctx");
+
return mtdPrefix + mtd + "(" + String.join(", ", args) + ")";
}
+
+ /** */
+ private static FieldCall scalar(String mtd) {
+ return new FieldCall(mtd, null, false, false);
+ }
+
+ /** */
+ private static FieldCall collection(String mtd, String collDesc, boolean compress) {
+ return new FieldCall(mtd, collDesc, compress, true);
+ }
+
+ /** */
+ private static FieldCall message(boolean compress) {
+ return new FieldCall("Message", null, compress, true);
+ }
}
/** */
diff --git a/modules/codegen/src/main/java/org/apache/ignite/internal/Order.java b/modules/codegen/src/main/java/org/apache/ignite/internal/Order.java
index 0e4562537c435..3eecc4cbca38c 100644
--- a/modules/codegen/src/main/java/org/apache/ignite/internal/Order.java
+++ b/modules/codegen/src/main/java/org/apache/ignite/internal/Order.java
@@ -21,6 +21,7 @@
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeature;
/**
* The annotation specifies the position of a field in the serialized and deserialized byte sequence of a {@code Message} class.
@@ -28,14 +29,38 @@
* The {@code value} indicates the index of the field in the serialization order.
* Fields annotated with {@code @Order} are processed in ascending order of their index.
*
By default, it is assumed that getters and setters are named as the annotated fields,
- * e.g. field 'val' should have getters and satters with name 'val' (according Ignite's to code-style).
+ * e.g. field 'val' should have getters and setters with name 'val' (according Ignite's to code-style).
*
This annotation must be used on non-static fields, and access to those fields
* should be performed strictly through corresponding getter and setter methods
* following the naming convention: {@code fieldName()} for getter and {@code fieldName(Type)} for setter.
+ *
+ * @see FeatureRegistry
*/
@Retention(RetentionPolicy.CLASS)
@Target(ElementType.FIELD)
public @interface Order {
/** @return Order of the field. */
int value();
+
+ /**
+ * {@link IgniteFeature} that introduced the field marked with the current annotation.
+ *
+ *
An annotated field is included in message serialization only when doing so does not break backward compatibility
+ * during a Rolling Upgrade.
+ *
+ * @return Name of the Ignite feature that introduced this field, or an empty string if the field is not guarded.
+ */
+ String introducedBy() default "";
+
+ /**
+ * {@link IgniteFeature} that deprecated the field marked with the current annotation.
+ *
+ *
Deprecation means that the field is planned for removal in a future release.
+ *
+ *
An annotated field is excluded from message serialization only when doing so does not break backward compatibility
+ * during a Rolling Upgrade.
+ *
+ * @return Name of the Ignite feature that deprecated this field, or an empty string if the field is not guarded.
+ */
+ String deprecatedBy() default "";
}
diff --git a/modules/codegen/src/main/java/org/apache/ignite/internal/idto/IDTOSerializerGenerator.java b/modules/codegen/src/main/java/org/apache/ignite/internal/idto/IDTOSerializerGenerator.java
index 87ef9dac8a190..80cf62c153791 100644
--- a/modules/codegen/src/main/java/org/apache/ignite/internal/idto/IDTOSerializerGenerator.java
+++ b/modules/codegen/src/main/java/org/apache/ignite/internal/idto/IDTOSerializerGenerator.java
@@ -34,7 +34,6 @@
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.UUID;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.annotation.processing.FilerException;
import javax.annotation.processing.ProcessingEnvironment;
@@ -52,6 +51,7 @@
import javax.lang.model.type.TypeMirror;
import javax.tools.Diagnostic;
import javax.tools.JavaFileObject;
+import org.apache.ignite.internal.MessageProcessor.FieldFeatureGuard;
import org.apache.ignite.internal.Order;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.lang.IgniteBiTuple;
@@ -61,6 +61,7 @@
import static org.apache.ignite.internal.MessageCompanionGenerator.TAB;
import static org.apache.ignite.internal.MessageCompanionGenerator.identicalFileIsAlreadyGenerated;
import static org.apache.ignite.internal.MessageCompanionGenerator.writeLicense;
+import static org.apache.ignite.internal.MessageProcessor.buildFieldFeatureGuard;
import static org.apache.ignite.internal.MessageSerializerGenerator.enumType;
import static org.apache.ignite.internal.MessageSerializerGenerator.qualifiedClassName;
import static org.apache.ignite.internal.idto.IgniteDataTransferObjectProcessor.DTO_CLASS;
@@ -253,6 +254,7 @@ private String generateSerializerCode() throws IOException {
imports.add(ObjectInput.class.getName());
imports.add(IOException.class.getName());
imports.add("org.apache.ignite.internal.util.typedef.internal.U");
+ imports.add("org.apache.ignite.internal.MessageSerializationContext");
if (type.getNestingKind() != NestingKind.TOP_LEVEL)
imports.add(type.getQualifiedName().toString());
@@ -312,7 +314,10 @@ private List generateWrite(Collection flds) {
List code = new ArrayList<>();
code.add("/** {@inheritDoc} */");
- code.add("@Override public void writeExternal(" + typeWithGeneric(type.asType()) + " obj, ObjectOutput out) throws IOException {");
+ code.add("@Override public void writeExternal(" +
+ typeWithGeneric(type.asType()) + " obj," +
+ " ObjectOutput out," +
+ " MessageSerializationContext ctx) throws IOException {");
fieldsSerdes(flds).forEach(line -> code.add(TAB + line));
@@ -328,8 +333,10 @@ private List generateRead(Collection flds) {
List code = new ArrayList<>();
code.add("/** {@inheritDoc} */");
- code.add("@Override public void readExternal(" + typeWithGeneric(type.asType()) + " obj, ObjectInput in) " +
- "throws IOException, ClassNotFoundException {");
+ code.add("@Override public void readExternal(" +
+ typeWithGeneric(type.asType()) + " obj," +
+ " ObjectInput in," +
+ " MessageSerializationContext ctx) throws IOException, ClassNotFoundException {");
fieldsSerdes(flds).forEach(line -> code.add(TAB + line));
@@ -343,9 +350,27 @@ private List generateRead(Collection flds) {
* @return Lines to serdes fields.
*/
private List fieldsSerdes(Collection flds) {
- return flds.stream()
- .flatMap(fld -> variableCode(fld.asType(), "obj." + fld.getSimpleName().toString()))
- .collect(Collectors.toList());
+ List res = new ArrayList<>();
+
+ for (VariableElement fld : flds) {
+ List lines = variableCode(fld.asType(), "obj." + fld.getSimpleName()).toList();
+
+ FieldFeatureGuard guard = buildFieldFeatureGuard(env, fld);
+
+ if (guard == null)
+ res.addAll(lines);
+ else {
+ imports.add(guard.registry());
+
+ res.add("if (" + guard.expression() + ") {");
+
+ lines.forEach(line -> res.add(TAB + line));
+
+ res.add("}");
+ }
+ }
+
+ return res;
}
/**
diff --git a/modules/commons/src/main/java/org/apache/ignite/internal/MessageSerializationContext.java b/modules/commons/src/main/java/org/apache/ignite/internal/MessageSerializationContext.java
new file mode 100644
index 0000000000000..7db50eac2b28e
--- /dev/null
+++ b/modules/commons/src/main/java/org/apache/ignite/internal/MessageSerializationContext.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal;
+
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeature;
+
+/** Represents the context that determines how message fields are serialized and deserialized when transmitted between nodes. */
+public interface MessageSerializationContext {
+ /**
+ * @param feature Feature that deprecated the field.
+ * @return {@code true} if the message field should be included during message serialization or deserialization.
+ */
+ boolean includeFieldDeprecatedBy(IgniteFeature feature);
+
+ /**
+ * @param feature Feature that introduced the field.
+ * @return {@code true} if the message field should be included during message serialization or deserialization.
+ */
+ boolean includeFieldIntroducedBy(IgniteFeature feature);
+
+ /**
+ * {@link MessageSerializationContext} implementation that instructs the serialization framework to always
+ * serialize the actual message state: all newly introduced fields are included, and all deprecated fields are
+ * excluded.
+ */
+ MessageSerializationContext IGNORED = new MessageSerializationContext() {
+ /** {@inheritDoc} */
+ @Override public boolean includeFieldDeprecatedBy(IgniteFeature feature) {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean includeFieldIntroducedBy(IgniteFeature feature) {
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "MessageSerializationContext [IGNORED]";
+ }
+ };
+
+ /**
+ * Stub {@link MessageSerializationContext} implementation used when the serialization context has not yet been determined.
+ *
+ *
The serialization context is unavailable between connection establishment and serialization protocol negotiation.
+ * Messages sent during this period cannot rely on the {@link IgniteFeature} mechanism to adjust the message serialization
+ * in an RU-compatible way.
+ */
+ MessageSerializationContext UNNEGOTIATED = new MessageSerializationContext() {
+ /** {@inheritDoc} */
+ @Override public boolean includeFieldDeprecatedBy(IgniteFeature feature) {
+ throw buildError(feature);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean includeFieldIntroducedBy(IgniteFeature feature) {
+ throw buildError(feature);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "MessageSerializationContext [UNNEGOTIATED]";
+ }
+
+ /** */
+ private IllegalStateException buildError(IgniteFeature feature) {
+ return new IllegalStateException(
+ "A feature-guarded field was serialized before the peer's features were negotiated [feature=" + feature + ']'
+ );
+ }
+ };
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteFeature.java b/modules/commons/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteFeature.java
similarity index 100%
rename from modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteFeature.java
rename to modules/commons/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteFeature.java
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/OfflineTestCommandArgSerializer.java b/modules/control-utility/src/test/java/org/apache/ignite/util/OfflineTestCommandArgSerializer.java
index b59278d3db4f8..03af45f17ccf7 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/OfflineTestCommandArgSerializer.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/OfflineTestCommandArgSerializer.java
@@ -20,6 +20,7 @@
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.dto.IgniteDataTransferObject;
import org.apache.ignite.internal.dto.IgniteDataTransferObjectSerializer;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -36,12 +37,20 @@
*/
public class OfflineTestCommandArgSerializer implements IgniteDataTransferObjectSerializer {
/** {@inheritDoc} */
- @Override public void writeExternal(TestOfflineTestCommandArg obj, ObjectOutput out) throws IOException {
+ @Override public void writeExternal(
+ TestOfflineTestCommandArg obj,
+ ObjectOutput out,
+ MessageSerializationContext ctx
+ ) throws IOException {
U.writeString(out, obj.input);
}
/** {@inheritDoc} */
- @Override public void readExternal(TestOfflineTestCommandArg obj, ObjectInput in) throws IOException, ClassNotFoundException {
+ @Override public void readExternal(
+ TestOfflineTestCommandArg obj,
+ ObjectInput in,
+ MessageSerializationContext ctx
+ ) throws IOException, ClassNotFoundException {
obj.input = U.readString(in);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java
index 9ec9b68cc189e..b070629a38ba3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java
@@ -24,6 +24,7 @@
import java.util.UUID;
import java.util.function.Function;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.direct.state.DirectMessageState;
import org.apache.ignite.internal.direct.state.DirectMessageStateItem;
import org.apache.ignite.internal.direct.stream.DirectByteBufferStream;
@@ -344,7 +345,7 @@ public ByteBuffer getBuffer() {
}
/** {@inheritDoc} */
- @Nullable @Override public T readMessage(boolean compress) {
+ @Nullable @Override public T readMessage(boolean compress, MessageSerializationContext ctx) {
DirectByteBufferStream stream = curStream;
T msg;
@@ -352,10 +353,11 @@ public ByteBuffer getBuffer() {
if (compress)
msg = readCompressedMessageAndDeserialize(
stream,
- r -> r.state.item().stream.readMessage(r)
+ r -> r.state.item().stream.readMessage(r, ctx),
+ ctx
);
else {
- msg = stream.readMessage(this);
+ msg = stream.readMessage(this, ctx);
lastRead = stream.lastFinished();
}
@@ -397,10 +399,10 @@ public ByteBuffer getBuffer() {
}
/** {@inheritDoc} */
- @Override public T[] readObjectArray(MessageArrayType type) {
+ @Override public T[] readObjectArray(MessageArrayType type, MessageSerializationContext ctx) {
DirectByteBufferStream stream = curStream;
- T[] msg = stream.readObjectArray(type, this);
+ T[] msg = stream.readObjectArray(type, this, ctx);
lastRead = stream.lastFinished();
@@ -408,10 +410,10 @@ public ByteBuffer getBuffer() {
}
/** {@inheritDoc} */
- @Override public > C readCollection(MessageCollectionType type) {
+ @Override public > C readCollection(MessageCollectionType type, MessageSerializationContext ctx) {
DirectByteBufferStream stream = curStream;
- C col = stream.readCollection(type, this);
+ C col = stream.readCollection(type, this, ctx);
lastRead = stream.lastFinished();
@@ -419,7 +421,7 @@ public ByteBuffer getBuffer() {
}
/** {@inheritDoc} */
- @Override public > M readMap(MessageMapType type, boolean compress) {
+ @Override public > M readMap(MessageMapType type, boolean compress, MessageSerializationContext ctx) {
DirectByteBufferStream stream = curStream;
M map;
@@ -427,10 +429,11 @@ public ByteBuffer getBuffer() {
if (compress)
map = readCompressedMessageAndDeserialize(
stream,
- r -> r.state.item().stream.readMap(type, r)
+ r -> r.state.item().stream.readMap(type, r, ctx),
+ ctx
);
else {
- map = stream.readMap(type, this);
+ map = stream.readMap(type, this, ctx);
lastRead = stream.lastFinished();
}
@@ -509,8 +512,12 @@ public ByteBuffer getBuffer() {
}
/** @return Deserialized object. */
- private T readCompressedMessageAndDeserialize(DirectByteBufferStream stream, Function fun) {
- Message msg = stream.readMessage(this);
+ private T readCompressedMessageAndDeserialize(
+ DirectByteBufferStream stream,
+ Function fun,
+ MessageSerializationContext ctx
+ ) {
+ Message msg = stream.readMessage(this, ctx);
lastRead = stream.lastFinished();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
index 319aae7b7e947..f09632cfbac4e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
@@ -23,6 +23,7 @@
import java.util.Map;
import java.util.UUID;
import java.util.function.Consumer;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.direct.state.DirectMessageState;
import org.apache.ignite.internal.direct.state.DirectMessageStateItem;
import org.apache.ignite.internal.direct.stream.DirectByteBufferStream;
@@ -334,17 +335,18 @@ public ByteBuffer getBuffer() {
}
/** {@inheritDoc} */
- @Override public boolean writeMessage(@Nullable Message msg, boolean compress) {
+ @Override public boolean writeMessage(@Nullable Message msg, boolean compress, MessageSerializationContext ctx) {
DirectByteBufferStream stream = curStream;
if (compress)
writeCompressedMessage(
- w -> w.state.item().stream.writeMessage(msg, w),
+ w -> w.state.item().stream.writeMessage(msg, w, ctx),
msg == null,
- stream
+ stream,
+ ctx
);
else
- stream.writeMessage(msg, this);
+ stream.writeMessage(msg, this, ctx);
return stream.lastFinished();
}
@@ -377,35 +379,36 @@ public ByteBuffer getBuffer() {
}
/** {@inheritDoc} */
- @Override public boolean writeObjectArray(T[] arr, MessageArrayType type) {
+ @Override public boolean writeObjectArray(T[] arr, MessageArrayType type, MessageSerializationContext ctx) {
DirectByteBufferStream stream = curStream;
- stream.writeObjectArray(arr, type, this);
+ stream.writeObjectArray(arr, type, this, ctx);
return stream.lastFinished();
}
/** {@inheritDoc} */
- @Override public boolean writeCollection(Collection col, MessageCollectionType type) {
+ @Override public boolean writeCollection(Collection col, MessageCollectionType type, MessageSerializationContext ctx) {
DirectByteBufferStream stream = curStream;
- stream.writeCollection(col, type, this);
+ stream.writeCollection(col, type, this, ctx);
return stream.lastFinished();
}
/** {@inheritDoc} */
- @Override public boolean writeMap(Map map, MessageMapType type, boolean compress) {
+ @Override public boolean writeMap(Map map, MessageMapType type, boolean compress, MessageSerializationContext ctx) {
DirectByteBufferStream stream = curStream;
if (compress)
writeCompressedMessage(
- w -> w.state.item().stream.writeMap(map, type, w),
+ w -> w.state.item().stream.writeMap(map, type, w, ctx),
map == null,
- stream
+ stream,
+ ctx
);
else
- stream.writeMap(map, type, this);
+ stream.writeMap(map, type, this, ctx);
return stream.lastFinished();
}
@@ -485,8 +488,14 @@ public ByteBuffer getBuffer() {
* @param consumer Consumer.
* @param isNull {@code True} if message is null.
* @param stream Byte buffer stream.
+ * @param ctx Serialization context.
*/
- private void writeCompressedMessage(Consumer consumer, boolean isNull, DirectByteBufferStream stream) {
+ private void writeCompressedMessage(
+ Consumer consumer,
+ boolean isNull,
+ DirectByteBufferStream stream,
+ MessageSerializationContext ctx
+ ) {
if (isNull) {
stream.writeShort(Short.MIN_VALUE);
@@ -536,7 +545,7 @@ private void writeCompressedMessage(Consumer consumer, bool
stream.serializeFinished(true);
}
- stream.writeMessage(stream.compressedMessage(), this);
+ stream.writeMessage(stream.compressedMessage(), this, ctx);
if (stream.lastFinished()) {
stream.compressedMessage(null);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/IgniteMessageSerializationContext.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/IgniteMessageSerializationContext.java
new file mode 100644
index 0000000000000..5b27320ce27b6
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/IgniteMessageSerializationContext.java
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.direct;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.MessageSerializationContext;
+import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteComponentFeatureSet;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeature;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeatureSet;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteNodeFeatureSet;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.spi.discovery.tcp.internal.UnsupportedNodeVersionException;
+import org.jetbrains.annotations.Nullable;
+
+/** */
+public class IgniteMessageSerializationContext implements MessageSerializationContext {
+ /** */
+ @GridToStringInclude
+ private final Map ctxByComponent;
+
+ /** */
+ private IgniteMessageSerializationContext(Map ctxByComponent) {
+ this.ctxByComponent = ctxByComponent;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean includeFieldIntroducedBy(IgniteFeature feature) {
+ return componentContext(feature).includeFieldIntroducedBy(feature.id());
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean includeFieldDeprecatedBy(IgniteFeature feature) {
+ return componentContext(feature).includeFieldDeprecatedBy(feature.id());
+ }
+
+ /** */
+ private ComponentMessageSerializationContext componentContext(IgniteFeature feature) {
+ ComponentMessageSerializationContext cmpCtx = ctxByComponent.get(feature.componentName());
+
+ if (cmpCtx == null) {
+ throw new IllegalStateException(
+ "A field is guarded by a feature of an undeclared component" +
+ " [feature=" + feature +
+ ", component=" + feature.componentName() +
+ ", declaredComponents=" + ctxByComponent.keySet() + ']'
+ );
+ }
+
+ return cmpCtx;
+ }
+
+ /** */
+ public static IgniteMessageSerializationContext buildForPeers(
+ Ignite loc,
+ ClusterNode rmt
+ ) throws UnsupportedNodeVersionException, ClusterTopologyCheckedException {
+ GridKernalContext ctx = ((IgniteEx)loc).context();
+
+ return buildForPeers(ctx.localNodeFeatures(), ctx.discovery().resolveNodeFeatures(rmt));
+ }
+
+ /** */
+ public static IgniteMessageSerializationContext buildForPeers(
+ IgniteNodeFeatureSet loc,
+ IgniteNodeFeatureSet rmt
+ ) throws UnsupportedNodeVersionException {
+ assert loc != null;
+
+ if (rmt == null) {
+ throw new UnsupportedNodeVersionException("Failed to build the message serialization context for the remote node." +
+ " The remote node's feature set is unavailable.");
+ }
+
+ Set components = new HashSet<>(loc.components());
+
+ components.addAll(rmt.components());
+
+ Map ctxByComponent = new HashMap<>();
+
+ for (String cmp : components) {
+ ComponentMessageSerializationContext ctx = resolveComponentSerializationContext(
+ cmp,
+ loc.componentFeatures(cmp),
+ rmt.componentFeatures(cmp)
+ );
+
+ ctxByComponent.put(cmp, ctx);
+ }
+
+ return new IgniteMessageSerializationContext(ctxByComponent);
+ }
+
+ /** */
+ public static MessageSerializationContext buildForInitiator(@Nullable IgniteNodeFeatureSet initiatorFeatures) {
+ if (initiatorFeatures == null)
+ return UNNEGOTIATED;
+
+ Map ctxByComponent = new HashMap<>();
+
+ for (IgniteComponentFeatureSet cmpFeatures : initiatorFeatures.values()) {
+ ctxByComponent.put(
+ cmpFeatures.componentName(),
+ new ComponentMessageSerializationContext(cmpFeatures.features(), cmpFeatures.features()));
+ }
+
+ return new IgniteMessageSerializationContext(ctxByComponent);
+ }
+
+ /** */
+ private static ComponentMessageSerializationContext resolveComponentSerializationContext(
+ String cmpName,
+ @Nullable IgniteComponentFeatureSet locCmpFeatures,
+ @Nullable IgniteComponentFeatureSet rmtCmpFeatures
+ ) throws UnsupportedNodeVersionException {
+ assert locCmpFeatures != null || rmtCmpFeatures != null;
+
+ // One of the sides has no component configured. This may happen when one side uses an RU-unaware plugin version
+ // while the other uses an RU-aware version. In this case, all newly introduced fields are skipped, while all
+ // deprecated fields are included.
+ if (locCmpFeatures == null || rmtCmpFeatures == null)
+ return new ComponentMessageSerializationContext(null, null);
+
+ int c = locCmpFeatures.version().compareTo(rmtCmpFeatures.version());
+
+ if (c == 0) {
+ assert locCmpFeatures.features().equals(rmtCmpFeatures.features());
+
+ // Both newly introduced and deprecated fields are included. During an RU, a node builds messages according
+ // to both the old logical version (while RU is in progress, deprecated fields are used and newly introduced
+ // fields are not) and the new logical version (after RU is finished, newly introduced fields are used and
+ // deprecated fields are not).
+ return new ComponentMessageSerializationContext(null, rmtCmpFeatures.features());
+ }
+ else {
+ IgniteComponentFeatureSet src = c < 0 ? locCmpFeatures : rmtCmpFeatures;
+ IgniteComponentFeatureSet target = c < 0 ? rmtCmpFeatures : locCmpFeatures;
+
+ if (!src.isUpgradableTo(target)) {
+ throw new UnsupportedNodeVersionException("Remote node component versions are not supported" +
+ " [component=" + cmpName +
+ ", locComponent=" + locCmpFeatures +
+ ", rmtComponent=" + rmtCmpFeatures + ']');
+ }
+
+ // The old version dictates the serialization rules.
+ return new ComponentMessageSerializationContext(src.features(), src.features());
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(IgniteMessageSerializationContext.class, this);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ return Objects.equals(ctxByComponent, ((IgniteMessageSerializationContext)o).ctxByComponent);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return Objects.hashCode(ctxByComponent);
+ }
+
+ /** */
+ private static final class ComponentMessageSerializationContext {
+ /** */
+ @GridToStringInclude
+ @Nullable private final IgniteFeatureSet excludedDeprecatedFields;
+
+ /** */
+ @GridToStringInclude
+ @Nullable private final IgniteFeatureSet includedIntroducedFields;
+
+ /** */
+ private ComponentMessageSerializationContext(
+ @Nullable IgniteFeatureSet excludedDeprecatedFields,
+ @Nullable IgniteFeatureSet includedIntroducedFields
+ ) {
+ this.excludedDeprecatedFields = excludedDeprecatedFields;
+ this.includedIntroducedFields = includedIntroducedFields;
+ }
+
+ /** */
+ boolean includeFieldIntroducedBy(int featureId) {
+ return includedIntroducedFields != null && includedIntroducedFields.contains(featureId);
+ }
+
+ /** */
+ boolean includeFieldDeprecatedBy(int featureId) {
+ return excludedDeprecatedFields == null || !excludedDeprecatedFields.contains(featureId);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ ComponentMessageSerializationContext other = (ComponentMessageSerializationContext)o;
+
+ return Objects.equals(excludedDeprecatedFields, other.excludedDeprecatedFields)
+ && Objects.equals(includedIntroducedFields, other.includedIntroducedFields);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return Objects.hash(includedIntroducedFields, excludedDeprecatedFields);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(ComponentMessageSerializationContext.class, this);
+ }
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/DirectByteBufferStream.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/DirectByteBufferStream.java
index d2176dfb6f617..cdabbaa7c5699 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/DirectByteBufferStream.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/DirectByteBufferStream.java
@@ -34,6 +34,7 @@
import java.util.function.Supplier;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.binary.StringWriter;
import org.apache.ignite.internal.managers.communication.CompressedMessage;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -932,11 +933,12 @@ public void writeGridLongList(@Nullable GridLongList val) {
/**
* @param msg Message.
* @param writer Writer.
+ * @param ctx Serialization context.
*/
- public void writeMessage(Message msg, MessageWriter writer) {
+ public void writeMessage(Message msg, MessageWriter writer, MessageSerializationContext ctx) {
if (msg != null) {
if (buf.hasRemaining())
- nestedWrite(writer, () -> MessageSerialization.writeTo(msgFactory, msg, writer));
+ nestedWrite(writer, () -> MessageSerialization.writeTo(msgFactory, msg, writer, ctx));
else
lastFinished = false;
}
@@ -948,8 +950,9 @@ public void writeMessage(Message msg, MessageWriter writer) {
* @param arr Array.
* @param type Type.
* @param writer Writer.
+ * @param ctx Serialization context.
*/
- public void writeObjectArray(T[] arr, MessageArrayType type, MessageWriter writer) {
+ public void writeObjectArray(T[] arr, MessageArrayType type, MessageWriter writer, MessageSerializationContext ctx) {
if (arr != null) {
int len = arr.length;
@@ -966,7 +969,7 @@ public void writeObjectArray(T[] arr, MessageArrayType type, MessageWriter w
if (arrCur == NULL)
arrCur = arr[arrPos++];
- write(type.valueType(), arrCur, writer);
+ write(type.valueType(), arrCur, writer, ctx);
if (!lastFinished)
return;
@@ -984,11 +987,12 @@ public void writeObjectArray(T[] arr, MessageArrayType type, MessageWriter w
* @param col Collection.
* @param type Type.
* @param writer Writer.
+ * @param ctx Serialization context.
*/
- public void writeCollection(Collection col, MessageCollectionType type, MessageWriter writer) {
+ public void writeCollection(Collection col, MessageCollectionType type, MessageWriter writer, MessageSerializationContext ctx) {
if (col != null) {
if (col instanceof List && col instanceof RandomAccess)
- writeRandomAccessList((List)col, type, writer);
+ writeRandomAccessList((List)col, type, writer, ctx);
else {
if (it == null) {
writeInt(col.size());
@@ -1003,7 +1007,7 @@ public void writeCollection(Collection col, MessageCollectionType type, M
if (cur == NULL)
cur = it.next();
- write(type.valueType(), cur, writer);
+ write(type.valueType(), cur, writer, ctx);
if (!lastFinished)
return;
@@ -1022,8 +1026,14 @@ public void writeCollection(Collection col, MessageCollectionType type, M
* @param list List.
* @param type Type.
* @param writer Writer.
- */
- private void writeRandomAccessList(List list, MessageCollectionType type, MessageWriter writer) {
+ * @param ctx Serialization context.
+ */
+ private void writeRandomAccessList(
+ List list,
+ MessageCollectionType type,
+ MessageWriter writer,
+ MessageSerializationContext ctx
+ ) {
assert list instanceof RandomAccess;
int size = list.size();
@@ -1041,7 +1051,7 @@ private void writeRandomAccessList(List list, MessageCollectionType type,
if (arrCur == NULL)
arrCur = list.get(arrPos++);
- write(type.valueType(), arrCur, writer);
+ write(type.valueType(), arrCur, writer, ctx);
if (!lastFinished)
return;
@@ -1056,8 +1066,9 @@ private void writeRandomAccessList(List list, MessageCollectionType type,
* @param map Map.
* @param type Type.
* @param writer Writer.
+ * @param ctx Serialization context.
*/
- public void writeMap(Map map, MessageMapType type, MessageWriter writer) {
+ public void writeMap(Map map, MessageMapType type, MessageWriter writer, MessageSerializationContext ctx) {
if (map != null) {
if (mapIt == null) {
writeInt(map.size());
@@ -1077,7 +1088,7 @@ public void writeMap(Map map, MessageMapType type, MessageWriter wr
e = (Map.Entry)mapCur;
if (!keyDone) {
- write(type.keyType(), e.getKey(), writer);
+ write(type.keyType(), e.getKey(), writer, ctx);
if (!lastFinished)
return;
@@ -1085,7 +1096,7 @@ public void writeMap(Map map, MessageMapType type, MessageWriter wr
keyDone = true;
}
- write(type.valueType(), e.getValue(), writer);
+ write(type.valueType(), e.getValue(), writer, ctx);
if (!lastFinished)
return;
@@ -1561,9 +1572,10 @@ public GridLongList readGridLongList() {
/**
* @param reader Reader.
+ * @param ctx Serialization context.
* @return Message.
*/
- public T readMessage(MessageReader reader) {
+ public T readMessage(MessageReader reader, MessageSerializationContext ctx) {
if (!msgTypeDone) {
if (buf.remaining() < Message.DIRECT_TYPE_SIZE) {
lastFinished = false;
@@ -1582,7 +1594,7 @@ public T readMessage(MessageReader reader) {
try {
reader.beforeNestedRead();
- lastFinished = MessageSerialization.readFrom(msgFactory, msg, reader);
+ lastFinished = MessageSerialization.readFrom(msgFactory, msg, reader, ctx);
}
finally {
reader.afterNestedRead(lastFinished);
@@ -1606,9 +1618,10 @@ public T readMessage(MessageReader reader) {
/**
* @param type Item type.
* @param reader Reader.
+ * @param ctx Serialization context.
* @return Array.
*/
- public T[] readObjectArray(MessageArrayType type, MessageReader reader) {
+ public T[] readObjectArray(MessageArrayType type, MessageReader reader, MessageSerializationContext ctx) {
if (readSize == -1) {
int size = readInt();
@@ -1623,7 +1636,7 @@ public T[] readObjectArray(MessageArrayType type, MessageReader reader) {
objArr = type.clazz() != null ? (Object[])Array.newInstance(type.clazz(), readSize) : new Object[readSize];
for (int i = readItems; i < readSize; i++) {
- Object item = read(type.valueType(), reader);
+ Object item = read(type.valueType(), reader, ctx);
if (!lastFinished)
return null;
@@ -1650,9 +1663,10 @@ public T[] readObjectArray(MessageArrayType type, MessageReader reader) {
*
* @param type Item type.
* @param reader Reader.
+ * @param ctx Serialization context.
* @return {@link ArrayList}, {@link HashSet} or {@link EnumSet}.
*/
- public > C readCollection(MessageCollectionType type, MessageReader reader) {
+ public > C readCollection(MessageCollectionType type, MessageReader reader, MessageSerializationContext ctx) {
if (readSize == -1) {
int size = readInt();
@@ -1667,7 +1681,7 @@ public > C readCollection(MessageCollectionType type, Me
col = newCollection(type);
for (int i = readItems; i < readSize; i++) {
- Object item = read(type.valueType(), reader);
+ Object item = read(type.valueType(), reader, ctx);
if (!lastFinished)
return null;
@@ -1702,9 +1716,10 @@ private Collection