You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by mp...@apache.org on 2026/09/30 14:51:02 UTC
(ignite) branch master updated: IGNITE-28851 Introduced Message Serialization Context for RU-aware transport between Ignite nodes (#13532)
This is an automated email from the ASF dual-hosted git repository.
petrov-mg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 7820b6c40b0 IGNITE-28851 Introduced Message Serialization Context for RU-aware transport between Ignite nodes (#13532)
7820b6c40b0 is described below
commit 7820b6c40b0be40f0661a152b88d9c0801c36a11
Author: Mikhail Petrov <32...@users.noreply.github.com>
AuthorDate: Wed Sep 30 17:50:56 2026 +0300
IGNITE-28851 Introduced Message Serialization Context for RU-aware transport between Ignite nodes (#13532)
---
.../direct/JmhDirectMessageReaderBenchmark.java | 5 +-
.../internal/{Order.java => FeatureGated.java} | 34 +-
.../apache/ignite/internal/MessageProcessor.java | 108 +++++
.../internal/MessageSerializerGenerator.java | 104 ++++-
.../java/org/apache/ignite/internal/Order.java | 27 +-
.../internal/MessageSerializationContext.java | 90 ++++
.../rollingupgrade/feature/IgniteFeature.java | 0
.../internal/direct/DirectMessageReader.java | 31 +-
.../internal/direct/DirectMessageWriter.java | 37 +-
.../direct/IgniteMessageSerializationContext.java | 231 ++++++++++
.../direct/stream/DirectByteBufferStream.java | 81 ++--
.../communication/CompressedMessageSerializer.java | 5 +-
.../managers/discovery/GridDiscoveryManager.java | 24 +
...atureRegistry.java => CoreFeatureRegistry.java} | 4 +-
.../feature/IgniteComponentFeatureSet.java | 5 +
.../feature/IgniteCoreFeatureSet.java | 2 +-
.../tcp/internal/ClusterStateProvider.java | 5 +
.../tcp/internal/GridNioServerWrapper.java | 9 +-
.../tcp/internal/InboundConnectionHandler.java | 20 +
.../tcp/internal/TcpHandshakeExecutor.java | 7 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 8 +-
.../ignite/spi/discovery/tcp/ServerImpl.java | 15 +-
.../spi/discovery/tcp/TcpDiscoveryIoSession.java | 21 +-
.../ignite/spi/discovery/tcp/TcpDiscoverySpi.java | 33 --
.../tcp/internal/ClientMessageHolder.java | 15 +-
.../internal/TcpDiscoveryMessageSerializer.java | 18 +-
.../internal/codegen/MessageProcessorTest.java | 40 ++
.../direct/DirectMarshallingMessagesTest.java | 5 +-
.../managers/AbstractMessageSerializationTest.java | 22 +-
.../communication/CompressedMessageTest.java | 15 +-
.../GridIoManagerOrderedUnmarshalFailureTest.java | 5 +-
.../cache/CacheMetricsCacheSizeTest.java | 6 +-
...niteCacheContinuousQueryImmutableEntryTest.java | 5 +-
.../QueryEntityMessageSerializationTest.java | 5 +-
.../feature/TestPluginReleaseFeatures_2_0_0.java | 3 +
.../feature/TestPluginReleaseFeatures_2_1_0.java | 5 +-
.../RollingUpgradeMessageSerializationTest.java | 510 +++++++++++++++++++++
.../rollingupgrade/message/TestCoreMessage.java | 107 +++++
.../message/TestDefaultRegistryMessage.java | 104 +++++
.../TestMessage.java} | 38 +-
.../rollingupgrade/message/TestPluginMessage.java | 107 +++++
...rviceConfigurationMessageSerializationTest.java | 5 +-
.../DistributedProcessResultMarshallingTest.java | 9 +-
.../nio/IgniteExceptionInNioWorkerSelfTest.java | 5 +-
.../communication/MessageMarshalOnceTest.java | 9 +-
...ecurityBasicPermissionSetSerializationTest.java | 5 +-
.../tcp/TestDelayMessageSerializer.java | 5 +-
.../tcp/DiscoveryUnmarshalVulnerabilityTest.java | 12 +-
.../apache/ignite/testframework/GridTestNode.java | 33 +-
.../junits/common/GridCommonAbstractTest.java | 4 +-
.../ignite/testsuites/IgniteBasicTestSuite.java | 2 +
.../resources/codegen/ChildMessageSerializer.java | 5 +-
.../codegen/CorrectEmptyMessageSerializer.java | 5 +-
.../CustomMapperEnumFieldsMessageSerializer.java | 9 +-
.../DefaultMapperEnumFieldsMessageSerializer.java | 13 +-
.../codegen/TestCollectionsMessageSerializer.java | 109 ++---
.../codegen/TestEnumSetMessageSerializer.java | 17 +-
.../codegen/TestFeatureConflictMessage.java} | 9 +-
.../codegen/TestFeatureRegistry.java} | 11 +-
.../codegen/TestInvalidFeatureMessage.java} | 10 +-
.../codegen/TestInvalidFeatureRegistry.java} | 6 +-
.../codegen/TestMapMessageSerializer.java | 109 ++---
.../codegen/TestMarshallableMessageSerializer.java | 5 +-
.../TestMarshalledArrayMapMessageSerializer.java | 21 +-
.../TestMarshalledCollectionMessageSerializer.java | 9 +-
.../TestMarshalledMapMessageSerializer.java | 13 +-
.../codegen/TestMarshalledMessageSerializer.java | 5 +-
.../TestMarshalledObjectsMessageSerializer.java | 9 +-
.../resources/codegen/TestMessageSerializer.java | 21 +-
.../codegen/TestRollingUpgradeAwareMessage.java} | 26 +-
.../TestRollingUpgradeAwareMessageSerializer.java | 127 +++++
.../codegen/TestUnknownFeatureMessage.java} | 9 +-
.../ignite/internal/util/nio/GridDirectParser.java | 3 +-
.../ignite/internal/util/nio/GridNioServer.java | 36 +-
.../internal/util/nio/GridNioSessionMetaKey.java | 5 +-
.../internal/util/nio/MessageSerialization.java | 33 +-
.../extensions/communication/MessageReader.java | 23 +-
.../communication/MessageSerializer.java | 8 +-
.../extensions/communication/MessageWriter.java | 23 +-
.../zk/internal/DiscoveryMessageParser.java | 6 +-
80 files changed, 2243 insertions(+), 447 deletions(-)
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 7c93e2be90b..22701785a65 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.State;
import org.openjdk.jmh.annotations.Warmup;
import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.MessageSerializationContext.LATEST_SCHEMA;
import static org.openjdk.jmh.annotations.Mode.Throughput;
/** Benchmarks the {@link DirectMessageReader} compressed-field hot path. */
@@ -96,7 +97,7 @@ public class JmhDirectMessageReaderBenchmark {
writer.setBuffer(buf);
- boolean finished = writer.writeMessage(msg, true);
+ boolean finished = writer.writeMessage(msg, true, LATEST_SCHEMA);
if (!finished)
throw new IllegalStateException("Message does not fit into the buffer.");
@@ -111,7 +112,7 @@ public class JmhDirectMessageReaderBenchmark {
reader.setBuffer(buf);
- Message msg = reader.readMessage(true);
+ Message msg = reader.readMessage(true, LATEST_SCHEMA);
reader.reset();
diff --git a/modules/codegen/src/main/java/org/apache/ignite/internal/Order.java b/modules/codegen/src/main/java/org/apache/ignite/internal/FeatureGated.java
similarity index 50%
copy from modules/codegen/src/main/java/org/apache/ignite/internal/Order.java
copy to modules/codegen/src/main/java/org/apache/ignite/internal/FeatureGated.java
index 0e4562537c4..82d646b8160 100644
--- a/modules/codegen/src/main/java/org/apache/ignite/internal/Order.java
+++ b/modules/codegen/src/main/java/org/apache/ignite/internal/FeatureGated.java
@@ -21,21 +21,29 @@ 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;
/**
- * The annotation specifies the position of a field in the serialized and deserialized byte sequence of a {@code Message} class.
- * <p>
- * 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.
- * <p> 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).
- * <p> 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.
+ * Optional complement to the {@link Order} annotations of a class, telling where the {@link IgniteFeature}s named in
+ * {@link Order#introducedBy()} and {@link Order#deprecatedBy()} are declared.
+ *
+ * <p>Feature names are resolved as constants of the {@link #registry()} class. Without this annotation they are resolved
+ * against the Ignite Core Feature Registry, so it is needed primarily by classes whose fields are gated by features of
+ * another registry, such as the features of a plugin:</p>
+ * <pre>
+ * @FeatureGated(registry = MyPluginFeatures.class)
+ * public class MyPluginMessage implements Message {
+ * @Order(value = 0, introducedBy = "MY_FEATURE")
+ * String fld;
+ * }
+ * </pre>
+ *
+ * @see Order
+ * @see IgniteFeature
*/
@Retention(RetentionPolicy.CLASS)
-@Target(ElementType.FIELD)
-public @interface Order {
- /** @return Order of the field. */
- int value();
+@Target(ElementType.TYPE)
+public @interface FeatureGated {
+ /** @return Class of the feature registry, or {@link Void} if the Ignite Core Feature Registry is used. */
+ Class<?> registry() default Void.class;
}
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 8b4ead738ee..b5edc8553a9 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.Modifier;
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 javax.tools.Diagnostic;
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 feature gates against unless it declares one with {@link FeatureGated#registry()}. */
+ static final String DFLT_FEATURE_REG_CLS =
+ "org.apache.ignite.internal.processors.rollingupgrade.feature.CoreFeatureRegistry";
+
+ /** */
+ 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";
@@ -425,4 +435,102 @@ public class MessageProcessor extends AbstractProcessor {
TypeElement typeElement = elementUtils.getTypeElement(clazz);
return typeElement != null ? typeElement.asType() : null;
}
+
+ /** */
+ @Nullable public static FieldFeatureGate buildFieldFeatureGate(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<String> 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 FieldFeatureGate(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<Modifier> 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) {
+ FeatureGated ann = cls.getAnnotation(FeatureGated.class);
+
+ if (ann == null)
+ return DFLT_FEATURE_REG_CLS;
+
+ String regCls;
+
+ try {
+ regCls = ann.registry().getName();
+ }
+ catch (MirroredTypeException e) {
+ regCls = qualifiedClassName(e.getTypeMirror());
+ }
+
+ return Void.class.getName().equals(regCls) ? DFLT_FEATURE_REG_CLS : regCls;
+ }
+
+ /** */
+ public record FieldFeatureGate(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 81680f1b108..5956fdbbfa1 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.DeclaredType;
import javax.lang.model.type.PrimitiveType;
import javax.lang.model.type.TypeKind;
import javax.lang.model.type.TypeMirror;
+import org.apache.ignite.internal.MessageProcessor.FieldFeatureGate;
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.COMPRESSED_MESSAGE_CLA
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.buildFieldFeatureGate;
/** 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 @@ public class MessageSerializerGenerator extends MessageCompanionGenerator {
private void generateMethod(List<String> code, List<VariableElement> 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 @@ public class MessageSerializerGenerator extends MessageCompanionGenerator {
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 @@ public class MessageSerializerGenerator extends MessageCompanionGenerator {
* @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++;
+ FieldFeatureGate gate = buildFieldFeatureGate(env, field);
+
+ if (gate != null) {
+ imports.add(gate.registry());
+
+ write.add(indentedLine("if (%s) {", gate.expression()));
+
+ indent++;
+ }
+
returnFalseIf(write, "!" + writeExpr);
+ if (gate != null) {
+ indent--;
+
+ write.add(indentedLine("}"));
+ }
+
write.add(EMPTY);
write.add(indentedLine("writer.incrementState();"));
write.add(EMPTY);
@@ -335,11 +359,27 @@ public class MessageSerializerGenerator extends MessageCompanionGenerator {
indent++;
+ FieldFeatureGate gate = buildFieldFeatureGate(env, field);
+
+ if (gate != null) {
+ imports.add(gate.registry());
+
+ read.add(indentedLine("if (%s) {", gate.expression()));
+
+ indent++;
+ }
+
read.add(indentedLine("%s = %s;", fieldRef(field), readExpr));
read.add(EMPTY);
returnFalseIf(read, "!reader.isLastRead()");
+ if (gate != null) {
+ indent--;
+
+ read.add(indentedLine("}"));
+ }
+
read.add(EMPTY);
read.add(indentedLine("reader.incrementState();"));
read.add(EMPTY);
@@ -360,13 +400,13 @@ public class MessageSerializerGenerator extends MessageCompanionGenerator {
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 @@ public class MessageSerializerGenerator extends MessageCompanionGenerator {
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 @@ public class MessageSerializerGenerator extends MessageCompanionGenerator {
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 @@ public class MessageSerializerGenerator extends MessageCompanionGenerator {
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 0e4562537c4..03cec3b0e05 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.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;
/**
* The annotation specifies the position of a field in the serialized and deserialized byte sequence of a {@code Message} class.
@@ -28,14 +29,38 @@ import java.lang.annotation.Target;
* 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.
* <p> 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).
* <p> 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 FeatureGated
*/
@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.
+ *
+ * <p>An annotated field is included in message serialization only when doing so does not break backward compatibility
+ * during a Rolling Upgrade.</p>
+ *
+ * @return Name of the Ignite feature that introduced this field, or an empty string if the field is not gated.
+ */
+ String introducedBy() default "";
+
+ /**
+ * {@link IgniteFeature} that deprecated the field marked with the current annotation.
+ *
+ * <p>Deprecation means that the field is planned for removal in a future release.</p>
+ *
+ * <p>An annotated field is excluded from message serialization only when doing so does not break backward compatibility
+ * during a Rolling Upgrade.</p>
+ *
+ * @return Name of the Ignite feature that deprecated this field, or an empty string if the field is not gated.
+ */
+ String deprecatedBy() default "";
}
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 00000000000..3bb16926a05
--- /dev/null
+++ b/modules/commons/src/main/java/org/apache/ignite/internal/MessageSerializationContext.java
@@ -0,0 +1,90 @@
+/*
+ * 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 latest message fields schema: all newly introduced fields are included, and all deprecated fields are
+ * excluded.
+ */
+ MessageSerializationContext LATEST_SCHEMA = 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 [LATEST_SCHEMA]";
+ }
+ };
+
+ /**
+ * {@link MessageSerializationContext} implementation that permits only messages whose schema is immutable across
+ * versions, i.e. declares no fields gated by an {@link IgniteFeature}: evaluating any feature gate throws
+ * {@link IllegalStateException}.
+ *
+ * <p>Used between connection establishment and serialization protocol negotiation, when the peer's features are not
+ * yet known. Messages exchanged during this period must therefore declare no fields gated by
+ * {@code @Order(introducedBy = ...)} or {@code @Order(deprecatedBy = ...)}.</p>
+ */
+ MessageSerializationContext IMMUTABLE_SCHEMA = 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 [IMMUTABLE_SCHEMA]";
+ }
+
+ /** */
+ private IllegalStateException buildError(IgniteFeature feature) {
+ return new IllegalStateException(
+ "A feature-gated 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/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 9ec9b68cc18..b070629a38b 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.Map;
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 class DirectMessageReader implements MessageReader {
}
/** {@inheritDoc} */
- @Nullable @Override public <T extends Message> T readMessage(boolean compress) {
+ @Nullable @Override public <T extends Message> T readMessage(boolean compress, MessageSerializationContext ctx) {
DirectByteBufferStream stream = curStream;
T msg;
@@ -352,10 +353,11 @@ public class DirectMessageReader implements MessageReader {
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 class DirectMessageReader implements MessageReader {
}
/** {@inheritDoc} */
- @Override public <T> T[] readObjectArray(MessageArrayType type) {
+ @Override public <T> 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 class DirectMessageReader implements MessageReader {
}
/** {@inheritDoc} */
- @Override public <C extends Collection<?>> C readCollection(MessageCollectionType type) {
+ @Override public <C extends Collection<?>> 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 class DirectMessageReader implements MessageReader {
}
/** {@inheritDoc} */
- @Override public <M extends Map<?, ?>> M readMap(MessageMapType type, boolean compress) {
+ @Override public <M extends Map<?, ?>> M readMap(MessageMapType type, boolean compress, MessageSerializationContext ctx) {
DirectByteBufferStream stream = curStream;
M map;
@@ -427,10 +429,11 @@ public class DirectMessageReader implements MessageReader {
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 class DirectMessageReader implements MessageReader {
}
/** @return Deserialized object. */
- private <T> T readCompressedMessageAndDeserialize(DirectByteBufferStream stream, Function<DirectMessageReader, T> fun) {
- Message msg = stream.readMessage(this);
+ private <T> T readCompressedMessageAndDeserialize(
+ DirectByteBufferStream stream,
+ Function<DirectMessageReader, T> 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 319aae7b7e9..f09632cfbac 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.Collection;
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 class DirectMessageWriter implements MessageWriter {
}
/** {@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 class DirectMessageWriter implements MessageWriter {
}
/** {@inheritDoc} */
- @Override public <T> boolean writeObjectArray(T[] arr, MessageArrayType type) {
+ @Override public <T> 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 <T> boolean writeCollection(Collection<T> col, MessageCollectionType type) {
+ @Override public <T> boolean writeCollection(Collection<T> 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 <K, V> boolean writeMap(Map<K, V> map, MessageMapType type, boolean compress) {
+ @Override public <K, V> boolean writeMap(Map<K, V> 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 class DirectMessageWriter implements MessageWriter {
* @param consumer Consumer.
* @param isNull {@code True} if message is null.
* @param stream Byte buffer stream.
+ * @param ctx Serialization context.
*/
- private void writeCompressedMessage(Consumer<DirectMessageWriter> consumer, boolean isNull, DirectByteBufferStream stream) {
+ private void writeCompressedMessage(
+ Consumer<DirectMessageWriter> consumer,
+ boolean isNull,
+ DirectByteBufferStream stream,
+ MessageSerializationContext ctx
+ ) {
if (isNull) {
stream.writeShort(Short.MIN_VALUE);
@@ -536,7 +545,7 @@ public class DirectMessageWriter implements MessageWriter {
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 00000000000..f4248588ced
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/IgniteMessageSerializationContext.java
@@ -0,0 +1,231 @@
+/*
+ * 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<String, ComponentMessageSerializationContext> ctxByComponent;
+
+ /** */
+ private IgniteMessageSerializationContext(Map<String, ComponentMessageSerializationContext> ctxByComponent) {
+ this.ctxByComponent = ctxByComponent;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean includeFieldIntroducedBy(IgniteFeature feature) {
+ return componentContext(feature.componentName()).includeFieldIntroducedBy(feature.id());
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean includeFieldDeprecatedBy(IgniteFeature feature) {
+ return componentContext(feature.componentName()).includeFieldDeprecatedBy(feature.id());
+ }
+
+ /** */
+ private ComponentMessageSerializationContext componentContext(String cmpName) {
+ ComponentMessageSerializationContext cmpCtx = ctxByComponent.get(cmpName);
+
+ if (cmpCtx == null) {
+ throw new IllegalStateException("A field is gated by a feature of an undeclared component" +
+ " [component=" + cmpName +
+ ", 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<String> components = new HashSet<>(loc.components());
+
+ components.addAll(rmt.components());
+
+ Map<String, ComponentMessageSerializationContext> 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);
+ }
+
+ /** */
+ 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 d2176dfb6f6..cdabbaa7c56 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.BooleanSupplier;
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 class DirectByteBufferStream {
/**
* @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 class DirectByteBufferStream {
* @param arr Array.
* @param type Type.
* @param writer Writer.
+ * @param ctx Serialization context.
*/
- public <T> void writeObjectArray(T[] arr, MessageArrayType type, MessageWriter writer) {
+ public <T> void writeObjectArray(T[] arr, MessageArrayType type, MessageWriter writer, MessageSerializationContext ctx) {
if (arr != null) {
int len = arr.length;
@@ -966,7 +969,7 @@ public class DirectByteBufferStream {
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 class DirectByteBufferStream {
* @param col Collection.
* @param type Type.
* @param writer Writer.
+ * @param ctx Serialization context.
*/
- public <T> void writeCollection(Collection<T> col, MessageCollectionType type, MessageWriter writer) {
+ public <T> void writeCollection(Collection<T> col, MessageCollectionType type, MessageWriter writer, MessageSerializationContext ctx) {
if (col != null) {
if (col instanceof List && col instanceof RandomAccess)
- writeRandomAccessList((List<T>)col, type, writer);
+ writeRandomAccessList((List<T>)col, type, writer, ctx);
else {
if (it == null) {
writeInt(col.size());
@@ -1003,7 +1007,7 @@ public class DirectByteBufferStream {
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 class DirectByteBufferStream {
* @param list List.
* @param type Type.
* @param writer Writer.
- */
- private <T> void writeRandomAccessList(List<T> list, MessageCollectionType type, MessageWriter writer) {
+ * @param ctx Serialization context.
+ */
+ private <T> void writeRandomAccessList(
+ List<T> list,
+ MessageCollectionType type,
+ MessageWriter writer,
+ MessageSerializationContext ctx
+ ) {
assert list instanceof RandomAccess;
int size = list.size();
@@ -1041,7 +1051,7 @@ public class DirectByteBufferStream {
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 @@ public class DirectByteBufferStream {
* @param map Map.
* @param type Type.
* @param writer Writer.
+ * @param ctx Serialization context.
*/
- public <K, V> void writeMap(Map<K, V> map, MessageMapType type, MessageWriter writer) {
+ public <K, V> void writeMap(Map<K, V> map, MessageMapType type, MessageWriter writer, MessageSerializationContext ctx) {
if (map != null) {
if (mapIt == null) {
writeInt(map.size());
@@ -1077,7 +1088,7 @@ public class DirectByteBufferStream {
e = (Map.Entry<K, V>)mapCur;
if (!keyDone) {
- write(type.keyType(), e.getKey(), writer);
+ write(type.keyType(), e.getKey(), writer, ctx);
if (!lastFinished)
return;
@@ -1085,7 +1096,7 @@ public class DirectByteBufferStream {
keyDone = true;
}
- write(type.valueType(), e.getValue(), writer);
+ write(type.valueType(), e.getValue(), writer, ctx);
if (!lastFinished)
return;
@@ -1561,9 +1572,10 @@ public class DirectByteBufferStream {
/**
* @param reader Reader.
+ * @param ctx Serialization context.
* @return Message.
*/
- public <T extends Message> T readMessage(MessageReader reader) {
+ public <T extends Message> T readMessage(MessageReader reader, MessageSerializationContext ctx) {
if (!msgTypeDone) {
if (buf.remaining() < Message.DIRECT_TYPE_SIZE) {
lastFinished = false;
@@ -1582,7 +1594,7 @@ public class DirectByteBufferStream {
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 class DirectByteBufferStream {
/**
* @param type Item type.
* @param reader Reader.
+ * @param ctx Serialization context.
* @return Array.
*/
- public <T> T[] readObjectArray(MessageArrayType type, MessageReader reader) {
+ public <T> T[] readObjectArray(MessageArrayType type, MessageReader reader, MessageSerializationContext ctx) {
if (readSize == -1) {
int size = readInt();
@@ -1623,7 +1636,7 @@ public class DirectByteBufferStream {
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 class DirectByteBufferStream {
*
* @param type Item type.
* @param reader Reader.
+ * @param ctx Serialization context.
* @return {@link ArrayList}, {@link HashSet} or {@link EnumSet}.
*/
- public <C extends Collection<?>> C readCollection(MessageCollectionType type, MessageReader reader) {
+ public <C extends Collection<?>> C readCollection(MessageCollectionType type, MessageReader reader, MessageSerializationContext ctx) {
if (readSize == -1) {
int size = readInt();
@@ -1667,7 +1681,7 @@ public class DirectByteBufferStream {
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 @@ public class DirectByteBufferStream {
/**
* @param type Value type.
* @param reader Reader.
+ * @param ctx Serialization context.
* @return Map.
*/
- public <M extends Map<?, ?>> M readMap(MessageMapType type, MessageReader reader) {
+ public <M extends Map<?, ?>> M readMap(MessageMapType type, MessageReader reader, MessageSerializationContext ctx) {
if (readSize == -1) {
int size = readInt();
@@ -1720,7 +1735,7 @@ public class DirectByteBufferStream {
for (int i = readItems; i < readSize; i++) {
if (!keyDone) {
- Object key = read(type.keyType(), reader);
+ Object key = read(type.keyType(), reader, ctx);
if (!lastFinished)
return null;
@@ -1729,7 +1744,7 @@ public class DirectByteBufferStream {
keyDone = true;
}
- Object val = read(type.valueType(), reader);
+ Object val = read(type.valueType(), reader, ctx);
if (!lastFinished)
return null;
@@ -2009,8 +2024,9 @@ public class DirectByteBufferStream {
* @param type Type.
* @param val Value.
* @param writer Writer.
+ * @param ctx Serialization context.
*/
- protected <K, V> void write(MessageType type, Object val, MessageWriter writer) {
+ protected <K, V> void write(MessageType type, Object val, MessageWriter writer, MessageSerializationContext ctx) {
switch (type.type()) {
case BYTE:
writeByte((Byte)val);
@@ -2138,17 +2154,17 @@ public class DirectByteBufferStream {
break;
case MAP:
- nestedWrite(writer, () -> writer.writeMap((Map<K, V>)val, (MessageMapType)type));
+ nestedWrite(writer, () -> writer.writeMap((Map<K, V>)val, (MessageMapType)type, ctx));
break;
case COLLECTION:
- nestedWrite(writer, () -> writer.writeCollection((Collection<V>)val, (MessageCollectionType)type));
+ nestedWrite(writer, () -> writer.writeCollection((Collection<V>)val, (MessageCollectionType)type, ctx));
break;
case ARRAY:
- nestedWrite(writer, () -> writer.writeObjectArray((V[])val, (MessageArrayType)type));
+ nestedWrite(writer, () -> writer.writeObjectArray((V[])val, (MessageArrayType)type, ctx));
break;
@@ -2158,7 +2174,7 @@ public class DirectByteBufferStream {
break;
case MSG:
- writeMessage((Message)val, writer);
+ writeMessage((Message)val, writer, ctx);
break;
@@ -2182,9 +2198,10 @@ public class DirectByteBufferStream {
/**
* @param type Type.
* @param reader Reader.
+ * @param ctx Serialization context.
* @return Value.
*/
- protected Object read(MessageType type, MessageReader reader) {
+ protected Object read(MessageType type, MessageReader reader, MessageSerializationContext ctx) {
switch (type.type()) {
case BYTE:
return readByte();
@@ -2262,19 +2279,19 @@ public class DirectByteBufferStream {
return readGridLongList();
case MAP:
- return nestedRead(reader, () -> reader.readMap((MessageMapType)type));
+ return nestedRead(reader, () -> reader.readMap((MessageMapType)type, ctx));
case COLLECTION:
- return nestedRead(reader, () -> reader.readCollection((MessageCollectionType)type));
+ return nestedRead(reader, () -> reader.readCollection((MessageCollectionType)type, ctx));
case ARRAY:
- return nestedRead(reader, () -> reader.readObjectArray((MessageArrayType)type));
+ return nestedRead(reader, () -> reader.readObjectArray((MessageArrayType)type, ctx));
case ENUM:
return ((MessageEnumType)type).decode(readByte());
case MSG:
- return readMessage(reader);
+ return readMessage(reader, ctx);
default:
throw new IllegalArgumentException("Unknown type: " + type);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/CompressedMessageSerializer.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/CompressedMessageSerializer.java
index 7661a114219..4f96b37f88e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/CompressedMessageSerializer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/CompressedMessageSerializer.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.managers.communication;
import java.util.ArrayList;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
@@ -28,7 +29,7 @@ import static org.apache.ignite.internal.managers.communication.CompressedMessag
/** Message serializer for compressed message. */
public class CompressedMessageSerializer implements MessageSerializer<CompressedMessage> {
/** {@inheritDoc} */
- @Override public boolean writeTo(CompressedMessage msg, MessageWriter writer) {
+ @Override public boolean writeTo(CompressedMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -74,7 +75,7 @@ public class CompressedMessageSerializer implements MessageSerializer<Compressed
}
/** {@inheritDoc} */
- @Override public boolean readFrom(CompressedMessage msg, MessageReader reader) {
+ @Override public boolean readFrom(CompressedMessage msg, MessageReader reader, MessageSerializationContext ctx) {
assert msg.chunk == null : msg.chunk;
while (true) {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
index 009bbfa0dc2..d8f9a4e428f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
@@ -71,6 +71,7 @@ import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.NodeStoppingException;
+import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.NodeOrderComparator;
import org.apache.ignite.internal.events.DiscoveryCustomEvent;
import org.apache.ignite.internal.managers.GridManagerAdapter;
@@ -88,6 +89,7 @@ import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMess
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
import org.apache.ignite.internal.processors.cluster.IGridClusterStateProcessor;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteNodeFeatureSet;
import org.apache.ignite.internal.processors.security.SecurityContext;
import org.apache.ignite.internal.systemview.ClusterNodeViewWalker;
import org.apache.ignite.internal.systemview.NodeAttributeViewWalker;
@@ -1814,6 +1816,28 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
}
+ /** */
+ public @Nullable IgniteNodeFeatureSet resolveNodeFeatures(ClusterNode node) throws ClusterTopologyCheckedException {
+ assert node instanceof IgniteClusterNode : node;
+
+ IgniteNodeFeatureSet features = ((IgniteClusterNode)node).features();
+
+ if (features != null)
+ return features;
+
+ ClusterNode resolvedNode = node(node.id());
+
+ if (resolvedNode == null)
+ resolvedNode = historicalNode(node.id());
+
+ if (resolvedNode == null)
+ throw new ClusterTopologyCheckedException("Failed to resolve the remote node by ID [nodeId=" + node.id() + ']');
+
+ assert resolvedNode instanceof IgniteClusterNode : resolvedNode;
+
+ return ((IgniteClusterNode)resolvedNode).features();
+ }
+
/**
* @param nodeId ID of the node.
* @return Node for ID.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/SupportedFeatureRegistry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/CoreFeatureRegistry.java
similarity index 97%
rename from modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/SupportedFeatureRegistry.java
rename to modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/CoreFeatureRegistry.java
index 7b3e55b85d3..92b7eeb1a8c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/SupportedFeatureRegistry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/CoreFeatureRegistry.java
@@ -20,7 +20,7 @@ package org.apache.ignite.internal.processors.rollingupgrade.feature;
import org.apache.ignite.internal.processors.rollingupgrade.RollingUpgradeProcessor;
/**
- * Declares the {@link IgniteFeature}s supported by the current Ignite version.
+ * Declares the {@link IgniteCoreFeature}s supported by the current Ignite version.
* The complete set of declared {@link IgniteFeature}s defines the behavior of a node running the current version
* and determines whether a Rolling Upgrade is supported between the current Ignite version and older Ignite versions.
*
@@ -90,7 +90,7 @@ import org.apache.ignite.internal.processors.rollingupgrade.RollingUpgradeProces
* @see IgniteFeatureManager
* @see RollingUpgradeProcessor
*/
-public class SupportedFeatureRegistry {
+public class CoreFeatureRegistry {
/** */
public static final IgniteFeature ROLLING_UPGRADE_FEATURE = new IgniteCoreFeature(0);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteComponentFeatureSet.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteComponentFeatureSet.java
index b51bd6e1851..99d0bbc6a3f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteComponentFeatureSet.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteComponentFeatureSet.java
@@ -62,6 +62,11 @@ public abstract class IgniteComponentFeatureSet implements Message, Externalizab
return ver;
}
+ /** */
+ public IgniteFeatureSet features() {
+ return features;
+ }
+
/** */
public boolean contains(int featureId) {
return features.contains(featureId);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteCoreFeatureSet.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteCoreFeatureSet.java
index f5173872ef0..7df54d449be 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteCoreFeatureSet.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/IgniteCoreFeatureSet.java
@@ -31,7 +31,7 @@ public class IgniteCoreFeatureSet extends IgniteComponentFeatureSet {
/** */
static IgniteCoreFeatureSet INSTANCE = new IgniteCoreFeatureSet(
IgniteVersionUtils.VER,
- IgniteFeatureSet.buildFrom(SupportedFeatureRegistry.class)
+ IgniteFeatureSet.buildFrom(CoreFeatureRegistry.class)
);
/** */
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ClusterStateProvider.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ClusterStateProvider.java
index b75bc033d9a..d4383c096ce 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ClusterStateProvider.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ClusterStateProvider.java
@@ -96,6 +96,11 @@ public class ClusterStateProvider {
return disconnected;
}
+ /** */
+ public Ignite ignite() {
+ return ignite;
+ }
+
/**
* @return {@code True} if ssl enabled.
*/
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/GridNioServerWrapper.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/GridNioServerWrapper.java
index 7874adb5629..9368f9f8a83 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/GridNioServerWrapper.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/GridNioServerWrapper.java
@@ -103,9 +103,13 @@ import org.apache.ignite.spi.communication.tcp.messages.HandshakeWaitMessageSeri
import org.apache.ignite.spi.communication.tcp.messages.NodeIdMessage;
import org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage;
import org.apache.ignite.spi.discovery.IgniteDiscoveryThread;
+import org.apache.ignite.spi.discovery.tcp.internal.UnsupportedNodeVersionException;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.internal.direct.IgniteMessageSerializationContext.buildForPeers;
import static org.apache.ignite.internal.thread.pool.IgniteScheduledThreadPoolExecutor.newSingleThreadScheduledExecutor;
+import static org.apache.ignite.internal.util.nio.GridNioServer.RECOVERY_DESC_META_KEY;
+import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.MSG_SER_CTX;
import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.SSL_META;
import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.COMMUNICATION_METRICS_GROUP_NAME;
import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.CONN_IDX_META;
@@ -525,7 +529,8 @@ public class GridNioServerWrapper {
meta.put(CONSISTENT_ID_META, node.consistentId());
meta.put(CONN_IDX_META, connKey);
- meta.put(GridNioServer.RECOVERY_DESC_META_KEY, recoveryDesc);
+ meta.put(RECOVERY_DESC_META_KEY, recoveryDesc);
+ meta.put(MSG_SER_CTX.ordinal(), buildForPeers(stateProvider.ignite(), node));
ses = nioSrv.createSession(ch, meta, false, null).get();
}
@@ -574,7 +579,7 @@ public class GridNioServerWrapper {
break;
}
}
- catch (ClusterTopologyCheckedException e) {
+ catch (ClusterTopologyCheckedException | UnsupportedNodeVersionException e) {
throw e;
}
catch (Exception e) {
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/InboundConnectionHandler.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/InboundConnectionHandler.java
index 9aa8bd8f4d4..fe3feb4cb2f 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/InboundConnectionHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/InboundConnectionHandler.java
@@ -29,6 +29,7 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.failure.FailureContext;
import org.apache.ignite.failure.FailureType;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpi;
import org.apache.ignite.internal.processors.failure.FailureProcessor;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -54,8 +55,12 @@ import org.apache.ignite.spi.communication.tcp.messages.NodeIdMessage;
import org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage;
import org.apache.ignite.spi.discovery.DiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.internal.UnsupportedNodeVersionException;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.internal.MessageSerializationContext.IMMUTABLE_SCHEMA;
+import static org.apache.ignite.internal.direct.IgniteMessageSerializationContext.buildForPeers;
+import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.MSG_SER_CTX;
import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.CONN_IDX_META;
import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.CONSISTENT_ID_META;
import static org.apache.ignite.spi.communication.tcp.internal.CommunicationTcpUtils.NOOP;
@@ -210,6 +215,9 @@ public class InboundConnectionHandler extends GridNioServerListenerAdapter<Messa
/** {@inheritDoc} */
@Override public void onConnected(GridNioSession ses) {
+ if (ses.meta(MSG_SER_CTX.ordinal()) == null)
+ ses.addMeta(MSG_SER_CTX.ordinal(), IMMUTABLE_SCHEMA);
+
if (ses.accepted()) {
if (log.isInfoEnabled()) {
log.info("Accepted incoming communication connection [locAddr=" + ses.localAddress() +
@@ -512,6 +520,18 @@ public class InboundConnectionHandler extends GridNioServerListenerAdapter<Messa
assert old == null;
+ try {
+ ses.addMeta(MSG_SER_CTX.ordinal(), buildForPeers(igniteExSupplier.get(), rmtNode));
+ }
+ catch (UnsupportedNodeVersionException | ClusterTopologyCheckedException e) {
+ U.warn(log, "Failed to resolve message serialization context for remote node session" +
+ " [rmtNodeId=" + sndId + ", ses=" + ses + ", errMsg=" + e.getMessage() + ']');
+
+ ses.close();
+
+ return;
+ }
+
ClusterNode locNode = locNodeSupplier.get();
if (ses.remoteAddress() == null)
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpHandshakeExecutor.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpHandshakeExecutor.java
index c08a16844e3..073c500efee 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpHandshakeExecutor.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpHandshakeExecutor.java
@@ -39,6 +39,7 @@ import org.apache.ignite.spi.communication.tcp.messages.NodeIdMessage;
import org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.internal.MessageSerializationContext.IMMUTABLE_SCHEMA;
import static org.apache.ignite.internal.util.CommonUtils.makeMessageType;
import static org.apache.ignite.plugin.extensions.communication.Message.DIRECT_TYPE_SIZE;
import static org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage.NEED_WAIT;
@@ -172,7 +173,7 @@ public class TcpHandshakeExecutor {
NodeIdMessage nodeIdMsg = new NodeIdMessage();
reader.setBuffer(buf);
- MessageSerialization.readFrom(msgFactory, nodeIdMsg, reader);
+ MessageSerialization.readFrom(msgFactory, nodeIdMsg, reader, IMMUTABLE_SCHEMA);
reader.reset();
return nodeIdMsg.nodeId();
@@ -191,7 +192,7 @@ public class TcpHandshakeExecutor {
writer.setBuffer(buf);
- MessageSerialization.writeTo(msgFactory, msg, writer);
+ MessageSerialization.writeTo(msgFactory, msg, writer, IMMUTABLE_SCHEMA);
buf.flip();
@@ -241,7 +242,7 @@ public class TcpHandshakeExecutor {
reader.setBuffer(buf);
- fininshed = MessageSerialization.readFrom(msgFactory, msg, reader);
+ fininshed = MessageSerialization.readFrom(msgFactory, msg, reader, IMMUTABLE_SCHEMA);
readPos = buf.position();
}
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
index dd1aed7a2a2..344c97ea74f 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
@@ -666,8 +666,10 @@ class ClientImpl extends TcpDiscoveryImpl {
* @param addr Address.
* @return Socket, connect response and client acknowledge support flag.
*/
- @Nullable private T2<TcpDiscoveryIoSession, Integer> sendJoinRequest(boolean recon,
- InetSocketAddress addr) throws InterruptedException {
+ @Nullable private T2<TcpDiscoveryIoSession, Integer> sendJoinRequest(
+ boolean recon,
+ InetSocketAddress addr
+ ) throws InterruptedException {
assert addr != null;
if (log.isDebugEnabled())
@@ -711,7 +713,7 @@ class ClientImpl extends TcpDiscoveryImpl {
TcpDiscoveryHandshakeResponse res = spi.readHandshakeResponse(ses, ackTimeout0);
- spi.validateRemoteFeatures(res.nodeFeatures());
+ ses.applyMessageSerializationContext(res.nodeFeatures());
// Convert the addresses once.
Collection<InetSocketAddress> redirectAddrs = res.redirectAddresses();
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
index aa6e42a0236..7ec09237569 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
@@ -1509,7 +1509,7 @@ class ServerImpl extends TcpDiscoveryImpl {
TcpDiscoveryHandshakeResponse res = spi.readHandshakeResponse(ses, timeoutHelper.nextTimeoutChunk(ackTimeout0));
- spi.validateRemoteFeatures(res.nodeFeatures());
+ ses.applyMessageSerializationContext(res.nodeFeatures());
if (msg instanceof TcpDiscoveryJoinRequestMessage) {
boolean ignore = false;
@@ -3244,7 +3244,7 @@ class ServerImpl extends TcpDiscoveryImpl {
: new ClientMessageHolder(rebuiltMsg);
try {
- msgToSend.serialize(cliMsgSer);
+ msgToSend.serialize(cliMsgSer, worker.session().serializationContext());
}
catch (IgniteCheckedException e) {
U.error(log, "Failed to serialize message: " + msgToSend, e);
@@ -3426,7 +3426,7 @@ class ServerImpl extends TcpDiscoveryImpl {
if (log.isDebugEnabled())
log.debug("Handshake response: " + res);
- spi.validateRemoteFeatures(res.nodeFeatures());
+ ses.applyMessageSerializationContext(res.nodeFeatures());
// We should take previousNodeAlive flag into account
// only if we received the response from the correct node.
@@ -6641,7 +6641,7 @@ class ServerImpl extends TcpDiscoveryImpl {
spi.writeMessage(ses, res, spi.getEffectiveSocketTimeout(srvSock));
- spi.validateRemoteFeatures(req.nodeFeatures());
+ ses.applyMessageSerializationContext(req.nodeFeatures());
// It can happen if a remote node is stopped and it has a loopback address in the list of addresses,
// the local node sends a handshake request message on the loopback address, so we get here.
@@ -7499,6 +7499,11 @@ class ServerImpl extends TcpDiscoveryImpl {
lastMetricsUpdateMsgTimeNanos = System.nanoTime();
}
+ /** */
+ TcpDiscoveryIoSession session() {
+ return ses;
+ }
+
/**
* @param clientVer Client version.
*/
@@ -7619,7 +7624,7 @@ class ServerImpl extends TcpDiscoveryImpl {
* @param timeout Timeout.
*/
private void writeMessage(ClientMessageHolder msgHolder, long timeout) throws IgniteCheckedException, IOException {
- byte[] msgBytes = msgHolder.messageBytes();
+ byte[] msgBytes = msgHolder.messageBytes(ses.serializationContext());
if (msgBytes != null)
spi.write(ses, msgBytes, timeout);
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java
index 4b81a151fd7..741b7fc8b0b 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java
@@ -35,9 +35,12 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.direct.DirectMessageReader;
+import org.apache.ignite.internal.direct.IgniteMessageSerializationContext;
import org.apache.ignite.internal.managers.communication.DiscoveryMarshalling;
import org.apache.ignite.internal.managers.communication.UnknownMessageException;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteNodeFeatureSet;
import org.apache.ignite.internal.util.CommonUtils;
import org.apache.ignite.internal.util.nio.MessageSerialization;
import org.apache.ignite.internal.util.typedef.X;
@@ -47,6 +50,7 @@ import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryMessageSerializer;
+import org.apache.ignite.spi.discovery.tcp.internal.UnsupportedNodeVersionException;
import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -103,6 +107,9 @@ public class TcpDiscoveryIoSession implements AutoCloseable {
/** Buffered socket input stream. */
private final CompositeInputStream in;
+ /** */
+ private volatile MessageSerializationContext serCtx = MessageSerializationContext.IMMUTABLE_SCHEMA;
+
/**
* Creates a new discovery I/O session bound to the given socket.
*
@@ -133,6 +140,16 @@ public class TcpDiscoveryIoSession implements AutoCloseable {
}
}
+ /** */
+ void applyMessageSerializationContext(@Nullable IgniteNodeFeatureSet rmtFeatures) throws UnsupportedNodeVersionException {
+ serCtx = IgniteMessageSerializationContext.buildForPeers(ctx.localNodeFeatures(), rmtFeatures);
+ }
+
+ /** @return Serialization context the two nodes of this session agreed on. */
+ public MessageSerializationContext serializationContext() {
+ return serCtx;
+ }
+
/**
* Writes a discovery message to the underlying socket output stream.
* Refer to the class description for the rationale behind synchronized access.
@@ -142,7 +159,7 @@ public class TcpDiscoveryIoSession implements AutoCloseable {
*/
synchronized void writeMessage(TcpDiscoveryAbstractMessage msg) throws IgniteCheckedException, IOException {
try {
- msgSer.writeTo(msg, out);
+ msgSer.writeTo(msg, out, serCtx);
out.flush();
}
@@ -214,7 +231,7 @@ public class TcpDiscoveryIoSession implements AutoCloseable {
readBuf.limit(read);
- finished = MessageSerialization.readFrom(msgFactory, msg, msgReader);
+ finished = MessageSerialization.readFrom(msgFactory, msg, msgReader, serCtx);
// Server Discovery only sends next message to next Server upon receiving a receipt for the previous one.
// This behaviour guarantees that we never read a next message from the buffer right after the end of
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
index 17352d1f72b..19903ffb5fb 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
@@ -58,8 +58,6 @@ import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.managers.communication.UnknownMessageException;
import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpi;
import org.apache.ignite.internal.processors.metric.MetricRegistryImpl;
-import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteComponentFeatureSet;
-import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteNodeFeatureSet;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
@@ -98,7 +96,6 @@ import org.apache.ignite.spi.discovery.DiscoverySpiOrderSupport;
import org.apache.ignite.spi.discovery.tcp.internal.DiscoveryDataPacket;
import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode;
import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryStatistics;
-import org.apache.ignite.spi.discovery.tcp.internal.UnsupportedNodeVersionException;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.jdbc.TcpDiscoveryJdbcIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder;
@@ -1681,36 +1678,6 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter implements IgniteDiscovery
}
}
- /** */
- void validateRemoteFeatures(IgniteNodeFeatureSet rmtFeatures) throws IgniteCheckedException {
- if (rmtFeatures == null) {
- throw new UnsupportedNodeVersionException(
- "Failed to obtain remote node features. The remote node may be running an unsupported Ignite version," +
- " which may result in unexpected handshake message serialization");
- }
-
- for (IgniteComponentFeatureSet rmtCmpFeatures : rmtFeatures.values()) {
- IgniteComponentFeatureSet locCmpFeatures = locNode.features().componentFeatures(rmtCmpFeatures.componentName());
-
- if (locCmpFeatures == null)
- continue;
-
- int c = locCmpFeatures.version().compareTo(rmtCmpFeatures.version());
-
- if (c == 0)
- continue;
-
- IgniteComponentFeatureSet src = c > 0 ? rmtCmpFeatures : locCmpFeatures;
- IgniteComponentFeatureSet target = c > 0 ? locCmpFeatures : rmtCmpFeatures;
-
- if (!src.isUpgradableTo(target)) {
- throw new UnsupportedNodeVersionException("Remote node component versions are not supported" +
- " [locComponents=" + locNode.features() +
- ", rmtComponents=" + rmtFeatures + ']');
- }
- }
- }
-
/**
* Writes raw data to the session socket.
*
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/ClientMessageHolder.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/ClientMessageHolder.java
index 1652824dd15..047da61b2d2 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/ClientMessageHolder.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/ClientMessageHolder.java
@@ -17,7 +17,10 @@
package org.apache.ignite.spi.discovery.tcp.internal;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
import org.jetbrains.annotations.Nullable;
@@ -27,7 +30,7 @@ public class ClientMessageHolder {
private final TcpDiscoveryAbstractMessage msg;
/** */
- private volatile byte[] msgBytes;
+ private final Map<MessageSerializationContext, byte[]> bytesByCtx = new ConcurrentHashMap<>(1);
/** */
public ClientMessageHolder(TcpDiscoveryAbstractMessage msg) {
@@ -42,14 +45,14 @@ public class ClientMessageHolder {
}
/** */
- public byte @Nullable [] messageBytes() {
- return msgBytes;
+ public byte @Nullable [] messageBytes(MessageSerializationContext ctx) {
+ return bytesByCtx.get(ctx);
}
/** */
- public void serialize(TcpDiscoveryMessageSerializer ser) throws IgniteCheckedException {
- if (msgBytes == null)
- msgBytes = ser.serialize(msg);
+ public void serialize(TcpDiscoveryMessageSerializer ser, MessageSerializationContext ctx) throws IgniteCheckedException {
+ if (!bytesByCtx.containsKey(ctx))
+ bytesByCtx.put(ctx, ser.serialize(msg, ctx));
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryMessageSerializer.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryMessageSerializer.java
index 45aa1d9b767..9102876206b 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryMessageSerializer.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryMessageSerializer.java
@@ -22,6 +22,7 @@ import java.io.OutputStream;
import java.nio.ByteBuffer;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.direct.DirectMessageWriter;
import org.apache.ignite.internal.managers.communication.DiscoveryMarshalling;
import org.apache.ignite.internal.util.io.GridByteArrayOutputStream;
@@ -54,10 +55,15 @@ public class TcpDiscoveryMessageSerializer {
*
* @param msg Discovery message to serialize.
* @param out Output stream to write serialized message.
+ * @param serCtx Serialization context the recipient agreed on.
* @throws IgniteCheckedException If serialization fails.
* @throws IOException If serialization fails.
*/
- public void writeTo(TcpDiscoveryAbstractMessage msg, OutputStream out) throws IgniteCheckedException, IOException {
+ public void writeTo(
+ TcpDiscoveryAbstractMessage msg,
+ OutputStream out,
+ MessageSerializationContext serCtx
+ ) throws IgniteCheckedException, IOException {
DiscoveryMarshalling.marshal(msg, ctx, null);
writer.reset();
@@ -69,7 +75,7 @@ public class TcpDiscoveryMessageSerializer {
// Should be cleared before first operation.
buf.clear();
- finished = MessageSerialization.writeTo(ctx.messageFactory(), msg, writer);
+ finished = MessageSerialization.writeTo(ctx.messageFactory(), msg, writer, serCtx);
out.write(buf.array(), 0, buf.position());
}
@@ -80,12 +86,16 @@ public class TcpDiscoveryMessageSerializer {
* Serializes a discovery message into a byte array.
*
* @param msg Discovery message to serialize.
+ * @param serCtx Serialization context the recipient agreed on.
* @return Serialized byte array containing the message data.
* @throws IgniteCheckedException If serialization fails.
*/
- public byte[] serialize(TcpDiscoveryAbstractMessage msg) throws IgniteCheckedException {
+ public byte[] serialize(
+ TcpDiscoveryAbstractMessage msg,
+ MessageSerializationContext serCtx
+ ) throws IgniteCheckedException {
try (GridByteArrayOutputStream out = new GridByteArrayOutputStream()) {
- writeTo(msg, out);
+ writeTo(msg, out, serCtx);
return out.toByteArray();
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/codegen/MessageProcessorTest.java b/modules/core/src/test/java/org/apache/ignite/internal/codegen/MessageProcessorTest.java
index 1105f85cf3f..862b2d88b65 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/codegen/MessageProcessorTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/codegen/MessageProcessorTest.java
@@ -93,6 +93,46 @@ public class MessageProcessorTest {
.hasSourceEquivalentTo(javaFile("TestMessageMarshaller.java"));
}
+ /** */
+ @Test
+ public void testRollingUpgradeAwareMessage() {
+ Compilation compilation = compile("TestFeatureRegistry.java", "TestRollingUpgradeAwareMessage.java");
+
+ assertThat(compilation).succeeded();
+
+ assertThat(compilation)
+ .generatedSourceFile("org.apache.ignite.internal.TestRollingUpgradeAwareMessageSerializer")
+ .hasSourceEquivalentTo(javaFile("TestRollingUpgradeAwareMessageSerializer.java"));
+ }
+
+ /** */
+ @Test
+ public void testUnknownFeatureConstantRejected() {
+ Compilation compilation = compile("TestUnknownFeatureMessage.java");
+
+ assertThat(compilation).failed();
+ assertThat(compilation).hadErrorContaining("Failed to resolve feature in the registry by its name [reg=");
+ assertThat(compilation).hadErrorContaining(", feature=NO_SUCH_FEATURE]");
+ }
+
+ /** */
+ @Test
+ public void testSameFeatureInBothGatesRejected() {
+ Compilation compilation = compile("TestFeatureConflictMessage.java");
+
+ assertThat(compilation).failed();
+ assertThat(compilation).hadErrorContaining("must not reference the same feature");
+ }
+
+ /** */
+ @Test
+ public void testFeatureConstantOfWrongTypeRejected() {
+ Compilation compilation = compile("TestInvalidFeatureRegistry.java", "TestInvalidFeatureMessage.java");
+
+ assertThat(compilation).failed();
+ assertThat(compilation).hadErrorContaining("must be of type IgniteFeature [reg=");
+ }
+
/** */
@Test
public void testCollectionsMessage() {
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/direct/DirectMarshallingMessagesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/direct/DirectMarshallingMessagesTest.java
index e9874e03055..5eab512eb6e 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/direct/DirectMarshallingMessagesTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/direct/DirectMarshallingMessagesTest.java
@@ -31,6 +31,7 @@ import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.apache.ignite.transactions.TransactionIsolation;
import org.junit.Test;
+import static org.apache.ignite.internal.MessageSerializationContext.LATEST_SCHEMA;
import static org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE;
@@ -139,7 +140,7 @@ public class DirectMarshallingMessagesTest extends GridCommonAbstractTest {
writer.setBuffer(chunk);
- fullyWritten = writer.writeMessage(srcMsg, false);
+ fullyWritten = writer.writeMessage(srcMsg, false, LATEST_SCHEMA);
chunk.flip();
@@ -168,7 +169,7 @@ public class DirectMarshallingMessagesTest extends GridCommonAbstractTest {
reader.setBuffer(chunk);
- resMsg = reader.readMessage(false);
+ resMsg = reader.readMessage(false, LATEST_SCHEMA);
pos += chunk.position();
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/AbstractMessageSerializationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/managers/AbstractMessageSerializationTest.java
index bfa3c8e3142..88c11a10eb3 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/managers/AbstractMessageSerializationTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/managers/AbstractMessageSerializationTest.java
@@ -25,6 +25,7 @@ import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.managers.communication.IgniteMessageFactoryImpl;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheObject;
@@ -47,6 +48,7 @@ import org.jetbrains.annotations.Nullable;
import org.junit.Test;
import static java.lang.Integer.MAX_VALUE;
+import static org.apache.ignite.internal.MessageSerializationContext.LATEST_SCHEMA;
import static org.apache.ignite.plugin.extensions.communication.CollectionImplementationType.HASH_SET;
import static org.junit.Assert.assertEquals;
@@ -106,13 +108,13 @@ public abstract class AbstractMessageSerializationTest {
initializeMessage(msg);
- while (!MessageSerialization.writeTo(msgFactory, msg, writer)) {
+ while (!MessageSerialization.writeTo(msgFactory, msg, writer, LATEST_SCHEMA)) {
// No-op.
}
msg = msgFactory.create(msgType);
- while (!MessageSerialization.readFrom(msgFactory, msg, reader)) {
+ while (!MessageSerialization.readFrom(msgFactory, msg, reader, LATEST_SCHEMA)) {
// No-op.
}
@@ -295,22 +297,22 @@ public abstract class AbstractMessageSerializationTest {
}
/** {@inheritDoc} */
- @Override public boolean writeMessage(Message val, boolean compress) {
+ @Override public boolean writeMessage(Message val, boolean compress, MessageSerializationContext ctx) {
return writeField(Message.class);
}
/** {@inheritDoc} */
- @Override public <T> boolean writeObjectArray(T[] arr, MessageArrayType type) {
+ @Override public <T> boolean writeObjectArray(T[] arr, MessageArrayType type, MessageSerializationContext ctx) {
return writeField(Object[].class);
}
/** {@inheritDoc} */
- @Override public <T> boolean writeCollection(Collection<T> col, MessageCollectionType type) {
+ @Override public <T> boolean writeCollection(Collection<T> col, MessageCollectionType type, MessageSerializationContext ctx) {
return writeField(type.collectionImplementationType() == HASH_SET ? Set.class : Collection.class);
}
/** {@inheritDoc} */
- @Override public <K, V> boolean writeMap(Map<K, V> map, MessageMapType type, boolean compress) {
+ @Override public <K, V> boolean writeMap(Map<K, V> map, MessageMapType type, boolean compress, MessageSerializationContext ctx) {
return writeField(type.linked() ? LinkedHashMap.class : HashMap.class);
}
@@ -537,7 +539,7 @@ public abstract class AbstractMessageSerializationTest {
}
/** {@inheritDoc} */
- @Override public <T extends Message> T readMessage(boolean compress) {
+ @Override public <T extends Message> T readMessage(boolean compress, MessageSerializationContext ctx) {
readField(Message.class);
return null;
@@ -565,21 +567,21 @@ public abstract class AbstractMessageSerializationTest {
}
/** {@inheritDoc} */
- @Override public <T> T[] readObjectArray(MessageArrayType type) {
+ @Override public <T> T[] readObjectArray(MessageArrayType type, MessageSerializationContext ctx) {
readField(Object[].class);
return null;
}
/** {@inheritDoc} */
- @Override public <C extends Collection<?>> C readCollection(MessageCollectionType type) {
+ @Override public <C extends Collection<?>> C readCollection(MessageCollectionType type, MessageSerializationContext ctx) {
readField(type.collectionImplementationType() == HASH_SET ? Set.class : Collection.class);
return null;
}
/** {@inheritDoc} */
- @Override public <M extends Map<?, ?>> M readMap(MessageMapType type, boolean compress) {
+ @Override public <M extends Map<?, ?>> M readMap(MessageMapType type, boolean compress, MessageSerializationContext ctx) {
readField(type.linked() ? LinkedHashMap.class : HashMap.class);
return null;
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/CompressedMessageTest.java b/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/CompressedMessageTest.java
index 046dbbb6f55..2f9c77e4299 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/CompressedMessageTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/CompressedMessageTest.java
@@ -42,6 +42,7 @@ import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
import org.apache.ignite.testframework.GridTestUtils;
import org.junit.Test;
+import static org.apache.ignite.internal.MessageSerializationContext.LATEST_SCHEMA;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -70,7 +71,7 @@ public class CompressedMessageTest {
ByteBuffer msgBuf = ByteBuffer.allocate(40_960);
while (!finished) {
- finished = writer.writeMessage(fullMsg, true);
+ finished = writer.writeMessage(fullMsg, true, LATEST_SCHEMA);
if (checkChunkCnt) {
DirectMessageState<?> state = U.field(writer, "state");
@@ -104,7 +105,7 @@ public class CompressedMessageTest {
reader.setBuffer(msgBuf);
- Message readMsg = reader.readMessage(true);
+ Message readMsg = reader.readMessage(true, LATEST_SCHEMA);
assertTrue(readMsg instanceof GridDhtPartitionsFullMessage);
@@ -133,7 +134,7 @@ public class CompressedMessageTest {
reader.setBuffer(buf);
GridTestUtils.assertThrows(null,
- () -> MessageSerialization.readFrom(MSG_FACTORY, new CompressedMessage(), reader),
+ () -> MessageSerialization.readFrom(MSG_FACTORY, new CompressedMessage(), reader, LATEST_SCHEMA),
IgniteException.class,
"unexpected null chunk");
}
@@ -156,7 +157,7 @@ public class CompressedMessageTest {
reader.setBuffer(buf);
GridTestUtils.assertThrows(null,
- () -> MessageSerialization.readFrom(MSG_FACTORY, new CompressedMessage(), reader),
+ () -> MessageSerialization.readFrom(MSG_FACTORY, new CompressedMessage(), reader, LATEST_SCHEMA),
IgniteException.class,
"Invalid compressed message data size");
}
@@ -217,7 +218,7 @@ public class CompressedMessageTest {
writer.setBuffer(tmpBuf);
- assertTrue(writer.writeMessage(fullMessage(), false));
+ assertTrue(writer.writeMessage(fullMessage(), false, LATEST_SCHEMA));
tmpBuf.flip();
@@ -231,7 +232,7 @@ public class CompressedMessageTest {
wireWriter.setBuffer(wire);
- assertTrue(wireWriter.writeMessage(compressedMsg, false));
+ assertTrue(wireWriter.writeMessage(compressedMsg, false, LATEST_SCHEMA));
wire.flip();
@@ -239,7 +240,7 @@ public class CompressedMessageTest {
reader.setBuffer(wire);
- GridTestUtils.assertThrows(null, () -> reader.readMessage(true), IgniteException.class, "ended unexpectedly");
+ GridTestUtils.assertThrows(null, () -> reader.readMessage(true, LATEST_SCHEMA), IgniteException.class, "ended unexpectedly");
}
/** */
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerOrderedUnmarshalFailureTest.java b/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerOrderedUnmarshalFailureTest.java
index 4b51337eac0..2a5ce91b963 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerOrderedUnmarshalFailureTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerOrderedUnmarshalFailureTest.java
@@ -28,6 +28,7 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.CoreMessagesProvider;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.processors.cache.CacheObjectContext;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.marshaller.Marshaller;
@@ -186,7 +187,7 @@ public class GridIoManagerOrderedUnmarshalFailureTest extends GridCommonAbstract
/** Writes the two fields behind the header. */
private static class Serializer implements MessageSerializer<FailingUnmarshalMessage> {
/** {@inheritDoc} */
- @Override public boolean writeTo(FailingUnmarshalMessage msg, MessageWriter writer) {
+ @Override public boolean writeTo(FailingUnmarshalMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -212,7 +213,7 @@ public class GridIoManagerOrderedUnmarshalFailureTest extends GridCommonAbstract
}
/** {@inheritDoc} */
- @Override public boolean readFrom(FailingUnmarshalMessage msg, MessageReader reader) {
+ @Override public boolean readFrom(FailingUnmarshalMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
msg.seq = reader.readInt();
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheMetricsCacheSizeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheMetricsCacheSizeTest.java
index a4a0db43330..bd321cb6736 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheMetricsCacheSizeTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheMetricsCacheSizeTest.java
@@ -39,6 +39,8 @@ import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryMetricsUpdateMes
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
+import static org.apache.ignite.internal.MessageSerializationContext.LATEST_SCHEMA;
+
/**
* This test checks metrics cacheSize.
* <ul>
@@ -121,7 +123,7 @@ public class CacheMetricsCacheSizeTest extends GridCommonAbstractTest {
// 2kb should be enough for an empty message even if it is a relatively large metrics message.
msgWritter.setBuffer(ByteBuffer.allocate(2048));
- assertTrue(MessageSerialization.writeTo(msgFactory, msg, msgWritter));
+ assertTrue(MessageSerialization.writeTo(msgFactory, msg, msgWritter, LATEST_SCHEMA));
assertTrue(msgWritter.getBuffer().hasRemaining());
@@ -133,7 +135,7 @@ public class CacheMetricsCacheSizeTest extends GridCommonAbstractTest {
TcpDiscoveryMetricsUpdateMessage msg2 = new TcpDiscoveryMetricsUpdateMessage();
- assertTrue(MessageSerialization.readFrom(msgFactory, msg2, msgReader));
+ assertTrue(MessageSerialization.readFrom(msgFactory, msg2, msgReader, LATEST_SCHEMA));
Map<Integer, CacheMetricsMessage> cacheMetrics2 = msg2.serversFullMetricsMessages().values().iterator().next()
.cachesMetricsMessages();
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java
index 45a5fb3381f..572c733e826 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java
@@ -45,6 +45,7 @@ import org.junit.Test;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.internal.MessageSerializationContext.LATEST_SCHEMA;
/**
*
@@ -204,7 +205,7 @@ public class IgniteCacheContinuousQueryImmutableEntryTest extends GridCommonAbst
// Skip write class header.
writer.onHeaderWritten();
- MessageSerialization.writeTo(msgFactory, e, writer);
+ MessageSerialization.writeTo(msgFactory, e, writer, LATEST_SCHEMA);
CacheContinuousQueryEntry res = new CacheContinuousQueryEntry();
@@ -212,7 +213,7 @@ public class IgniteCacheContinuousQueryImmutableEntryTest extends GridCommonAbst
reader.setBuffer(ByteBuffer.wrap(buf.array()));
- MessageSerialization.readFrom(msgFactory, res, reader);
+ MessageSerialization.readFrom(msgFactory, res, reader, LATEST_SCHEMA);
return res;
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/query/schema/message/QueryEntityMessageSerializationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/query/schema/message/QueryEntityMessageSerializationTest.java
index f7fea391425..0397c89a65a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/query/schema/message/QueryEntityMessageSerializationTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/query/schema/message/QueryEntityMessageSerializationTest.java
@@ -48,6 +48,7 @@ import org.apache.ignite.testframework.junits.GridTestKernalContext;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
+import static org.apache.ignite.internal.MessageSerializationContext.LATEST_SCHEMA;
import static org.apache.ignite.internal.util.CommonUtils.makeMessageType;
/** Test for serialization round-trip of {@link QueryEntityMessage} and {@link QueryEntityExMessage}. */
@@ -173,7 +174,7 @@ public class QueryEntityMessageSerializationTest extends GridCommonAbstractTest
DirectMessageWriter writer = new DirectMessageWriter(msgFactory);
writer.setBuffer(buf);
- assertTrue(MessageSerialization.writeTo(msgFactory, msg, writer));
+ assertTrue(MessageSerialization.writeTo(msgFactory, msg, writer, LATEST_SCHEMA));
assertEquals("Writes" + ERROR_SUFFIX,
expReadsWritesCnt, writer.state());
@@ -184,7 +185,7 @@ public class QueryEntityMessageSerializationTest extends GridCommonAbstractTest
T res = (T)msgFactory.create(makeMessageType(buf.get(), buf.get()));
- assertTrue(MessageSerialization.readFrom(msgFactory, res, reader));
+ assertTrue(MessageSerialization.readFrom(msgFactory, res, reader, LATEST_SCHEMA));
assertEquals("Reads" + ERROR_SUFFIX,
expReadsWritesCnt, reader.state());
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java
index d661f5cff66..066d0403917 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java
@@ -19,6 +19,9 @@ package org.apache.ignite.internal.processors.rollingupgrade.feature;
/** */
public class TestPluginReleaseFeatures_2_0_0 {
+ /** */
+ public static final IgniteFeature VER_1_0_0_ID_0_FEATURE = TestPluginReleaseFeatures_1_0_0.VER_1_0_0_ID_0_FEATURE;
+
/** */
public static final IgniteFeature VER_2_0_0_ID_1_FEATURE = new TestPluginFeature(1);
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java
index edb4b3b3646..385f7ded6a9 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java
@@ -20,7 +20,10 @@ package org.apache.ignite.internal.processors.rollingupgrade.feature;
/** */
public class TestPluginReleaseFeatures_2_1_0 {
/** */
- public static final IgniteFeature VER_2_1_0_ID_1_FEATURE = new TestPluginFeature(1);
+ public static final IgniteFeature VER_1_0_0_ID_0_FEATURE = TestPluginReleaseFeatures_2_0_0.VER_1_0_0_ID_0_FEATURE;
+
+ /** */
+ public static final IgniteFeature VER_2_0_0_ID_1_FEATURE = TestPluginReleaseFeatures_2_0_0.VER_2_0_0_ID_1_FEATURE;
/** */
public static final IgniteFeature VER_2_1_0_ID_2_FEATURE = new TestPluginFeature(2);
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/RollingUpgradeMessageSerializationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/RollingUpgradeMessageSerializationTest.java
new file mode 100644
index 00000000000..8ba97aa82c1
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/RollingUpgradeMessageSerializationTest.java
@@ -0,0 +1,510 @@
+/*
+ * 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.processors.rollingupgrade.message;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.processors.rollingupgrade.AbstractRollingUpgradeTest;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.MessagesPluginProvider;
+import org.junit.Test;
+
+import static org.apache.ignite.internal.processors.rollingupgrade.message.TestMessage.A;
+import static org.apache.ignite.internal.processors.rollingupgrade.message.TestMessage.B;
+import static org.apache.ignite.internal.processors.rollingupgrade.message.TestMessage.C;
+import static org.apache.ignite.internal.processors.rollingupgrade.message.TestMessage.D;
+import static org.apache.ignite.internal.processors.rollingupgrade.message.TestMessage.E;
+import static org.apache.ignite.internal.processors.rollingupgrade.message.TestMessage.F;
+
+/** */
+public class RollingUpgradeMessageSerializationTest extends AbstractRollingUpgradeTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName, String ver) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName, ver);
+
+ cfg.setPluginProviders(org.apache.ignite.internal.util.typedef.F.concat(
+ cfg.getPluginProviders(),
+ new MessagesPluginProvider(
+ TestCoreMessage.class,
+ TestPluginMessage.class,
+ TestDefaultRegistryMessage.class))
+ );
+
+ return cfg;
+ }
+
+ /** */
+ @Test
+ public void testSameOldVersion() throws Exception {
+ checkMutualCoreMessageSend("2.19.0", "2.19.0", A, B, C, null, null, null);
+ }
+
+ /** */
+ @Test
+ public void testMixedPair() throws Exception {
+ checkMutualCoreMessageSend("2.19.0", "2.20.0", A, B, C, null, null, null);
+ }
+
+ /** */
+ @Test
+ public void testSameNewVersion() throws Exception {
+ checkMutualCoreMessageSend("2.20.0", "2.20.0", A, B, C, D, E, null);
+ }
+
+ /** */
+ @Test
+ public void testWindowOpenSameVersion() throws Exception {
+ checkMutualCoreMessageSend("2.19.2", "2.19.2", A, B, C, D, null, null);
+ }
+
+ /** */
+ @Test
+ public void testWindowOpenMixedPair() throws Exception {
+ checkMutualCoreMessageSend("2.19.2", "2.20.0", A, B, C, D, null, null);
+ }
+
+ /** */
+ @Test
+ public void testWindowClosed() throws Exception {
+ checkMutualCoreMessageSend("2.20.0", "2.20.1", A, null, C, null, E, null);
+ }
+
+ /** */
+ @Test
+ public void testDiscoveryNewerClient() throws Exception {
+ IgniteEx srv = startGrid(0, "2.19.0");
+
+ ru(srv).enableVersionUpgrade();
+
+ startClientGrid(1, "2.20.0");
+
+ checkCoreMessageBroadcast(srv, A, B, C, null, null, null);
+ }
+
+ /** */
+ @Test
+ public void testDiscoveryClientOriginated() throws Exception {
+ IgniteEx srv = startGrid(0, "2.19.0");
+
+ ru(srv).enableVersionUpgrade();
+
+ IgniteEx cli1 = startClientGrid(1, "2.20.0");
+
+ startClientGrid(2, "2.19.0");
+
+ checkCoreMessageBroadcast(cli1, A, B, C, null, null, null);
+ }
+
+ /** */
+ @Test
+ public void testDiscoveryClientsOnDifferentVersions() throws Exception {
+ startGrid(0, "2.19.0");
+ startGrid(1, "2.19.0");
+
+ ru(1).enableVersionUpgrade();
+
+ upgradeNodeVersion(0, "2.20.0");
+ upgradeNodeVersion(1, "2.20.0");
+
+ IgniteEx newVerCli = startClientGrid(2, "2.20.0");
+ IgniteEx oldVerCli = startClientGrid(3, "2.19.0");
+
+ Map<String, TestCoreMessage> receivedMsgs = sendOverDiscovery(grid(1), TestCoreMessage.build());
+
+ assertFields(A, B, C, D, E, null, receivedMsgs.get(newVerCli.name()));
+ assertFields(A, B, C, null, null, null, receivedMsgs.get(oldVerCli.name()));
+ }
+
+ /** */
+ @Test
+ public void testCommunicationWithClient() throws Exception {
+ IgniteEx srv = startGrid(0, "2.19.0");
+
+ ru(srv).enableVersionUpgrade();
+
+ IgniteEx client = startClientGrid(1, "2.20.0");
+
+ checkMutualCoreMessageSend(srv, client, A, B, C, null, null, null);
+ }
+
+ /** */
+ @Test
+ public void testDefaultRegistryMixedPair() throws Exception {
+ startServerNodes("2.19.0", "2.20.0");
+
+ checkMutualMessageSend(grid(0), grid(1), TestDefaultRegistryMessage::build, A, null, C, D, E, F);
+ }
+
+ /** */
+ @Test
+ public void testDiscoveryUniformRing() throws Exception {
+ startGrid(0, "2.20.0");
+ startGrid(1, "2.20.0");
+ startGrid(2, "2.20.0");
+
+ checkCoreMessageBroadcast(grid(1), A, B, C, D, E, null);
+ }
+
+ /** */
+ @Test
+ public void testDiscoveryMixedRing() throws Exception {
+ startGrid(0, "2.19.0");
+
+ ru(grid(0)).enableVersionUpgrade();
+
+ startGrid(1, "2.20.0");
+ startGrid(2, "2.20.0");
+
+ checkCoreMessageBroadcast(grid(1), A, B, C, null, null, null);
+ }
+
+ /** */
+ @Test
+ public void testCommunicationUpgradeOpensWindow() throws Exception {
+ startGrid(0, "2.19.0");
+ startGrid(1, "2.19.0");
+
+ checkMutualCoreMessageSend(grid(0), grid(1), A, B, C, null, null, null);
+
+ ru(1).enableVersionUpgrade();
+
+ upgradeNodeVersion(0, "2.19.2");
+ upgradeNodeVersion(1, "2.19.2");
+
+ checkMutualCoreMessageSend(grid(0), grid(1), A, B, C, D, null, null);
+ }
+
+ /** */
+ @Test
+ public void testCommunicationUpgradeAgreesNewFeature() throws Exception {
+ startGrid(0, "2.19.2");
+ startGrid(1, "2.19.2");
+
+ checkMutualCoreMessageSend(grid(0), grid(1), A, B, C, D, null, null);
+
+ ru(1).enableVersionUpgrade();
+
+ upgradeNodeVersion(0, "2.19.2", "2.20.0");
+
+ checkMutualCoreMessageSend(grid(0), grid(1), A, B, C, D, null, null);
+
+ upgradeNodeVersion(1, "2.19.2", "2.20.0");
+
+ checkMutualCoreMessageSend(grid(0), grid(1), A, B, C, D, E, null);
+ }
+
+ /** */
+ @Test
+ public void testPluginDiffersCoreMatches() throws Exception {
+ startServerNodes("2.20.0 | 1.0.0", "2.20.0 | 2.0.0");
+
+ checkMutualCoreMessageSend(grid(0), grid(1), A, B, C, D, E, null);
+
+ checkMutualMessageSend(grid(0), grid(1), TestPluginMessage::build, A, B, C, D, null, null);
+ }
+
+ /** */
+ @Test
+ public void testPluginSameVersion() throws Exception {
+ startServerNodes("2.20.0 | 2.0.0", "2.20.0 | 2.0.0");
+
+ checkMutualCoreMessageSend(grid(0), grid(1), A, B, C, D, E, null);
+
+ checkMutualMessageSend(grid(0), grid(1), TestPluginMessage::build, A, B, C, D, E, null);
+ }
+
+ /** */
+ @Test
+ public void testPluginMissingOnClient() throws Exception {
+ IgniteEx srv = startGrid(0, "2.20.0 | 2.0.0");
+
+ ru(srv).enableVersionUpgrade();
+
+ IgniteEx cli = startClientGrid(1, "2.20.0");
+
+ checkMutualMessageSend(srv, cli, TestPluginMessage::build, A, B, C, null, null, null);
+
+ checkMutualCoreMessageSend(srv, cli, A, B, C, D, E, null);
+ }
+
+ /** */
+ @Test
+ public void testWholeUpgradeProcess() throws Exception {
+ startGrid(0, "2.19.0");
+ startGrid(1, "2.19.0");
+ startClientGrid(2, "2.19.0");
+
+ checkMessagesTransmissionBetweenAllNodes(A, B, C, null, null, null);
+
+ ru(1).enableVersionUpgrade();
+
+ checkMessagesTransmissionBetweenAllNodes(A, B, C, null, null, null);
+
+ upgradeNodeVersion(0, "2.19.0", "2.19.2");
+
+ checkMessagesTransmissionBetweenAllNodes(A, B, C, null, null, null);
+
+ upgradeNodeVersion(1, "2.19.0", "2.19.2");
+
+ checkMutualCoreMessageSend(grid(0), grid(1), A, B, C, D, null, null);
+ checkMutualCoreMessageSend(grid(0), grid(2), A, B, C, null, null, null);
+ checkMutualCoreMessageSend(grid(1), grid(2), A, B, C, null, null, null);
+
+ upgradeNodeVersion(2, "2.19.0", "2.19.2");
+
+ checkMessagesTransmissionBetweenAllNodes(A, B, C, D, null, null);
+
+ finalizeClusterVersion(0, "2.19.2");
+
+ checkMessagesTransmissionBetweenAllNodes(A, B, C, D, null, null);
+
+ ru(1).enableVersionUpgrade();
+
+ upgradeNodeVersion(0, "2.19.2", "2.20.0");
+
+ checkMessagesTransmissionBetweenAllNodes(A, B, C, D, null, null);
+
+ upgradeNodeVersion(1, "2.19.2", "2.20.0");
+
+ checkMutualCoreMessageSend(grid(0), grid(1), A, B, C, D, E, null);
+ checkMutualCoreMessageSend(grid(0), grid(2), A, B, C, D, null, null);
+ checkMutualCoreMessageSend(grid(1), grid(2), A, B, C, D, null, null);
+
+ upgradeNodeVersion(2, "2.19.2", "2.20.0");
+
+ checkMessagesTransmissionBetweenAllNodes(A, B, C, D, E, null);
+
+ finalizeClusterVersion(0, "2.20.0");
+
+ checkMessagesTransmissionBetweenAllNodes(A, B, C, D, E, null);
+
+ ru(1).enableVersionUpgrade();
+
+ upgradeNodeVersion(0, "2.20.0", "2.20.1");
+
+ checkMutualCoreMessageSend(grid(0), grid(1), A, null, C, null, E, null);
+ checkMutualCoreMessageSend(grid(0), grid(2), A, null, C, null, E, null);
+ checkMutualCoreMessageSend(grid(1), grid(2), A, B, C, D, E, null);
+
+ upgradeNodeVersion(1, "2.20.0", "2.20.1");
+
+ checkMutualCoreMessageSend(grid(0), grid(1), A, B, C, D, E, F);
+ checkMutualCoreMessageSend(grid(0), grid(2), A, null, C, null, E, null);
+ checkMutualCoreMessageSend(grid(1), grid(2), A, null, C, null, E, null);
+
+ upgradeNodeVersion(2, "2.20.0", "2.20.1");
+
+ checkMessagesTransmissionBetweenAllNodes(A, B, C, D, E, F);
+
+ finalizeClusterVersion(0, "2.20.1");
+
+ checkMessagesTransmissionBetweenAllNodes(A, B, C, D, E, F);
+ }
+
+ /** */
+ private void checkMessagesTransmissionBetweenAllNodes(
+ String expA,
+ String expB,
+ String expC,
+ String expD,
+ String expE,
+ String expF
+ ) throws Exception {
+ List<Ignite> clusterNodes = Ignition.allGrids();
+
+ for (int i = 0; i < clusterNodes.size(); i++) {
+ for (int j = i + 1; j < clusterNodes.size(); j++) {
+ checkMutualCoreMessageSend(
+ (IgniteEx)clusterNodes.get(i), (IgniteEx)clusterNodes.get(j), expA, expB, expC, expD, expE, expF);
+ }
+ }
+ }
+
+ /** */
+ private void checkMutualCoreMessageSend(
+ String firstVer,
+ String secondVer,
+ String expA,
+ String expB,
+ String expC,
+ String expD,
+ String expE,
+ String expF
+ ) throws Exception {
+ startServerNodes(firstVer, secondVer);
+
+ checkMutualCoreMessageSend(grid(0), grid(1), expA, expB, expC, expD, expE, expF);
+ }
+
+ /** */
+ private void checkMutualCoreMessageSend(
+ IgniteEx first,
+ IgniteEx second,
+ String expA,
+ String expB,
+ String expC,
+ String expD,
+ String expE,
+ String expF
+ ) throws Exception {
+ checkMutualMessageSend(first, second, TestCoreMessage::build, expA, expB, expC, expD, expE, expF);
+ }
+
+ /** */
+ private void checkCoreMessageBroadcast(
+ IgniteEx from,
+ String expA,
+ String expB,
+ String expC,
+ String expD,
+ String expE,
+ String expF
+ ) throws Exception {
+ Collection<TestCoreMessage> receivedMsgs = sendOverDiscovery(from, TestCoreMessage.build()).values();
+
+ for (TestCoreMessage msg : receivedMsgs)
+ assertFields(expA, expB, expC, expD, expE, expF, msg);
+ }
+
+ /** */
+ private <T extends DiscoveryCustomMessage & TestMessage> void checkMutualMessageSend(
+ IgniteEx first,
+ IgniteEx second,
+ Supplier<T> msgFactory,
+ String expA,
+ String expB,
+ String expC,
+ String expD,
+ String expE,
+ String expF
+ ) throws Exception {
+ checkReceivedMessageFields(first, second, msgFactory, expA, expB, expC, expD, expE, expF);
+ checkReceivedMessageFields(second, first, msgFactory, expA, expB, expC, expD, expE, expF);
+ }
+
+ /** */
+ private <T extends DiscoveryCustomMessage & TestMessage> void checkReceivedMessageFields(
+ IgniteEx from,
+ IgniteEx to,
+ Supplier<T> msgFactory,
+ String expA,
+ String expB,
+ String expC,
+ String expD,
+ String expE,
+ String expF
+ ) throws Exception {
+ assertFields(expA, expB, expC, expD, expE, expF, send(from, to, msgFactory.get()));
+
+ assertFields(expA, expB, expC, expD, expE, expF, sendOverDiscovery(from, msgFactory.get()).get(to.name()));
+ }
+
+ /** */
+ private <T extends Message & TestMessage> T send(IgniteEx from, IgniteEx to, T msg) throws Exception {
+ AtomicReference<T> got = new AtomicReference<>();
+ CountDownLatch latch = new CountDownLatch(1);
+
+ String topic = msg.getClass().getName();
+
+ to.context().io().addMessageListener(topic, (nodeId, rcvd, plc) -> {
+ got.set((T)rcvd);
+
+ latch.countDown();
+ });
+
+ ClusterNode rcvNode = from.context().discovery().node(to.localNode().id());
+
+ from.context().io().sendToCustomTopic(rcvNode, topic, msg, GridIoPolicy.PUBLIC_POOL);
+
+ assertTrue(latch.await(getTestTimeout(), TimeUnit.MILLISECONDS));
+
+ return got.get();
+ }
+
+ /** */
+ private <T extends DiscoveryCustomMessage & TestMessage> Map<String, T> sendOverDiscovery(
+ IgniteEx from,
+ T msg
+ ) throws Exception {
+ List<Ignite> clusterNodes = Ignition.allGrids();
+
+ Map<String, T> receivedMsgs = new ConcurrentHashMap<>();
+
+ CountDownLatch latch = new CountDownLatch(clusterNodes.size());
+
+ for (Ignite rcv : clusterNodes) {
+ String name = rcv.name();
+
+ ((IgniteEx)rcv).context().discovery().setCustomEventListener((Class<T>)msg.getClass(),
+ (v, n, m) -> {
+ receivedMsgs.put(name, m);
+
+ latch.countDown();
+ });
+ }
+
+ from.context().discovery().sendCustomEvent(msg);
+
+ assertTrue(latch.await(getTestTimeout(), TimeUnit.MILLISECONDS));
+
+ receivedMsgs.remove(from.name());
+
+ return receivedMsgs;
+ }
+
+ /** */
+ private void startServerNodes(String firstVer, String secondVer) throws Exception {
+ IgniteEx first = startGrid(0, firstVer);
+
+ if (!firstVer.equals(secondVer))
+ ru(first).enableVersionUpgrade();
+
+ startGrid(1, secondVer);
+ }
+
+ /** */
+ private static void assertFields(
+ String expA,
+ String expB,
+ String expC,
+ String expD,
+ String expE,
+ String expF,
+ TestMessage msg
+ ) {
+ assertEquals(expA, msg.fldA());
+ assertEquals(expB, msg.fldB());
+ assertEquals(expC, msg.fldC());
+ assertEquals(expD, msg.fldD());
+ assertEquals(expE, msg.fldE());
+ assertEquals(expF, msg.fldF());
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestCoreMessage.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestCoreMessage.java
new file mode 100644
index 00000000000..35234fac1a5
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestCoreMessage.java
@@ -0,0 +1,107 @@
+/*
+ * 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.processors.rollingupgrade.message;
+
+import org.apache.ignite.internal.FeatureGated;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.TestIgniteReleaseFeatures_2_20_1;
+import org.apache.ignite.lang.IgniteUuid;
+import org.jetbrains.annotations.Nullable;
+
+/** */
+@FeatureGated(registry = TestIgniteReleaseFeatures_2_20_1.class)
+public class TestCoreMessage extends DiscoveryCustomMessage implements TestMessage {
+ /** */
+ @Order(0)
+ String fldA;
+
+ /** */
+ @Order(value = 1, deprecatedBy = "VER_2_20_0_ID_3_FEATURE")
+ String fldB;
+
+ /** */
+ @Order(2)
+ String fldC;
+
+ /** */
+ @Order(value = 3, introducedBy = "VER_2_19_2_ID_1_FEATURE", deprecatedBy = "VER_2_20_0_ID_3_FEATURE")
+ String fldD;
+
+ /** */
+ @Order(value = 4, introducedBy = "VER_2_20_0_ID_3_FEATURE")
+ String fldE;
+
+ /** */
+ @Order(value = 5, introducedBy = "VER_2_20_1_ID_6_FEATURE")
+ String fldF;
+
+ /** */
+ public TestCoreMessage() {
+ super(IgniteUuid.randomUuid());
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public DiscoveryCustomMessage ackMessage() {
+ return null;
+ }
+
+ /** */
+ public static TestCoreMessage build() {
+ TestCoreMessage msg = new TestCoreMessage();
+
+ msg.fldA = A;
+ msg.fldB = B;
+ msg.fldC = C;
+ msg.fldD = D;
+ msg.fldE = E;
+ msg.fldF = F;
+
+ return msg;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldA() {
+ return fldA;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldB() {
+ return fldB;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldC() {
+ return fldC;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldD() {
+ return fldD;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldE() {
+ return fldE;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldF() {
+ return fldF;
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestDefaultRegistryMessage.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestDefaultRegistryMessage.java
new file mode 100644
index 00000000000..02e6173990b
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestDefaultRegistryMessage.java
@@ -0,0 +1,104 @@
+/*
+ * 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.processors.rollingupgrade.message;
+
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.lang.IgniteUuid;
+import org.jetbrains.annotations.Nullable;
+
+/** */
+public class TestDefaultRegistryMessage extends DiscoveryCustomMessage implements TestMessage {
+ /** */
+ @Order(0)
+ String fldA;
+
+ /** */
+ @Order(value = 1, deprecatedBy = "ROLLING_UPGRADE_FEATURE")
+ String fldB;
+
+ /** */
+ @Order(2)
+ String fldC;
+
+ /** */
+ @Order(3)
+ String fldD;
+
+ /** */
+ @Order(4)
+ String fldE;
+
+ /** */
+ @Order(value = 5, introducedBy = "ROLLING_UPGRADE_FEATURE")
+ String fldF;
+
+ /** */
+ public TestDefaultRegistryMessage() {
+ super(IgniteUuid.randomUuid());
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public DiscoveryCustomMessage ackMessage() {
+ return null;
+ }
+
+ /** */
+ public static TestDefaultRegistryMessage build() {
+ TestDefaultRegistryMessage msg = new TestDefaultRegistryMessage();
+
+ msg.fldA = A;
+ msg.fldB = B;
+ msg.fldC = C;
+ msg.fldD = D;
+ msg.fldE = E;
+ msg.fldF = F;
+
+ return msg;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldA() {
+ return fldA;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldB() {
+ return fldB;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldC() {
+ return fldC;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldD() {
+ return fldD;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldE() {
+ return fldE;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldF() {
+ return fldF;
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestMessage.java
similarity index 57%
copy from modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java
copy to modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestMessage.java
index edb4b3b3646..15c639bf6fd 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestMessage.java
@@ -15,13 +15,43 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.rollingupgrade.feature;
+package org.apache.ignite.internal.processors.rollingupgrade.message;
/** */
-public class TestPluginReleaseFeatures_2_1_0 {
+public interface TestMessage {
/** */
- public static final IgniteFeature VER_2_1_0_ID_1_FEATURE = new TestPluginFeature(1);
+ public static final String A = "A";
/** */
- public static final IgniteFeature VER_2_1_0_ID_2_FEATURE = new TestPluginFeature(2);
+ public static final String B = "B";
+
+ /** */
+ public static final String C = "C";
+
+ /** */
+ public static final String D = "D";
+
+ /** */
+ public static final String E = "E";
+
+ /** */
+ public static final String F = "F";
+
+ /** */
+ public String fldA();
+
+ /** */
+ public String fldB();
+
+ /** */
+ public String fldC();
+
+ /** */
+ public String fldD();
+
+ /** */
+ public String fldE();
+
+ /** */
+ public String fldF();
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestPluginMessage.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestPluginMessage.java
new file mode 100644
index 00000000000..f901f7667bf
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestPluginMessage.java
@@ -0,0 +1,107 @@
+/*
+ * 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.processors.rollingupgrade.message;
+
+import org.apache.ignite.internal.FeatureGated;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.TestPluginReleaseFeatures_2_1_0;
+import org.apache.ignite.lang.IgniteUuid;
+import org.jetbrains.annotations.Nullable;
+
+/** */
+@FeatureGated(registry = TestPluginReleaseFeatures_2_1_0.class)
+public class TestPluginMessage extends DiscoveryCustomMessage implements TestMessage {
+ /** */
+ @Order(0)
+ String fldA;
+
+ /** */
+ @Order(value = 1, deprecatedBy = "VER_2_0_0_ID_1_FEATURE")
+ String fldB;
+
+ /** */
+ @Order(2)
+ String fldC;
+
+ /** */
+ @Order(value = 3, introducedBy = "VER_1_0_0_ID_0_FEATURE", deprecatedBy = "VER_2_0_0_ID_1_FEATURE")
+ String fldD;
+
+ /** */
+ @Order(value = 4, introducedBy = "VER_2_0_0_ID_1_FEATURE")
+ String fldE;
+
+ /** */
+ @Order(value = 5, introducedBy = "VER_2_1_0_ID_2_FEATURE")
+ String fldF;
+
+ /** */
+ public TestPluginMessage() {
+ super(IgniteUuid.randomUuid());
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public DiscoveryCustomMessage ackMessage() {
+ return null;
+ }
+
+ /** */
+ public static TestPluginMessage build() {
+ TestPluginMessage msg = new TestPluginMessage();
+
+ msg.fldA = A;
+ msg.fldB = B;
+ msg.fldC = C;
+ msg.fldD = D;
+ msg.fldE = E;
+ msg.fldF = F;
+
+ return msg;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldA() {
+ return fldA;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldB() {
+ return fldB;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldC() {
+ return fldC;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldD() {
+ return fldD;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldE() {
+ return fldE;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String fldF() {
+ return fldF;
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/LazyServiceConfigurationMessageSerializationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/LazyServiceConfigurationMessageSerializationTest.java
index 1536b53e992..909be2c0ba8 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/LazyServiceConfigurationMessageSerializationTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/LazyServiceConfigurationMessageSerializationTest.java
@@ -37,6 +37,7 @@ import org.apache.ignite.testframework.junits.GridTestKernalContext;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
+import static org.apache.ignite.internal.MessageSerializationContext.LATEST_SCHEMA;
import static org.apache.ignite.internal.util.CommonUtils.makeMessageType;
import static org.junit.Assert.assertArrayEquals;
@@ -126,7 +127,7 @@ public class LazyServiceConfigurationMessageSerializationTest extends GridCommon
DirectMessageWriter writer = new DirectMessageWriter(msgFactory);
writer.setBuffer(buf);
- assertTrue(MessageSerialization.writeTo(msgFactory, msg, writer));
+ assertTrue(MessageSerialization.writeTo(msgFactory, msg, writer, LATEST_SCHEMA));
assertEquals("Writes" + ERROR_SUFFIX, expReadsWritesCnt, writer.state());
buf.flip();
@@ -136,7 +137,7 @@ public class LazyServiceConfigurationMessageSerializationTest extends GridCommon
T res = (T)msgFactory.create(makeMessageType(buf.get(), buf.get()));
- assertTrue(MessageSerialization.readFrom(msgFactory, res, reader));
+ assertTrue(MessageSerialization.readFrom(msgFactory, res, reader, LATEST_SCHEMA));
assertEquals("Reads" + ERROR_SUFFIX, expReadsWritesCnt, reader.state());
DiscoveryMarshalling.unmarshal(res, kctx);
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessResultMarshallingTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessResultMarshallingTest.java
index 0286297db11..8a8f0521739 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessResultMarshallingTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessResultMarshallingTest.java
@@ -29,6 +29,7 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.CoreMessagesProvider;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.managers.communication.CommunicationMarshalling;
import org.apache.ignite.internal.managers.communication.GridIoMessage;
@@ -239,7 +240,7 @@ public class DistributedProcessResultMarshallingTest extends GridCommonAbstractT
/** */
private static class PayloadSerializer implements MessageSerializer<PayloadMessage> {
/** {@inheritDoc} */
- @Override public boolean writeTo(PayloadMessage msg, MessageWriter writer) {
+ @Override public boolean writeTo(PayloadMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -255,7 +256,7 @@ public class DistributedProcessResultMarshallingTest extends GridCommonAbstractT
writer.incrementState();
case 1:
- if (!writer.writeMessage(msg.err))
+ if (!writer.writeMessage(msg.err, ctx))
return false;
writer.incrementState();
@@ -265,7 +266,7 @@ public class DistributedProcessResultMarshallingTest extends GridCommonAbstractT
}
/** {@inheritDoc} */
- @Override public boolean readFrom(PayloadMessage msg, MessageReader reader) {
+ @Override public boolean readFrom(PayloadMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
msg.valBytes = reader.readByteArray();
@@ -276,7 +277,7 @@ public class DistributedProcessResultMarshallingTest extends GridCommonAbstractT
reader.incrementState();
case 1:
- msg.err = reader.readMessage();
+ msg.err = reader.readMessage(ctx);
if (!reader.isLastRead())
return false;
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/IgniteExceptionInNioWorkerSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/IgniteExceptionInNioWorkerSelfTest.java
index 1128dc0bb41..ee37f837c39 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/IgniteExceptionInNioWorkerSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/IgniteExceptionInNioWorkerSelfTest.java
@@ -24,6 +24,7 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.GridTopic;
import org.apache.ignite.internal.IgniteDiagnosticRequest;
import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.plugin.AbstractTestPluginProvider;
import org.apache.ignite.plugin.ExtensionRegistry;
import org.apache.ignite.plugin.PluginContext;
@@ -92,7 +93,7 @@ public class IgniteExceptionInNioWorkerSelfTest extends GridCommonAbstractTest {
private boolean fail = true;
/** {@inheritDoc} */
- @Override public boolean writeTo(BrokenMessage msg, MessageWriter writer) {
+ @Override public boolean writeTo(BrokenMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -110,7 +111,7 @@ public class IgniteExceptionInNioWorkerSelfTest extends GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override public boolean readFrom(BrokenMessage msg, MessageReader reader) {
+ @Override public boolean readFrom(BrokenMessage msg, MessageReader reader, MessageSerializationContext ctx) {
return true;
}
diff --git a/modules/core/src/test/java/org/apache/ignite/plugin/extensions/communication/MessageMarshalOnceTest.java b/modules/core/src/test/java/org/apache/ignite/plugin/extensions/communication/MessageMarshalOnceTest.java
index 12f4e34afed..390446051b6 100644
--- a/modules/core/src/test/java/org/apache/ignite/plugin/extensions/communication/MessageMarshalOnceTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/plugin/extensions/communication/MessageMarshalOnceTest.java
@@ -27,6 +27,7 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.CoreMessagesProvider;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.GridTopic;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.managers.communication.GridIoMessage;
import org.apache.ignite.internal.managers.communication.GridIoPolicy;
@@ -157,7 +158,7 @@ public class MessageMarshalOnceTest extends GridCommonAbstractTest {
/** Header-only serializer for the fieldless {@link MarshalOnceCheckMessage}. */
private static class Serializer implements MessageSerializer<MarshalOnceCheckMessage> {
/** {@inheritDoc} */
- @Override public boolean writeTo(MarshalOnceCheckMessage msg, MessageWriter writer) {
+ @Override public boolean writeTo(MarshalOnceCheckMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -169,7 +170,7 @@ public class MessageMarshalOnceTest extends GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override public boolean readFrom(MarshalOnceCheckMessage msg, MessageReader reader) {
+ @Override public boolean readFrom(MarshalOnceCheckMessage msg, MessageReader reader, MessageSerializationContext ctx) {
return true;
}
@@ -209,7 +210,7 @@ public class MessageMarshalOnceTest extends GridCommonAbstractTest {
/** Header-only serializer for the fieldless {@link RetryCheckMessage}. */
private static class RetrySerializer implements MessageSerializer<RetryCheckMessage> {
/** {@inheritDoc} */
- @Override public boolean writeTo(RetryCheckMessage msg, MessageWriter writer) {
+ @Override public boolean writeTo(RetryCheckMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -221,7 +222,7 @@ public class MessageMarshalOnceTest extends GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override public boolean readFrom(RetryCheckMessage msg, MessageReader reader) {
+ @Override public boolean readFrom(RetryCheckMessage msg, MessageReader reader, MessageSerializationContext ctx) {
return true;
}
diff --git a/modules/core/src/test/java/org/apache/ignite/plugin/security/SecurityBasicPermissionSetSerializationTest.java b/modules/core/src/test/java/org/apache/ignite/plugin/security/SecurityBasicPermissionSetSerializationTest.java
index 2d13187a975..d91269bfdb4 100644
--- a/modules/core/src/test/java/org/apache/ignite/plugin/security/SecurityBasicPermissionSetSerializationTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/plugin/security/SecurityBasicPermissionSetSerializationTest.java
@@ -35,6 +35,7 @@ import org.apache.ignite.testframework.junits.GridTestKernalContext;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
+import static org.apache.ignite.internal.MessageSerializationContext.LATEST_SCHEMA;
import static org.apache.ignite.internal.util.CommonUtils.makeMessageType;
import static org.apache.ignite.plugin.security.SecurityPermission.ADMIN_CACHE;
import static org.apache.ignite.plugin.security.SecurityPermission.ADMIN_QUERY;
@@ -117,7 +118,7 @@ public class SecurityBasicPermissionSetSerializationTest extends GridCommonAbstr
DirectMessageWriter writer = new DirectMessageWriter(msgFactory);
writer.setBuffer(buf);
- assertTrue(MessageSerialization.writeTo(msgFactory, msg, writer));
+ assertTrue(MessageSerialization.writeTo(msgFactory, msg, writer, LATEST_SCHEMA));
buf.flip();
@@ -126,7 +127,7 @@ public class SecurityBasicPermissionSetSerializationTest extends GridCommonAbstr
T res = (T)msgFactory.create(makeMessageType(buf.get(), buf.get()));
- assertTrue(MessageSerialization.readFrom(msgFactory, res, reader));
+ assertTrue(MessageSerialization.readFrom(msgFactory, res, reader, LATEST_SCHEMA));
return res;
}
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TestDelayMessageSerializer.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TestDelayMessageSerializer.java
index ce905b9806f..1d61f44f57f 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TestDelayMessageSerializer.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TestDelayMessageSerializer.java
@@ -18,6 +18,7 @@
package org.apache.ignite.spi.communication.tcp;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
@@ -26,7 +27,7 @@ import org.apache.ignite.plugin.extensions.communication.MessageWriter;
/** Serializer for {@link TestDelayMessage} that injects an optional write delay for testing. */
public class TestDelayMessageSerializer implements MessageSerializer<TestDelayMessage> {
/** {@inheritDoc} */
- @Override public boolean writeTo(TestDelayMessage msg, MessageWriter writer) {
+ @Override public boolean writeTo(TestDelayMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -47,7 +48,7 @@ public class TestDelayMessageSerializer implements MessageSerializer<TestDelayMe
}
/** {@inheritDoc} */
- @Override public boolean readFrom(TestDelayMessage msg, MessageReader reader) {
+ @Override public boolean readFrom(TestDelayMessage msg, MessageReader reader, MessageSerializationContext ctx) {
return true;
}
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/DiscoveryUnmarshalVulnerabilityTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/DiscoveryUnmarshalVulnerabilityTest.java
index 8e8029c295e..7f652d10132 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/DiscoveryUnmarshalVulnerabilityTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/DiscoveryUnmarshalVulnerabilityTest.java
@@ -35,6 +35,7 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.CoreMessagesProvider;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.direct.DirectMessageWriter;
import org.apache.ignite.internal.managers.communication.DiscoveryMarshalling;
import org.apache.ignite.internal.managers.communication.IgniteMessageFactory;
@@ -60,6 +61,7 @@ import org.junit.Test;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_MARSHALLER_BLACKLIST;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_MARSHALLER_WHITELIST;
+import static org.apache.ignite.internal.MessageSerializationContext.LATEST_SCHEMA;
import static org.apache.ignite.testframework.GridTestUtils.loadMarshaller;
import static org.apache.ignite.testframework.GridTestUtils.loadSerializer;
@@ -254,7 +256,7 @@ public abstract class DiscoveryUnmarshalVulnerabilityTest extends GridCommonAbst
DiscoveryMarshalling.marshal(msg, grid(0).context(), null);
- writer.writeMessage(msg);
+ writer.writeMessage(msg, LATEST_SCHEMA);
return buf.flip().compact().array();
}
@@ -278,17 +280,17 @@ public abstract class DiscoveryUnmarshalVulnerabilityTest extends GridCommonAbst
private MessageSerializer<ExploitMessage> serde;
/** {@inheritDoc} */
- @Override public boolean writeTo(ExploitMessage msg, MessageWriter writer) {
+ @Override public boolean writeTo(ExploitMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
initIfNecessary();
- return serde.writeTo(msg, writer);
+ return serde.writeTo(msg, writer, ctx);
}
/** {@inheritDoc} */
- @Override public boolean readFrom(ExploitMessage msg, MessageReader reader) {
+ @Override public boolean readFrom(ExploitMessage msg, MessageReader reader, MessageSerializationContext ctx) {
initIfNecessary();
- return serde.readFrom(msg, reader);
+ return serde.readFrom(msg, reader, ctx);
}
/** {@inheritDoc} */
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/GridTestNode.java b/modules/core/src/test/java/org/apache/ignite/testframework/GridTestNode.java
index cd76954a1a2..e90cc868475 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/GridTestNode.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/GridTestNode.java
@@ -17,15 +17,19 @@
package org.apache.ignite.testframework;
+import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cluster.ClusterMetrics;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteNodeAttributes;
+import org.apache.ignite.internal.managers.discovery.IgniteClusterNode;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteNodeFeatureSet;
import org.apache.ignite.internal.util.lang.GridMetadataAwareAdapter;
import org.apache.ignite.lang.IgniteProductVersion;
@@ -34,7 +38,7 @@ import static org.apache.ignite.lang.IgniteProductVersion.fromString;
/**
* Test node.
*/
-public class GridTestNode extends GridMetadataAwareAdapter implements ClusterNode {
+public class GridTestNode extends GridMetadataAwareAdapter implements IgniteClusterNode {
/** */
private static final IgniteProductVersion VERSION = fromString("99.99.99");
@@ -103,13 +107,6 @@ public class GridTestNode extends GridMetadataAwareAdapter implements ClusterNod
return id;
}
- /**
- * @param consistentId Consistent ID.
- */
- public void consistentId(Object consistentId) {
- this.consistentId = consistentId;
- }
-
/** {@inheritDoc} */
@Override public Object consistentId() {
return consistentId;
@@ -213,6 +210,16 @@ public class GridTestNode extends GridMetadataAwareAdapter implements ClusterNod
return VERSION;
}
+ /** {@inheritDoc} */
+ @Override public IgniteNodeFeatureSet features() {
+ return IgniteNodeFeatureSet.LOCAL_CORE_FEATURES;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void setConsistentId(Serializable consistentId) {
+ this.consistentId = consistentId;
+ }
+
/**
* Sets node metrics.
*
@@ -222,6 +229,16 @@ public class GridTestNode extends GridMetadataAwareAdapter implements ClusterNod
this.metrics = metrics;
}
+ /** {@inheritDoc} */
+ @Override public Map<Integer, CacheMetrics> cacheMetrics() {
+ return Collections.emptyMap();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void setCacheMetrics(Map<Integer, CacheMetrics> cacheMetrics) {
+ // No-op.
+ }
+
/** {@inheritDoc} */
@Override public boolean isLocal() {
return false;
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
index 3b2a188e8b1..2ad409701d0 100755
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
@@ -1398,7 +1398,7 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
if (nodeInitializer != null)
nodeInitializer.apply(fakeNode);
- fakeNode.consistentId(joiningNodeConsistentId == null ? getTestIgniteInstanceName(nodes.size()) :
+ fakeNode.setConsistentId(joiningNodeConsistentId == null ? getTestIgniteInstanceName(nodes.size()) :
joiningNodeConsistentId);
nodes.add(fakeNode);
@@ -1449,7 +1449,7 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
GridTestNode fakeNode = new GridTestNode(UUID.randomUUID(), null);
- fakeNode.consistentId(getTestIgniteInstanceName(nodes.size()));
+ fakeNode.setConsistentId(getTestIgniteInstanceName(nodes.size()));
nodes.add(fakeNode);
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
index c8ff74b6ced..e8a54f1c0d4 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
@@ -67,6 +67,7 @@ import org.apache.ignite.internal.processors.rollingupgrade.CoreVersionRollingUp
import org.apache.ignite.internal.processors.rollingupgrade.PluginVersionRollingUpgradeTest;
import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeatureSetTest;
import org.apache.ignite.internal.processors.rollingupgrade.feature.ManagementApiVersionValidationTest;
+import org.apache.ignite.internal.processors.rollingupgrade.message.RollingUpgradeMessageSerializationTest;
import org.apache.ignite.internal.product.GridProductVersionSelfTest;
import org.apache.ignite.internal.util.ErrorMessageSelfTest;
import org.apache.ignite.internal.util.nio.IgniteExceptionInNioWorkerSelfTest;
@@ -112,6 +113,7 @@ import org.junit.runners.Suite;
CoreVersionRollingUpgradeTest.class,
PluginVersionRollingUpgradeTest.class,
+ RollingUpgradeMessageSerializationTest.class,
ManagementApiVersionValidationTest.class,
GridProductVersionSelfTest.class,
GridAffinityAssignmentV2Test.class,
diff --git a/modules/core/src/test/resources/codegen/ChildMessageSerializer.java b/modules/core/src/test/resources/codegen/ChildMessageSerializer.java
index 8f77e193422..1dd438554f9 100644
--- a/modules/core/src/test/resources/codegen/ChildMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/ChildMessageSerializer.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal;
import org.apache.ignite.internal.AbstractMessage;
import org.apache.ignite.internal.ChildMessage;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
@@ -30,7 +31,7 @@ import org.apache.ignite.plugin.extensions.communication.MessageWriter;
*/
public final class ChildMessageSerializer implements MessageSerializer<ChildMessage> {
/** */
- @Override public final boolean writeTo(ChildMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(ChildMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -68,7 +69,7 @@ public final class ChildMessageSerializer implements MessageSerializer<ChildMess
}
/** */
- @Override public final boolean readFrom(ChildMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(ChildMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
((AbstractMessage)msg).id = reader.readInt();
diff --git a/modules/core/src/test/resources/codegen/CorrectEmptyMessageSerializer.java b/modules/core/src/test/resources/codegen/CorrectEmptyMessageSerializer.java
index 9dfc2ef935a..28b912c8e29 100644
--- a/modules/core/src/test/resources/codegen/CorrectEmptyMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/CorrectEmptyMessageSerializer.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal;
import org.apache.ignite.internal.CorrectEmptyMessage;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
@@ -29,7 +30,7 @@ import org.apache.ignite.plugin.extensions.communication.MessageWriter;
*/
public final class CorrectEmptyMessageSerializer implements MessageSerializer<CorrectEmptyMessage> {
/** */
- @Override public final boolean writeTo(CorrectEmptyMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(CorrectEmptyMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -44,7 +45,7 @@ public final class CorrectEmptyMessageSerializer implements MessageSerializer<Co
}
/** */
- @Override public final boolean readFrom(CorrectEmptyMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(CorrectEmptyMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
}
diff --git a/modules/core/src/test/resources/codegen/CustomMapperEnumFieldsMessageSerializer.java b/modules/core/src/test/resources/codegen/CustomMapperEnumFieldsMessageSerializer.java
index fe0cb67561b..c7e64dec44b 100644
--- a/modules/core/src/test/resources/codegen/CustomMapperEnumFieldsMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/CustomMapperEnumFieldsMessageSerializer.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal;
import org.apache.ignite.internal.CustomMapperEnumFieldsMessage;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TransactionIsolationEnumMapper;
import org.apache.ignite.plugin.extensions.communication.CollectionImplementationType;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
@@ -41,7 +42,7 @@ public final class CustomMapperEnumFieldsMessageSerializer implements MessageSer
private static final MessageCollectionType isolationsCollDesc = new MessageCollectionType(new MessageCollectionType(new MessageEnumType<>(TransactionIsolation.class, transactionIsolationMapper::encode, transactionIsolationMapper::decode), CollectionImplementationType.ARRAY_LIST), CollectionImplementationType.ARRAY_LIST);
/** */
- @Override public final boolean writeTo(CustomMapperEnumFieldsMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(CustomMapperEnumFieldsMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -57,7 +58,7 @@ public final class CustomMapperEnumFieldsMessageSerializer implements MessageSer
writer.incrementState();
case 1:
- if (!writer.writeCollection(msg.isolations, isolationsCollDesc))
+ if (!writer.writeCollection(msg.isolations, isolationsCollDesc, ctx))
return false;
writer.incrementState();
@@ -67,7 +68,7 @@ public final class CustomMapperEnumFieldsMessageSerializer implements MessageSer
}
/** */
- @Override public final boolean readFrom(CustomMapperEnumFieldsMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(CustomMapperEnumFieldsMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
msg.txMode = transactionIsolationMapper.decode(reader.readByte());
@@ -78,7 +79,7 @@ public final class CustomMapperEnumFieldsMessageSerializer implements MessageSer
reader.incrementState();
case 1:
- msg.isolations = reader.readCollection(isolationsCollDesc);
+ msg.isolations = reader.readCollection(isolationsCollDesc, ctx);
if (!reader.isLastRead())
return false;
diff --git a/modules/core/src/test/resources/codegen/DefaultMapperEnumFieldsMessageSerializer.java b/modules/core/src/test/resources/codegen/DefaultMapperEnumFieldsMessageSerializer.java
index dba15be2846..9e707da48c7 100644
--- a/modules/core/src/test/resources/codegen/DefaultMapperEnumFieldsMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/DefaultMapperEnumFieldsMessageSerializer.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal;
import org.apache.ignite.internal.DefaultMapperEnumFieldsMessage;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord.PartitionState;
import org.apache.ignite.plugin.extensions.communication.CollectionImplementationType;
@@ -50,7 +51,7 @@ public final class DefaultMapperEnumFieldsMessageSerializer implements MessageSe
private static final MessageMapType isolationStringMapCollDesc = new MessageMapType(new MessageCollectionType(new MessageEnumType<>(TransactionIsolation.class, DefaultEnumMapper.INSTANCE::encode, b -> DefaultEnumMapper.INSTANCE.decode(transactionIsolationVals, b)), CollectionImplementationType.ARRAY_LIST), new MessageItemType(MessageCollectionItemType.STRING), false);
/** */
- @Override public final boolean writeTo(DefaultMapperEnumFieldsMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(DefaultMapperEnumFieldsMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -72,13 +73,13 @@ public final class DefaultMapperEnumFieldsMessageSerializer implements MessageSe
writer.incrementState();
case 2:
- if (!writer.writeMap(msg.isolationStringMap, isolationStringMapCollDesc))
+ if (!writer.writeMap(msg.isolationStringMap, isolationStringMapCollDesc, ctx))
return false;
writer.incrementState();
case 3:
- if (!writer.writeCollection(msg.partStates, partStatesCollDesc))
+ if (!writer.writeCollection(msg.partStates, partStatesCollDesc, ctx))
return false;
writer.incrementState();
@@ -88,7 +89,7 @@ public final class DefaultMapperEnumFieldsMessageSerializer implements MessageSe
}
/** */
- @Override public final boolean readFrom(DefaultMapperEnumFieldsMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(DefaultMapperEnumFieldsMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
msg.publicEnum = DefaultEnumMapper.INSTANCE.decode(transactionIsolationVals, reader.readByte());
@@ -107,7 +108,7 @@ public final class DefaultMapperEnumFieldsMessageSerializer implements MessageSe
reader.incrementState();
case 2:
- msg.isolationStringMap = reader.readMap(isolationStringMapCollDesc);
+ msg.isolationStringMap = reader.readMap(isolationStringMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -115,7 +116,7 @@ public final class DefaultMapperEnumFieldsMessageSerializer implements MessageSe
reader.incrementState();
case 3:
- msg.partStates = reader.readCollection(partStatesCollDesc);
+ msg.partStates = reader.readCollection(partStatesCollDesc, ctx);
if (!reader.isLastRead())
return false;
diff --git a/modules/core/src/test/resources/codegen/TestCollectionsMessageSerializer.java b/modules/core/src/test/resources/codegen/TestCollectionsMessageSerializer.java
index bd79c20a48c..d1814946151 100644
--- a/modules/core/src/test/resources/codegen/TestCollectionsMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/TestCollectionsMessageSerializer.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TestCollectionsMessage;
import org.apache.ignite.plugin.extensions.communication.CollectionImplementationType;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
@@ -86,7 +87,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
private static final MessageCollectionType uuidListCollDesc = new MessageCollectionType(new MessageItemType(MessageCollectionItemType.UUID), CollectionImplementationType.ARRAY_LIST);
/** */
- @Override public final boolean writeTo(TestCollectionsMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(TestCollectionsMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -96,157 +97,157 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
switch (writer.state()) {
case 0:
- if (!writer.writeCollection(msg.booleanArrayList, booleanArrayListCollDesc))
+ if (!writer.writeCollection(msg.booleanArrayList, booleanArrayListCollDesc, ctx))
return false;
writer.incrementState();
case 1:
- if (!writer.writeCollection(msg.byteArrayList, byteArrayListCollDesc))
+ if (!writer.writeCollection(msg.byteArrayList, byteArrayListCollDesc, ctx))
return false;
writer.incrementState();
case 2:
- if (!writer.writeCollection(msg.shortArrayList, shortArrayListCollDesc))
+ if (!writer.writeCollection(msg.shortArrayList, shortArrayListCollDesc, ctx))
return false;
writer.incrementState();
case 3:
- if (!writer.writeCollection(msg.intArrayList, intArrayListCollDesc))
+ if (!writer.writeCollection(msg.intArrayList, intArrayListCollDesc, ctx))
return false;
writer.incrementState();
case 4:
- if (!writer.writeCollection(msg.longArrayList, longArrayListCollDesc))
+ if (!writer.writeCollection(msg.longArrayList, longArrayListCollDesc, ctx))
return false;
writer.incrementState();
case 5:
- if (!writer.writeCollection(msg.charArrayList, charArrayListCollDesc))
+ if (!writer.writeCollection(msg.charArrayList, charArrayListCollDesc, ctx))
return false;
writer.incrementState();
case 6:
- if (!writer.writeCollection(msg.floatArrayList, floatArrayListCollDesc))
+ if (!writer.writeCollection(msg.floatArrayList, floatArrayListCollDesc, ctx))
return false;
writer.incrementState();
case 7:
- if (!writer.writeCollection(msg.doubleArrayList, doubleArrayListCollDesc))
+ if (!writer.writeCollection(msg.doubleArrayList, doubleArrayListCollDesc, ctx))
return false;
writer.incrementState();
case 8:
- if (!writer.writeCollection(msg.stringList, stringListCollDesc))
+ if (!writer.writeCollection(msg.stringList, stringListCollDesc, ctx))
return false;
writer.incrementState();
case 9:
- if (!writer.writeCollection(msg.uuidList, uuidListCollDesc))
+ if (!writer.writeCollection(msg.uuidList, uuidListCollDesc, ctx))
return false;
writer.incrementState();
case 10:
- if (!writer.writeCollection(msg.bitSetList, bitSetListCollDesc))
+ if (!writer.writeCollection(msg.bitSetList, bitSetListCollDesc, ctx))
return false;
writer.incrementState();
case 11:
- if (!writer.writeCollection(msg.igniteUuidList, igniteUuidListCollDesc))
+ if (!writer.writeCollection(msg.igniteUuidList, igniteUuidListCollDesc, ctx))
return false;
writer.incrementState();
case 12:
- if (!writer.writeCollection(msg.affTopVersionList, affTopVersionListCollDesc))
+ if (!writer.writeCollection(msg.affTopVersionList, affTopVersionListCollDesc, ctx))
return false;
writer.incrementState();
case 13:
- if (!writer.writeCollection(msg.boxedBooleanList, boxedBooleanListCollDesc))
+ if (!writer.writeCollection(msg.boxedBooleanList, boxedBooleanListCollDesc, ctx))
return false;
writer.incrementState();
case 14:
- if (!writer.writeCollection(msg.boxedByteList, boxedByteListCollDesc))
+ if (!writer.writeCollection(msg.boxedByteList, boxedByteListCollDesc, ctx))
return false;
writer.incrementState();
case 15:
- if (!writer.writeCollection(msg.boxedShortList, boxedShortListCollDesc))
+ if (!writer.writeCollection(msg.boxedShortList, boxedShortListCollDesc, ctx))
return false;
writer.incrementState();
case 16:
- if (!writer.writeCollection(msg.boxedIntList, boxedIntListCollDesc))
+ if (!writer.writeCollection(msg.boxedIntList, boxedIntListCollDesc, ctx))
return false;
writer.incrementState();
case 17:
- if (!writer.writeCollection(msg.boxedLongList, boxedLongListCollDesc))
+ if (!writer.writeCollection(msg.boxedLongList, boxedLongListCollDesc, ctx))
return false;
writer.incrementState();
case 18:
- if (!writer.writeCollection(msg.boxedCharList, boxedCharListCollDesc))
+ if (!writer.writeCollection(msg.boxedCharList, boxedCharListCollDesc, ctx))
return false;
writer.incrementState();
case 19:
- if (!writer.writeCollection(msg.boxedFloatList, boxedFloatListCollDesc))
+ if (!writer.writeCollection(msg.boxedFloatList, boxedFloatListCollDesc, ctx))
return false;
writer.incrementState();
case 20:
- if (!writer.writeCollection(msg.boxedDoubleList, boxedDoubleListCollDesc))
+ if (!writer.writeCollection(msg.boxedDoubleList, boxedDoubleListCollDesc, ctx))
return false;
writer.incrementState();
case 21:
- if (!writer.writeCollection(msg.messageList, messageListCollDesc))
+ if (!writer.writeCollection(msg.messageList, messageListCollDesc, ctx))
return false;
writer.incrementState();
case 22:
- if (!writer.writeCollection(msg.gridLongListList, gridLongListListCollDesc))
+ if (!writer.writeCollection(msg.gridLongListList, gridLongListListCollDesc, ctx))
return false;
writer.incrementState();
case 23:
- if (!writer.writeCollection(msg.boxedIntegerSet, boxedIntegerSetCollDesc))
+ if (!writer.writeCollection(msg.boxedIntegerSet, boxedIntegerSetCollDesc, ctx))
return false;
writer.incrementState();
case 24:
- if (!writer.writeCollection(msg.bitSetSet, bitSetSetCollDesc))
+ if (!writer.writeCollection(msg.bitSetSet, bitSetSetCollDesc, ctx))
return false;
writer.incrementState();
case 25:
- if (!writer.writeCollection(msg.cacheObjectSet, cacheObjectSetCollDesc))
+ if (!writer.writeCollection(msg.cacheObjectSet, cacheObjectSetCollDesc, ctx))
return false;
writer.incrementState();
@@ -256,10 +257,10 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
}
/** */
- @Override public final boolean readFrom(TestCollectionsMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(TestCollectionsMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
- msg.booleanArrayList = reader.readCollection(booleanArrayListCollDesc);
+ msg.booleanArrayList = reader.readCollection(booleanArrayListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -267,7 +268,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 1:
- msg.byteArrayList = reader.readCollection(byteArrayListCollDesc);
+ msg.byteArrayList = reader.readCollection(byteArrayListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -275,7 +276,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 2:
- msg.shortArrayList = reader.readCollection(shortArrayListCollDesc);
+ msg.shortArrayList = reader.readCollection(shortArrayListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -283,7 +284,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 3:
- msg.intArrayList = reader.readCollection(intArrayListCollDesc);
+ msg.intArrayList = reader.readCollection(intArrayListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -291,7 +292,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 4:
- msg.longArrayList = reader.readCollection(longArrayListCollDesc);
+ msg.longArrayList = reader.readCollection(longArrayListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -299,7 +300,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 5:
- msg.charArrayList = reader.readCollection(charArrayListCollDesc);
+ msg.charArrayList = reader.readCollection(charArrayListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -307,7 +308,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 6:
- msg.floatArrayList = reader.readCollection(floatArrayListCollDesc);
+ msg.floatArrayList = reader.readCollection(floatArrayListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -315,7 +316,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 7:
- msg.doubleArrayList = reader.readCollection(doubleArrayListCollDesc);
+ msg.doubleArrayList = reader.readCollection(doubleArrayListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -323,7 +324,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 8:
- msg.stringList = reader.readCollection(stringListCollDesc);
+ msg.stringList = reader.readCollection(stringListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -331,7 +332,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 9:
- msg.uuidList = reader.readCollection(uuidListCollDesc);
+ msg.uuidList = reader.readCollection(uuidListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -339,7 +340,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 10:
- msg.bitSetList = reader.readCollection(bitSetListCollDesc);
+ msg.bitSetList = reader.readCollection(bitSetListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -347,7 +348,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 11:
- msg.igniteUuidList = reader.readCollection(igniteUuidListCollDesc);
+ msg.igniteUuidList = reader.readCollection(igniteUuidListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -355,7 +356,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 12:
- msg.affTopVersionList = reader.readCollection(affTopVersionListCollDesc);
+ msg.affTopVersionList = reader.readCollection(affTopVersionListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -363,7 +364,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 13:
- msg.boxedBooleanList = reader.readCollection(boxedBooleanListCollDesc);
+ msg.boxedBooleanList = reader.readCollection(boxedBooleanListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -371,7 +372,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 14:
- msg.boxedByteList = reader.readCollection(boxedByteListCollDesc);
+ msg.boxedByteList = reader.readCollection(boxedByteListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -379,7 +380,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 15:
- msg.boxedShortList = reader.readCollection(boxedShortListCollDesc);
+ msg.boxedShortList = reader.readCollection(boxedShortListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -387,7 +388,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 16:
- msg.boxedIntList = reader.readCollection(boxedIntListCollDesc);
+ msg.boxedIntList = reader.readCollection(boxedIntListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -395,7 +396,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 17:
- msg.boxedLongList = reader.readCollection(boxedLongListCollDesc);
+ msg.boxedLongList = reader.readCollection(boxedLongListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -403,7 +404,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 18:
- msg.boxedCharList = reader.readCollection(boxedCharListCollDesc);
+ msg.boxedCharList = reader.readCollection(boxedCharListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -411,7 +412,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 19:
- msg.boxedFloatList = reader.readCollection(boxedFloatListCollDesc);
+ msg.boxedFloatList = reader.readCollection(boxedFloatListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -419,7 +420,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 20:
- msg.boxedDoubleList = reader.readCollection(boxedDoubleListCollDesc);
+ msg.boxedDoubleList = reader.readCollection(boxedDoubleListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -427,7 +428,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 21:
- msg.messageList = reader.readCollection(messageListCollDesc);
+ msg.messageList = reader.readCollection(messageListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -435,7 +436,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 22:
- msg.gridLongListList = reader.readCollection(gridLongListListCollDesc);
+ msg.gridLongListList = reader.readCollection(gridLongListListCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -443,7 +444,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 23:
- msg.boxedIntegerSet = reader.readCollection(boxedIntegerSetCollDesc);
+ msg.boxedIntegerSet = reader.readCollection(boxedIntegerSetCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -451,7 +452,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 24:
- msg.bitSetSet = reader.readCollection(bitSetSetCollDesc);
+ msg.bitSetSet = reader.readCollection(bitSetSetCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -459,7 +460,7 @@ public final class TestCollectionsMessageSerializer implements MessageSerializer
reader.incrementState();
case 25:
- msg.cacheObjectSet = reader.readCollection(cacheObjectSetCollDesc);
+ msg.cacheObjectSet = reader.readCollection(cacheObjectSetCollDesc, ctx);
if (!reader.isLastRead())
return false;
diff --git a/modules/core/src/test/resources/codegen/TestEnumSetMessageSerializer.java b/modules/core/src/test/resources/codegen/TestEnumSetMessageSerializer.java
index fb0afa111fa..d5cc64a6abd 100644
--- a/modules/core/src/test/resources/codegen/TestEnumSetMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/TestEnumSetMessageSerializer.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TestEnumSetMessage;
import org.apache.ignite.plugin.extensions.communication.CollectionImplementationType;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
@@ -46,7 +47,7 @@ public final class TestEnumSetMessageSerializer implements MessageSerializer<Tes
private static final MessageMapType isolationsMapCollDesc = new MessageMapType(new MessageItemType(MessageCollectionItemType.STRING), new MessageCollectionType(new MessageEnumType<>(TransactionIsolation.class, DefaultEnumMapper.INSTANCE::encode, b -> DefaultEnumMapper.INSTANCE.decode(transactionIsolationVals, b)), CollectionImplementationType.ENUM_SET), false);
/** */
- @Override public final boolean writeTo(TestEnumSetMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(TestEnumSetMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -56,19 +57,19 @@ public final class TestEnumSetMessageSerializer implements MessageSerializer<Tes
switch (writer.state()) {
case 0:
- if (!writer.writeCollection(msg.isolations, isolationsCollDesc))
+ if (!writer.writeCollection(msg.isolations, isolationsCollDesc, ctx))
return false;
writer.incrementState();
case 1:
- if (!writer.writeMap(msg.isolationsMap, isolationsMapCollDesc))
+ if (!writer.writeMap(msg.isolationsMap, isolationsMapCollDesc, ctx))
return false;
writer.incrementState();
case 2:
- if (!writer.writeCollection(msg.isolationsList, isolationsListCollDesc))
+ if (!writer.writeCollection(msg.isolationsList, isolationsListCollDesc, ctx))
return false;
writer.incrementState();
@@ -78,10 +79,10 @@ public final class TestEnumSetMessageSerializer implements MessageSerializer<Tes
}
/** */
- @Override public final boolean readFrom(TestEnumSetMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(TestEnumSetMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
- msg.isolations = reader.readCollection(isolationsCollDesc);
+ msg.isolations = reader.readCollection(isolationsCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -89,7 +90,7 @@ public final class TestEnumSetMessageSerializer implements MessageSerializer<Tes
reader.incrementState();
case 1:
- msg.isolationsMap = reader.readMap(isolationsMapCollDesc);
+ msg.isolationsMap = reader.readMap(isolationsMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -97,7 +98,7 @@ public final class TestEnumSetMessageSerializer implements MessageSerializer<Tes
reader.incrementState();
case 2:
- msg.isolationsList = reader.readCollection(isolationsListCollDesc);
+ msg.isolationsList = reader.readCollection(isolationsListCollDesc, ctx);
if (!reader.isLastRead())
return false;
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java b/modules/core/src/test/resources/codegen/TestFeatureConflictMessage.java
similarity index 74%
copy from modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java
copy to modules/core/src/test/resources/codegen/TestFeatureConflictMessage.java
index d661f5cff66..550e305805d 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java
+++ b/modules/core/src/test/resources/codegen/TestFeatureConflictMessage.java
@@ -15,10 +15,13 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.rollingupgrade.feature;
+package org.apache.ignite.internal;
+
+import org.apache.ignite.plugin.extensions.communication.Message;
/** */
-public class TestPluginReleaseFeatures_2_0_0 {
+public class TestFeatureConflictMessage implements Message {
/** */
- public static final IgniteFeature VER_2_0_0_ID_1_FEATURE = new TestPluginFeature(1);
+ @Order(value = 0, introducedBy = "ROLLING_UPGRADE_FEATURE", deprecatedBy = "ROLLING_UPGRADE_FEATURE")
+ int fld;
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java b/modules/core/src/test/resources/codegen/TestFeatureRegistry.java
similarity index 67%
copy from modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java
copy to modules/core/src/test/resources/codegen/TestFeatureRegistry.java
index edb4b3b3646..0771430e499 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java
+++ b/modules/core/src/test/resources/codegen/TestFeatureRegistry.java
@@ -15,13 +15,16 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.rollingupgrade.feature;
+package org.apache.ignite.internal;
+
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteCoreFeature;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeature;
/** */
-public class TestPluginReleaseFeatures_2_1_0 {
+public class TestFeatureRegistry {
/** */
- public static final IgniteFeature VER_2_1_0_ID_1_FEATURE = new TestPluginFeature(1);
+ public static final IgniteFeature FIRST_FEATURE = new IgniteCoreFeature(0);
/** */
- public static final IgniteFeature VER_2_1_0_ID_2_FEATURE = new TestPluginFeature(2);
+ public static final IgniteFeature SECOND_FEATURE = new IgniteCoreFeature(1);
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java b/modules/core/src/test/resources/codegen/TestInvalidFeatureMessage.java
similarity index 74%
copy from modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java
copy to modules/core/src/test/resources/codegen/TestInvalidFeatureMessage.java
index d661f5cff66..d2d564004ac 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java
+++ b/modules/core/src/test/resources/codegen/TestInvalidFeatureMessage.java
@@ -15,10 +15,14 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.rollingupgrade.feature;
+package org.apache.ignite.internal;
+
+import org.apache.ignite.plugin.extensions.communication.Message;
/** */
-public class TestPluginReleaseFeatures_2_0_0 {
+@FeatureGated(registry = TestInvalidFeatureRegistry.class)
+public class TestInvalidFeatureMessage implements Message {
/** */
- public static final IgniteFeature VER_2_0_0_ID_1_FEATURE = new TestPluginFeature(1);
+ @Order(value = 0, introducedBy = "NOT_A_FEATURE")
+ int fld;
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java b/modules/core/src/test/resources/codegen/TestInvalidFeatureRegistry.java
similarity index 80%
copy from modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java
copy to modules/core/src/test/resources/codegen/TestInvalidFeatureRegistry.java
index d661f5cff66..b4279de561a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java
+++ b/modules/core/src/test/resources/codegen/TestInvalidFeatureRegistry.java
@@ -15,10 +15,10 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.rollingupgrade.feature;
+package org.apache.ignite.internal;
/** */
-public class TestPluginReleaseFeatures_2_0_0 {
+public class TestInvalidFeatureRegistry {
/** */
- public static final IgniteFeature VER_2_0_0_ID_1_FEATURE = new TestPluginFeature(1);
+ public static final String NOT_A_FEATURE = "not a feature";
}
diff --git a/modules/core/src/test/resources/codegen/TestMapMessageSerializer.java b/modules/core/src/test/resources/codegen/TestMapMessageSerializer.java
index 2b6aa6b27ab..d369d403380 100644
--- a/modules/core/src/test/resources/codegen/TestMapMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/TestMapMessageSerializer.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TestMapMessage;
import org.apache.ignite.plugin.extensions.communication.CollectionImplementationType;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
@@ -87,7 +88,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
private static final MessageMapType uuidStringMapCollDesc = new MessageMapType(new MessageItemType(MessageCollectionItemType.UUID), new MessageItemType(MessageCollectionItemType.STRING), false);
/** */
- @Override public final boolean writeTo(TestMapMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(TestMapMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -97,157 +98,157 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
switch (writer.state()) {
case 0:
- if (!writer.writeMap(msg.booleanArrayBoxedLongMap, booleanArrayBoxedLongMapCollDesc))
+ if (!writer.writeMap(msg.booleanArrayBoxedLongMap, booleanArrayBoxedLongMapCollDesc, ctx))
return false;
writer.incrementState();
case 1:
- if (!writer.writeMap(msg.byteArrayBooleanArrayMap, byteArrayBooleanArrayMapCollDesc))
+ if (!writer.writeMap(msg.byteArrayBooleanArrayMap, byteArrayBooleanArrayMapCollDesc, ctx))
return false;
writer.incrementState();
case 2:
- if (!writer.writeMap(msg.shortArrayByteArrayMap, shortArrayByteArrayMapCollDesc))
+ if (!writer.writeMap(msg.shortArrayByteArrayMap, shortArrayByteArrayMapCollDesc, ctx))
return false;
writer.incrementState();
case 3:
- if (!writer.writeMap(msg.intArrayShortArrayMap, intArrayShortArrayMapCollDesc))
+ if (!writer.writeMap(msg.intArrayShortArrayMap, intArrayShortArrayMapCollDesc, ctx))
return false;
writer.incrementState();
case 4:
- if (!writer.writeMap(msg.longArrayIntArrayMap, longArrayIntArrayMapCollDesc))
+ if (!writer.writeMap(msg.longArrayIntArrayMap, longArrayIntArrayMapCollDesc, ctx))
return false;
writer.incrementState();
case 5:
- if (!writer.writeMap(msg.charArrayLongArrayMap, charArrayLongArrayMapCollDesc))
+ if (!writer.writeMap(msg.charArrayLongArrayMap, charArrayLongArrayMapCollDesc, ctx))
return false;
writer.incrementState();
case 6:
- if (!writer.writeMap(msg.floatArrayCharArrayMap, floatArrayCharArrayMapCollDesc))
+ if (!writer.writeMap(msg.floatArrayCharArrayMap, floatArrayCharArrayMapCollDesc, ctx))
return false;
writer.incrementState();
case 7:
- if (!writer.writeMap(msg.doubleArrayFloatArrayMap, doubleArrayFloatArrayMapCollDesc))
+ if (!writer.writeMap(msg.doubleArrayFloatArrayMap, doubleArrayFloatArrayMapCollDesc, ctx))
return false;
writer.incrementState();
case 8:
- if (!writer.writeMap(msg.stringDoubleArrayMap, stringDoubleArrayMapCollDesc))
+ if (!writer.writeMap(msg.stringDoubleArrayMap, stringDoubleArrayMapCollDesc, ctx))
return false;
writer.incrementState();
case 9:
- if (!writer.writeMap(msg.uuidStringMap, uuidStringMapCollDesc))
+ if (!writer.writeMap(msg.uuidStringMap, uuidStringMapCollDesc, ctx))
return false;
writer.incrementState();
case 10:
- if (!writer.writeMap(msg.bitSetUuidMap, bitSetUuidMapCollDesc))
+ if (!writer.writeMap(msg.bitSetUuidMap, bitSetUuidMapCollDesc, ctx))
return false;
writer.incrementState();
case 11:
- if (!writer.writeMap(msg.igniteUuidBitSetMap, igniteUuidBitSetMapCollDesc))
+ if (!writer.writeMap(msg.igniteUuidBitSetMap, igniteUuidBitSetMapCollDesc, ctx))
return false;
writer.incrementState();
case 12:
- if (!writer.writeMap(msg.affTopVersionIgniteUuidMap, affTopVersionIgniteUuidMapCollDesc))
+ if (!writer.writeMap(msg.affTopVersionIgniteUuidMap, affTopVersionIgniteUuidMapCollDesc, ctx))
return false;
writer.incrementState();
case 13:
- if (!writer.writeMap(msg.boxedBooleanAffTopVersionMap, boxedBooleanAffTopVersionMapCollDesc))
+ if (!writer.writeMap(msg.boxedBooleanAffTopVersionMap, boxedBooleanAffTopVersionMapCollDesc, ctx))
return false;
writer.incrementState();
case 14:
- if (!writer.writeMap(msg.boxedByteBoxedBooleanMap, boxedByteBoxedBooleanMapCollDesc))
+ if (!writer.writeMap(msg.boxedByteBoxedBooleanMap, boxedByteBoxedBooleanMapCollDesc, ctx))
return false;
writer.incrementState();
case 15:
- if (!writer.writeMap(msg.boxedShortBoxedByteMap, boxedShortBoxedByteMapCollDesc))
+ if (!writer.writeMap(msg.boxedShortBoxedByteMap, boxedShortBoxedByteMapCollDesc, ctx))
return false;
writer.incrementState();
case 16:
- if (!writer.writeMap(msg.boxedIntBoxedShortMap, boxedIntBoxedShortMapCollDesc))
+ if (!writer.writeMap(msg.boxedIntBoxedShortMap, boxedIntBoxedShortMapCollDesc, ctx))
return false;
writer.incrementState();
case 17:
- if (!writer.writeMap(msg.boxedLongBoxedIntMap, boxedLongBoxedIntMapCollDesc))
+ if (!writer.writeMap(msg.boxedLongBoxedIntMap, boxedLongBoxedIntMapCollDesc, ctx))
return false;
writer.incrementState();
case 18:
- if (!writer.writeMap(msg.boxedCharBoxedLongMap, boxedCharBoxedLongMapCollDesc))
+ if (!writer.writeMap(msg.boxedCharBoxedLongMap, boxedCharBoxedLongMapCollDesc, ctx))
return false;
writer.incrementState();
case 19:
- if (!writer.writeMap(msg.boxedFloatBoxedCharMap, boxedFloatBoxedCharMapCollDesc))
+ if (!writer.writeMap(msg.boxedFloatBoxedCharMap, boxedFloatBoxedCharMapCollDesc, ctx))
return false;
writer.incrementState();
case 20:
- if (!writer.writeMap(msg.boxedDoubleBoxedFloatMap, boxedDoubleBoxedFloatMapCollDesc))
+ if (!writer.writeMap(msg.boxedDoubleBoxedFloatMap, boxedDoubleBoxedFloatMapCollDesc, ctx))
return false;
writer.incrementState();
case 21:
- if (!writer.writeMap(msg.messageBoxedDoubleMap, messageBoxedDoubleMapCollDesc))
+ if (!writer.writeMap(msg.messageBoxedDoubleMap, messageBoxedDoubleMapCollDesc, ctx))
return false;
writer.incrementState();
case 22:
- if (!writer.writeMap(msg.integerGridLongListMap, integerGridLongListMapCollDesc))
+ if (!writer.writeMap(msg.integerGridLongListMap, integerGridLongListMapCollDesc, ctx))
return false;
writer.incrementState();
case 23:
- if (!writer.writeMap(msg.gridLongListIntegerMap, gridLongListIntegerMapCollDesc))
+ if (!writer.writeMap(msg.gridLongListIntegerMap, gridLongListIntegerMapCollDesc, ctx))
return false;
writer.incrementState();
case 24:
- if (!writer.writeMap(msg.gridlistDoubleMapUuidMap, gridlistDoubleMapUuidMapCollDesc))
+ if (!writer.writeMap(msg.gridlistDoubleMapUuidMap, gridlistDoubleMapUuidMapCollDesc, ctx))
return false;
writer.incrementState();
case 25:
- if (!writer.writeMap(msg.gridCacheObjectMap, gridCacheObjectMapCollDesc))
+ if (!writer.writeMap(msg.gridCacheObjectMap, gridCacheObjectMapCollDesc, ctx))
return false;
writer.incrementState();
@@ -257,10 +258,10 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
}
/** */
- @Override public final boolean readFrom(TestMapMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(TestMapMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
- msg.booleanArrayBoxedLongMap = reader.readMap(booleanArrayBoxedLongMapCollDesc);
+ msg.booleanArrayBoxedLongMap = reader.readMap(booleanArrayBoxedLongMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -268,7 +269,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 1:
- msg.byteArrayBooleanArrayMap = reader.readMap(byteArrayBooleanArrayMapCollDesc);
+ msg.byteArrayBooleanArrayMap = reader.readMap(byteArrayBooleanArrayMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -276,7 +277,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 2:
- msg.shortArrayByteArrayMap = reader.readMap(shortArrayByteArrayMapCollDesc);
+ msg.shortArrayByteArrayMap = reader.readMap(shortArrayByteArrayMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -284,7 +285,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 3:
- msg.intArrayShortArrayMap = reader.readMap(intArrayShortArrayMapCollDesc);
+ msg.intArrayShortArrayMap = reader.readMap(intArrayShortArrayMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -292,7 +293,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 4:
- msg.longArrayIntArrayMap = reader.readMap(longArrayIntArrayMapCollDesc);
+ msg.longArrayIntArrayMap = reader.readMap(longArrayIntArrayMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -300,7 +301,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 5:
- msg.charArrayLongArrayMap = reader.readMap(charArrayLongArrayMapCollDesc);
+ msg.charArrayLongArrayMap = reader.readMap(charArrayLongArrayMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -308,7 +309,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 6:
- msg.floatArrayCharArrayMap = reader.readMap(floatArrayCharArrayMapCollDesc);
+ msg.floatArrayCharArrayMap = reader.readMap(floatArrayCharArrayMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -316,7 +317,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 7:
- msg.doubleArrayFloatArrayMap = reader.readMap(doubleArrayFloatArrayMapCollDesc);
+ msg.doubleArrayFloatArrayMap = reader.readMap(doubleArrayFloatArrayMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -324,7 +325,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 8:
- msg.stringDoubleArrayMap = reader.readMap(stringDoubleArrayMapCollDesc);
+ msg.stringDoubleArrayMap = reader.readMap(stringDoubleArrayMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -332,7 +333,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 9:
- msg.uuidStringMap = reader.readMap(uuidStringMapCollDesc);
+ msg.uuidStringMap = reader.readMap(uuidStringMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -340,7 +341,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 10:
- msg.bitSetUuidMap = reader.readMap(bitSetUuidMapCollDesc);
+ msg.bitSetUuidMap = reader.readMap(bitSetUuidMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -348,7 +349,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 11:
- msg.igniteUuidBitSetMap = reader.readMap(igniteUuidBitSetMapCollDesc);
+ msg.igniteUuidBitSetMap = reader.readMap(igniteUuidBitSetMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -356,7 +357,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 12:
- msg.affTopVersionIgniteUuidMap = reader.readMap(affTopVersionIgniteUuidMapCollDesc);
+ msg.affTopVersionIgniteUuidMap = reader.readMap(affTopVersionIgniteUuidMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -364,7 +365,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 13:
- msg.boxedBooleanAffTopVersionMap = reader.readMap(boxedBooleanAffTopVersionMapCollDesc);
+ msg.boxedBooleanAffTopVersionMap = reader.readMap(boxedBooleanAffTopVersionMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -372,7 +373,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 14:
- msg.boxedByteBoxedBooleanMap = reader.readMap(boxedByteBoxedBooleanMapCollDesc);
+ msg.boxedByteBoxedBooleanMap = reader.readMap(boxedByteBoxedBooleanMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -380,7 +381,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 15:
- msg.boxedShortBoxedByteMap = reader.readMap(boxedShortBoxedByteMapCollDesc);
+ msg.boxedShortBoxedByteMap = reader.readMap(boxedShortBoxedByteMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -388,7 +389,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 16:
- msg.boxedIntBoxedShortMap = reader.readMap(boxedIntBoxedShortMapCollDesc);
+ msg.boxedIntBoxedShortMap = reader.readMap(boxedIntBoxedShortMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -396,7 +397,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 17:
- msg.boxedLongBoxedIntMap = reader.readMap(boxedLongBoxedIntMapCollDesc);
+ msg.boxedLongBoxedIntMap = reader.readMap(boxedLongBoxedIntMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -404,7 +405,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 18:
- msg.boxedCharBoxedLongMap = reader.readMap(boxedCharBoxedLongMapCollDesc);
+ msg.boxedCharBoxedLongMap = reader.readMap(boxedCharBoxedLongMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -412,7 +413,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 19:
- msg.boxedFloatBoxedCharMap = reader.readMap(boxedFloatBoxedCharMapCollDesc);
+ msg.boxedFloatBoxedCharMap = reader.readMap(boxedFloatBoxedCharMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -420,7 +421,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 20:
- msg.boxedDoubleBoxedFloatMap = reader.readMap(boxedDoubleBoxedFloatMapCollDesc);
+ msg.boxedDoubleBoxedFloatMap = reader.readMap(boxedDoubleBoxedFloatMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -428,7 +429,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 21:
- msg.messageBoxedDoubleMap = reader.readMap(messageBoxedDoubleMapCollDesc);
+ msg.messageBoxedDoubleMap = reader.readMap(messageBoxedDoubleMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -436,7 +437,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 22:
- msg.integerGridLongListMap = reader.readMap(integerGridLongListMapCollDesc);
+ msg.integerGridLongListMap = reader.readMap(integerGridLongListMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -444,7 +445,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 23:
- msg.gridLongListIntegerMap = reader.readMap(gridLongListIntegerMapCollDesc);
+ msg.gridLongListIntegerMap = reader.readMap(gridLongListIntegerMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -452,7 +453,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 24:
- msg.gridlistDoubleMapUuidMap = reader.readMap(gridlistDoubleMapUuidMapCollDesc);
+ msg.gridlistDoubleMapUuidMap = reader.readMap(gridlistDoubleMapUuidMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -460,7 +461,7 @@ public final class TestMapMessageSerializer implements MessageSerializer<TestMap
reader.incrementState();
case 25:
- msg.gridCacheObjectMap = reader.readMap(gridCacheObjectMapCollDesc);
+ msg.gridCacheObjectMap = reader.readMap(gridCacheObjectMapCollDesc, ctx);
if (!reader.isLastRead())
return false;
diff --git a/modules/core/src/test/resources/codegen/TestMarshallableMessageSerializer.java b/modules/core/src/test/resources/codegen/TestMarshallableMessageSerializer.java
index b48bc85b38b..92ed2a2001d 100644
--- a/modules/core/src/test/resources/codegen/TestMarshallableMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/TestMarshallableMessageSerializer.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TestMarshallableMessage;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
@@ -29,7 +30,7 @@ import org.apache.ignite.plugin.extensions.communication.MessageWriter;
*/
public final class TestMarshallableMessageSerializer implements MessageSerializer<TestMarshallableMessage> {
/** */
- @Override public final boolean writeTo(TestMarshallableMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(TestMarshallableMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -61,7 +62,7 @@ public final class TestMarshallableMessageSerializer implements MessageSerialize
}
/** */
- @Override public final boolean readFrom(TestMarshallableMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(TestMarshallableMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
msg.iv = reader.readInt();
diff --git a/modules/core/src/test/resources/codegen/TestMarshalledArrayMapMessageSerializer.java b/modules/core/src/test/resources/codegen/TestMarshalledArrayMapMessageSerializer.java
index 4f3a7c0e51f..52f8785c633 100644
--- a/modules/core/src/test/resources/codegen/TestMarshalledArrayMapMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/TestMarshalledArrayMapMessageSerializer.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal;
import java.util.List;
import org.apache.ignite.internal.GridTopicMessage;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TestMarshalledArrayMapMessage;
import org.apache.ignite.plugin.extensions.communication.CollectionImplementationType;
import org.apache.ignite.plugin.extensions.communication.MessageArrayType;
@@ -45,7 +46,7 @@ public final class TestMarshalledArrayMapMessageSerializer implements MessageSer
private static final MessageArrayType mapValsCollDesc = new MessageArrayType(new MessageCollectionType(new MessageItemType(MessageCollectionItemType.MSG), CollectionImplementationType.ARRAY_LIST), List.class);
/** */
- @Override public final boolean writeTo(TestMarshalledArrayMapMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(TestMarshalledArrayMapMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -55,25 +56,25 @@ public final class TestMarshalledArrayMapMessageSerializer implements MessageSer
switch (writer.state()) {
case 0:
- if (!writer.writeObjectArray(msg.mapKeys, mapKeysCollDesc))
+ if (!writer.writeObjectArray(msg.mapKeys, mapKeysCollDesc, ctx))
return false;
writer.incrementState();
case 1:
- if (!writer.writeObjectArray(msg.mapVals, mapValsCollDesc))
+ if (!writer.writeObjectArray(msg.mapVals, mapValsCollDesc, ctx))
return false;
writer.incrementState();
case 2:
- if (!writer.writeObjectArray(msg.fixedMapKeys, fixedMapKeysCollDesc))
+ if (!writer.writeObjectArray(msg.fixedMapKeys, fixedMapKeysCollDesc, ctx))
return false;
writer.incrementState();
case 3:
- if (!writer.writeObjectArray(msg.fixedMapVals, fixedMapValsCollDesc))
+ if (!writer.writeObjectArray(msg.fixedMapVals, fixedMapValsCollDesc, ctx))
return false;
writer.incrementState();
@@ -83,10 +84,10 @@ public final class TestMarshalledArrayMapMessageSerializer implements MessageSer
}
/** */
- @Override public final boolean readFrom(TestMarshalledArrayMapMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(TestMarshalledArrayMapMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
- msg.mapKeys = reader.readObjectArray(mapKeysCollDesc);
+ msg.mapKeys = reader.readObjectArray(mapKeysCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -94,7 +95,7 @@ public final class TestMarshalledArrayMapMessageSerializer implements MessageSer
reader.incrementState();
case 1:
- msg.mapVals = reader.readObjectArray(mapValsCollDesc);
+ msg.mapVals = reader.readObjectArray(mapValsCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -102,7 +103,7 @@ public final class TestMarshalledArrayMapMessageSerializer implements MessageSer
reader.incrementState();
case 2:
- msg.fixedMapKeys = reader.readObjectArray(fixedMapKeysCollDesc);
+ msg.fixedMapKeys = reader.readObjectArray(fixedMapKeysCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -110,7 +111,7 @@ public final class TestMarshalledArrayMapMessageSerializer implements MessageSer
reader.incrementState();
case 3:
- msg.fixedMapVals = reader.readObjectArray(fixedMapValsCollDesc);
+ msg.fixedMapVals = reader.readObjectArray(fixedMapValsCollDesc, ctx);
if (!reader.isLastRead())
return false;
diff --git a/modules/core/src/test/resources/codegen/TestMarshalledCollectionMessageSerializer.java b/modules/core/src/test/resources/codegen/TestMarshalledCollectionMessageSerializer.java
index 43f7bf2f2e9..7b831f8ec04 100644
--- a/modules/core/src/test/resources/codegen/TestMarshalledCollectionMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/TestMarshalledCollectionMessageSerializer.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TestMarshalledCollectionMessage;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.plugin.extensions.communication.MessageArrayType;
@@ -36,7 +37,7 @@ public final class TestMarshalledCollectionMessageSerializer implements MessageS
private static final MessageArrayType keysArrCollDesc = new MessageArrayType(new MessageItemType(MessageCollectionItemType.GRID_CACHE_VERSION), GridCacheVersion.class);
/** */
- @Override public final boolean writeTo(TestMarshalledCollectionMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(TestMarshalledCollectionMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -46,7 +47,7 @@ public final class TestMarshalledCollectionMessageSerializer implements MessageS
switch (writer.state()) {
case 0:
- if (!writer.writeObjectArray(msg.keysArr, keysArrCollDesc))
+ if (!writer.writeObjectArray(msg.keysArr, keysArrCollDesc, ctx))
return false;
writer.incrementState();
@@ -56,10 +57,10 @@ public final class TestMarshalledCollectionMessageSerializer implements MessageS
}
/** */
- @Override public final boolean readFrom(TestMarshalledCollectionMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(TestMarshalledCollectionMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
- msg.keysArr = reader.readObjectArray(keysArrCollDesc);
+ msg.keysArr = reader.readObjectArray(keysArrCollDesc, ctx);
if (!reader.isLastRead())
return false;
diff --git a/modules/core/src/test/resources/codegen/TestMarshalledMapMessageSerializer.java b/modules/core/src/test/resources/codegen/TestMarshalledMapMessageSerializer.java
index dd0fcb22533..e3aca2bf684 100644
--- a/modules/core/src/test/resources/codegen/TestMarshalledMapMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/TestMarshalledMapMessageSerializer.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TestMarshalledMapMessage;
import org.apache.ignite.plugin.extensions.communication.CollectionImplementationType;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
@@ -38,7 +39,7 @@ public final class TestMarshalledMapMessageSerializer implements MessageSerializ
private static final MessageCollectionType mapValsCollDesc = new MessageCollectionType(new MessageItemType(MessageCollectionItemType.GRID_CACHE_VERSION), CollectionImplementationType.ARRAY_LIST);
/** */
- @Override public final boolean writeTo(TestMarshalledMapMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(TestMarshalledMapMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -48,13 +49,13 @@ public final class TestMarshalledMapMessageSerializer implements MessageSerializ
switch (writer.state()) {
case 0:
- if (!writer.writeCollection(msg.mapKeys, mapKeysCollDesc))
+ if (!writer.writeCollection(msg.mapKeys, mapKeysCollDesc, ctx))
return false;
writer.incrementState();
case 1:
- if (!writer.writeCollection(msg.mapVals, mapValsCollDesc))
+ if (!writer.writeCollection(msg.mapVals, mapValsCollDesc, ctx))
return false;
writer.incrementState();
@@ -64,10 +65,10 @@ public final class TestMarshalledMapMessageSerializer implements MessageSerializ
}
/** */
- @Override public final boolean readFrom(TestMarshalledMapMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(TestMarshalledMapMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
- msg.mapKeys = reader.readCollection(mapKeysCollDesc);
+ msg.mapKeys = reader.readCollection(mapKeysCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -75,7 +76,7 @@ public final class TestMarshalledMapMessageSerializer implements MessageSerializ
reader.incrementState();
case 1:
- msg.mapVals = reader.readCollection(mapValsCollDesc);
+ msg.mapVals = reader.readCollection(mapValsCollDesc, ctx);
if (!reader.isLastRead())
return false;
diff --git a/modules/core/src/test/resources/codegen/TestMarshalledMessageSerializer.java b/modules/core/src/test/resources/codegen/TestMarshalledMessageSerializer.java
index fa81057b604..b924687a0d0 100644
--- a/modules/core/src/test/resources/codegen/TestMarshalledMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/TestMarshalledMessageSerializer.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TestMarshalledMessage;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
@@ -29,7 +30,7 @@ import org.apache.ignite.plugin.extensions.communication.MessageWriter;
*/
public final class TestMarshalledMessageSerializer implements MessageSerializer<TestMarshalledMessage> {
/** */
- @Override public final boolean writeTo(TestMarshalledMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(TestMarshalledMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -49,7 +50,7 @@ public final class TestMarshalledMessageSerializer implements MessageSerializer<
}
/** */
- @Override public final boolean readFrom(TestMarshalledMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(TestMarshalledMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
msg.dataBytes = reader.readByteArray();
diff --git a/modules/core/src/test/resources/codegen/TestMarshalledObjectsMessageSerializer.java b/modules/core/src/test/resources/codegen/TestMarshalledObjectsMessageSerializer.java
index b68578b06a6..68d62d88791 100644
--- a/modules/core/src/test/resources/codegen/TestMarshalledObjectsMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/TestMarshalledObjectsMessageSerializer.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TestMarshalledObjectsMessage;
import org.apache.ignite.plugin.extensions.communication.CollectionImplementationType;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
@@ -36,7 +37,7 @@ public final class TestMarshalledObjectsMessageSerializer implements MessageSeri
private static final MessageCollectionType dataBytesCollDesc = new MessageCollectionType(new MessageItemType(MessageCollectionItemType.BYTE_ARR), CollectionImplementationType.ARRAY_LIST);
/** */
- @Override public final boolean writeTo(TestMarshalledObjectsMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(TestMarshalledObjectsMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -46,7 +47,7 @@ public final class TestMarshalledObjectsMessageSerializer implements MessageSeri
switch (writer.state()) {
case 0:
- if (!writer.writeCollection(msg.dataBytes, dataBytesCollDesc))
+ if (!writer.writeCollection(msg.dataBytes, dataBytesCollDesc, ctx))
return false;
writer.incrementState();
@@ -56,10 +57,10 @@ public final class TestMarshalledObjectsMessageSerializer implements MessageSeri
}
/** */
- @Override public final boolean readFrom(TestMarshalledObjectsMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(TestMarshalledObjectsMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
- msg.dataBytes = reader.readCollection(dataBytesCollDesc);
+ msg.dataBytes = reader.readCollection(dataBytesCollDesc, ctx);
if (!reader.isLastRead())
return false;
diff --git a/modules/core/src/test/resources/codegen/TestMessageSerializer.java b/modules/core/src/test/resources/codegen/TestMessageSerializer.java
index 7aa0a054471..1811c09f590 100644
--- a/modules/core/src/test/resources/codegen/TestMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/TestMessageSerializer.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.TestMessage;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.plugin.extensions.communication.MessageArrayType;
@@ -40,7 +41,7 @@ public final class TestMessageSerializer implements MessageSerializer<TestMessag
private static final MessageArrayType verArrCollDesc = new MessageArrayType(new MessageItemType(MessageCollectionItemType.GRID_CACHE_VERSION), GridCacheVersion.class);
/** */
- @Override public final boolean writeTo(TestMessage msg, MessageWriter writer) {
+ @Override public final boolean writeTo(TestMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
if (!writer.isHeaderWritten()) {
if (!writer.writeHeader(msg.directType()))
return false;
@@ -68,13 +69,13 @@ public final class TestMessageSerializer implements MessageSerializer<TestMessag
writer.incrementState();
case 3:
- if (!writer.writeObjectArray(msg.strArr, strArrCollDesc))
+ if (!writer.writeObjectArray(msg.strArr, strArrCollDesc, ctx))
return false;
writer.incrementState();
case 4:
- if (!writer.writeObjectArray(msg.intMatrix, intMatrixCollDesc))
+ if (!writer.writeObjectArray(msg.intMatrix, intMatrixCollDesc, ctx))
return false;
writer.incrementState();
@@ -86,7 +87,7 @@ public final class TestMessageSerializer implements MessageSerializer<TestMessag
writer.incrementState();
case 6:
- if (!writer.writeObjectArray(msg.verArr, verArrCollDesc))
+ if (!writer.writeObjectArray(msg.verArr, verArrCollDesc, ctx))
return false;
writer.incrementState();
@@ -140,7 +141,7 @@ public final class TestMessageSerializer implements MessageSerializer<TestMessag
writer.incrementState();
case 15:
- if (!writer.writeMessage(msg.nioMsg))
+ if (!writer.writeMessage(msg.nioMsg, ctx))
return false;
writer.incrementState();
@@ -150,7 +151,7 @@ public final class TestMessageSerializer implements MessageSerializer<TestMessag
}
/** */
- @Override public final boolean readFrom(TestMessage msg, MessageReader reader) {
+ @Override public final boolean readFrom(TestMessage msg, MessageReader reader, MessageSerializationContext ctx) {
switch (reader.state()) {
case 0:
msg.id = reader.readInt();
@@ -177,7 +178,7 @@ public final class TestMessageSerializer implements MessageSerializer<TestMessag
reader.incrementState();
case 3:
- msg.strArr = reader.readObjectArray(strArrCollDesc);
+ msg.strArr = reader.readObjectArray(strArrCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -185,7 +186,7 @@ public final class TestMessageSerializer implements MessageSerializer<TestMessag
reader.incrementState();
case 4:
- msg.intMatrix = reader.readObjectArray(intMatrixCollDesc);
+ msg.intMatrix = reader.readObjectArray(intMatrixCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -201,7 +202,7 @@ public final class TestMessageSerializer implements MessageSerializer<TestMessag
reader.incrementState();
case 6:
- msg.verArr = reader.readObjectArray(verArrCollDesc);
+ msg.verArr = reader.readObjectArray(verArrCollDesc, ctx);
if (!reader.isLastRead())
return false;
@@ -273,7 +274,7 @@ public final class TestMessageSerializer implements MessageSerializer<TestMessag
reader.incrementState();
case 15:
- msg.nioMsg = reader.readMessage();
+ msg.nioMsg = reader.readMessage(ctx);
if (!reader.isLastRead())
return false;
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java b/modules/core/src/test/resources/codegen/TestRollingUpgradeAwareMessage.java
similarity index 57%
copy from modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java
copy to modules/core/src/test/resources/codegen/TestRollingUpgradeAwareMessage.java
index edb4b3b3646..1bec6706646 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_1_0.java
+++ b/modules/core/src/test/resources/codegen/TestRollingUpgradeAwareMessage.java
@@ -15,13 +15,31 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.rollingupgrade.feature;
+package org.apache.ignite.internal;
+
+import org.apache.ignite.plugin.extensions.communication.Message;
/** */
-public class TestPluginReleaseFeatures_2_1_0 {
+@FeatureGated(registry = TestFeatureRegistry.class)
+public class TestRollingUpgradeAwareMessage implements Message {
+ /** */
+ @Order(0)
+ int plain;
+
/** */
- public static final IgniteFeature VER_2_1_0_ID_1_FEATURE = new TestPluginFeature(1);
+ @Order(value = 1, deprecatedBy = "FIRST_FEATURE")
+ String oldFld;
/** */
- public static final IgniteFeature VER_2_1_0_ID_2_FEATURE = new TestPluginFeature(2);
+ @Order(value = 2, introducedBy = "FIRST_FEATURE")
+ String newFld;
+
+ /** */
+ @Order(value = 3, introducedBy = "FIRST_FEATURE", deprecatedBy = "SECOND_FEATURE")
+ long windowed;
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 0;
+ }
}
diff --git a/modules/core/src/test/resources/codegen/TestRollingUpgradeAwareMessageSerializer.java b/modules/core/src/test/resources/codegen/TestRollingUpgradeAwareMessageSerializer.java
new file mode 100644
index 00000000000..f14fe2a71e3
--- /dev/null
+++ b/modules/core/src/test/resources/codegen/TestRollingUpgradeAwareMessageSerializer.java
@@ -0,0 +1,127 @@
+/*
+ * 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.MessageSerializationContext;
+import org.apache.ignite.internal.TestFeatureRegistry;
+import org.apache.ignite.internal.TestRollingUpgradeAwareMessage;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+
+/**
+ * This class is generated automatically.
+ *
+ * @see org.apache.ignite.internal.MessageProcessor
+ */
+public final class TestRollingUpgradeAwareMessageSerializer implements MessageSerializer<TestRollingUpgradeAwareMessage> {
+ /** */
+ @Override public final boolean writeTo(TestRollingUpgradeAwareMessage msg, MessageWriter writer, MessageSerializationContext ctx) {
+ if (!writer.isHeaderWritten()) {
+ if (!writer.writeHeader(msg.directType()))
+ return false;
+
+ writer.onHeaderWritten();
+ }
+
+ switch (writer.state()) {
+ case 0:
+ if (!writer.writeInt(msg.plain))
+ return false;
+
+ writer.incrementState();
+
+ case 1:
+ if (ctx.includeFieldDeprecatedBy(TestFeatureRegistry.FIRST_FEATURE)) {
+ if (!writer.writeString(msg.oldFld))
+ return false;
+ }
+
+ writer.incrementState();
+
+ case 2:
+ if (ctx.includeFieldIntroducedBy(TestFeatureRegistry.FIRST_FEATURE)) {
+ if (!writer.writeString(msg.newFld))
+ return false;
+ }
+
+ writer.incrementState();
+
+ case 3:
+ if (ctx.includeFieldIntroducedBy(TestFeatureRegistry.FIRST_FEATURE) && ctx.includeFieldDeprecatedBy(TestFeatureRegistry.SECOND_FEATURE)) {
+ if (!writer.writeLong(msg.windowed))
+ return false;
+ }
+
+ writer.incrementState();
+
+ }
+
+ return true;
+ }
+
+ /** */
+ @Override public final boolean readFrom(TestRollingUpgradeAwareMessage msg, MessageReader reader, MessageSerializationContext ctx) {
+ switch (reader.state()) {
+ case 0:
+ msg.plain = reader.readInt();
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 1:
+ if (ctx.includeFieldDeprecatedBy(TestFeatureRegistry.FIRST_FEATURE)) {
+ msg.oldFld = reader.readString();
+
+ if (!reader.isLastRead())
+ return false;
+ }
+
+ reader.incrementState();
+
+ case 2:
+ if (ctx.includeFieldIntroducedBy(TestFeatureRegistry.FIRST_FEATURE)) {
+ msg.newFld = reader.readString();
+
+ if (!reader.isLastRead())
+ return false;
+ }
+
+ reader.incrementState();
+
+ case 3:
+ if (ctx.includeFieldIntroducedBy(TestFeatureRegistry.FIRST_FEATURE) && ctx.includeFieldDeprecatedBy(TestFeatureRegistry.SECOND_FEATURE)) {
+ msg.windowed = reader.readLong();
+
+ if (!reader.isLastRead())
+ return false;
+ }
+
+ reader.incrementState();
+
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public final TestRollingUpgradeAwareMessage createMessage() {
+ return new TestRollingUpgradeAwareMessage();
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java b/modules/core/src/test/resources/codegen/TestUnknownFeatureMessage.java
similarity index 78%
copy from modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java
copy to modules/core/src/test/resources/codegen/TestUnknownFeatureMessage.java
index d661f5cff66..7e1db62bb29 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/TestPluginReleaseFeatures_2_0_0.java
+++ b/modules/core/src/test/resources/codegen/TestUnknownFeatureMessage.java
@@ -15,10 +15,13 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.rollingupgrade.feature;
+package org.apache.ignite.internal;
+
+import org.apache.ignite.plugin.extensions.communication.Message;
/** */
-public class TestPluginReleaseFeatures_2_0_0 {
+public class TestUnknownFeatureMessage implements Message {
/** */
- public static final IgniteFeature VER_2_0_0_ID_1_FEATURE = new TestPluginFeature(1);
+ @Order(value = 0, introducedBy = "NO_SUCH_FEATURE")
+ int fld;
}
diff --git a/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java b/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
index f30fbc5cfc0..43f89a6ab11 100644
--- a/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
+++ b/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
@@ -29,6 +29,7 @@ import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.util.CommonUtils.makeMessageType;
+import static org.apache.ignite.internal.util.nio.MessageSerialization.resolveSerializationContext;
/**
* Parser for direct messages.
@@ -86,7 +87,7 @@ public class GridDirectParser implements GridNioParser {
if (msg != null && buf.hasRemaining()) {
reader.setBuffer(buf);
- finished = MessageSerialization.readFrom(msgFactory, msg, reader);
+ finished = MessageSerialization.readFrom(msgFactory, msg, reader, resolveSerializationContext(ses));
}
if (finished) {
diff --git a/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java b/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
index a067277648d..09926571b29 100644
--- a/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
+++ b/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
@@ -54,6 +54,7 @@ import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.processors.odbc.ClientMessage;
import org.apache.ignite.internal.util.CommonUtils;
import org.apache.ignite.internal.util.GridConcurrentHashSet;
@@ -85,6 +86,7 @@ import static org.apache.ignite.failure.FailureType.CRITICAL_ERROR;
import static org.apache.ignite.failure.FailureType.SYSTEM_WORKER_TERMINATION;
import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.MSG_WRITER;
import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.NIO_OPERATION;
+import static org.apache.ignite.internal.util.nio.MessageSerialization.resolveSerializationContext;
/**
* TCP NIO server. Due to asynchronous nature of connections processing
@@ -1443,6 +1445,8 @@ public class GridNioServer<T> {
MessageWriter writer = messageWriter(ses);
+ MessageSerializationContext serCtx = resolveSerializationContext(ses);
+
boolean handshakeFinished = sslFilter.lock(ses);
try {
@@ -1506,7 +1510,7 @@ public class GridNioServer<T> {
List<SessionWriteRequest> pendingRequests = new ArrayList<>(2);
if (req != null)
- finished = writeToBuffer(writer, buf, req, pendingRequests);
+ finished = writeToBuffer(writer, buf, req, pendingRequests, serCtx);
// Fill up as many messages as possible to write buffer.
while (finished) {
@@ -1518,7 +1522,7 @@ public class GridNioServer<T> {
if (req == null)
break;
- finished = writeToBuffer(writer, buf, req, pendingRequests);
+ finished = writeToBuffer(writer, buf, req, pendingRequests, serCtx);
}
int sesBufLimit = buf.limit();
@@ -1592,13 +1596,15 @@ public class GridNioServer<T> {
* @param buf Buffer to write.
* @param req Source of data.
* @param pendingRequests List of requests which was successfully written.
+ * @param serCtx Serialization context of the session.
* @return {@code true} if message successfully written to buffer and {@code false} otherwise.
*/
private boolean writeToBuffer(
MessageWriter writer,
ByteBuffer buf,
SessionWriteRequest req,
- List<SessionWriteRequest> pendingRequests
+ List<SessionWriteRequest> pendingRequests,
+ MessageSerializationContext serCtx
) {
Message msg;
boolean finished;
@@ -1614,7 +1620,7 @@ public class GridNioServer<T> {
else {
writer.setBuffer(buf);
- finished = MessageSerialization.writeTo(messageFactory(), msg, writer);
+ finished = MessageSerialization.writeTo(messageFactory(), msg, writer, serCtx);
}
if (finished) {
@@ -1692,6 +1698,8 @@ public class GridNioServer<T> {
MessageWriter writer = messageWriter(ses);
+ MessageSerializationContext serCtx = resolveSerializationContext(ses);
+
if (req == null) {
req = systemMessage(ses);
@@ -1709,7 +1717,7 @@ public class GridNioServer<T> {
boolean finished = false;
if (req != null)
- finished = writeToBuffer(ses, buf, req, writer);
+ finished = writeToBuffer(ses, buf, req, writer, serCtx);
// Fill up as many messages as possible to write buffer.
while (finished) {
@@ -1723,7 +1731,7 @@ public class GridNioServer<T> {
if (req == null)
break;
- finished = writeToBuffer(ses, buf, req, writer);
+ finished = writeToBuffer(ses, buf, req, writer, serCtx);
}
buf.flip();
@@ -1781,14 +1789,20 @@ public class GridNioServer<T> {
}
/**
- * @param writer Customizer of writing.
+ * @param ses Session the message is written to.
* @param buf Buffer to write.
* @param req Source of data.
- * @param ses Session for notification about writting.
+ * @param writer Customizer of writing.
+ * @param serCtx Serialization context of the session.
* @return {@code true} if message successfully written to buffer and {@code false} otherwise.
*/
- private boolean writeToBuffer(GridSelectorNioSessionImpl ses, ByteBuffer buf, SessionWriteRequest req,
- MessageWriter writer) {
+ private boolean writeToBuffer(
+ GridSelectorNioSessionImpl ses,
+ ByteBuffer buf,
+ SessionWriteRequest req,
+ MessageWriter writer,
+ MessageSerializationContext serCtx
+ ) {
Message msg;
boolean finished;
msg = (Message)req.message();
@@ -1803,7 +1817,7 @@ public class GridNioServer<T> {
else {
writer.setBuffer(buf);
- finished = MessageSerialization.writeTo(msgFactory, msg, writer);
+ finished = MessageSerialization.writeTo(msgFactory, msg, writer, serCtx);
}
if (finished) {
diff --git a/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioSessionMetaKey.java b/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioSessionMetaKey.java
index fbdcc6d5d34..9a5836d14cc 100644
--- a/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioSessionMetaKey.java
+++ b/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioSessionMetaKey.java
@@ -42,7 +42,10 @@ public enum GridNioSessionMetaKey {
MARSHALLER_ID,
/** Message writer. */
- MSG_WRITER;
+ MSG_WRITER,
+
+ /** Message serialization context. */
+ MSG_SER_CTX;
/** Maximum count of NIO session keys in system. */
public static final int MAX_KEYS_CNT = 64;
diff --git a/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/MessageSerialization.java b/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/MessageSerialization.java
index b436edb764d..ee47ffd3c77 100644
--- a/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/MessageSerialization.java
+++ b/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/MessageSerialization.java
@@ -17,11 +17,13 @@
package org.apache.ignite.internal.util.nio;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import org.jetbrains.annotations.Nullable;
/**
* Resolve-and-dispatch entry points for {@link MessageSerializer}: each looks up the serializer registered for the
@@ -40,11 +42,19 @@ public final class MessageSerialization {
* @param factory Message factory.
* @param msg Message instance.
* @param writer Writer.
+ * @param ctx Serialization context.
* @param <M> Message type.
* @return Whether message was fully written.
*/
- public static <M extends Message> boolean writeTo(MessageFactory factory, M msg, MessageWriter writer) {
- return resolve(factory, msg).writeTo(msg, writer);
+ public static <M extends Message> boolean writeTo(
+ MessageFactory factory,
+ M msg,
+ MessageWriter writer,
+ MessageSerializationContext ctx
+ ) {
+ assert ctx != null : "Message is serialized without a serialization context [msg=" + msg.getClass().getName() + ']';
+
+ return resolveMessageserializer(factory, msg).writeTo(msg, writer, ctx);
}
/**
@@ -53,16 +63,29 @@ public final class MessageSerialization {
* @param factory Message factory.
* @param msg Message instance.
* @param reader Reader.
+ * @param ctx Serialization context.
* @param <M> Message type.
* @return Whether message was fully read.
*/
- public static <M extends Message> boolean readFrom(MessageFactory factory, M msg, MessageReader reader) {
- return resolve(factory, msg).readFrom(msg, reader);
+ public static <M extends Message> boolean readFrom(
+ MessageFactory factory,
+ M msg,
+ MessageReader reader,
+ MessageSerializationContext ctx
+ ) {
+ assert ctx != null : "Message is deserialized without a serialization context [msg=" + msg.getClass().getName() + ']';
+
+ return resolveMessageserializer(factory, msg).readFrom(msg, reader, ctx);
+ }
+
+ /** */
+ public static @Nullable MessageSerializationContext resolveSerializationContext(GridNioSession ses) {
+ return ses.meta(GridNioSessionMetaKey.MSG_SER_CTX.ordinal());
}
/** @return the serializer registered for {@code msg}'s direct type. */
@SuppressWarnings("unchecked")
- private static <M extends Message> MessageSerializer<M> resolve(MessageFactory factory, M msg) {
+ private static <M extends Message> MessageSerializer<M> resolveMessageserializer(MessageFactory factory, M msg) {
return (MessageSerializer<M>)factory.serializer(msg.directType());
}
}
diff --git a/modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageReader.java b/modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageReader.java
index 796acc40a19..faf50669796 100644
--- a/modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageReader.java
+++ b/modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageReader.java
@@ -22,6 +22,7 @@ import java.util.BitSet;
import java.util.Collection;
import java.util.Map;
import java.util.UUID;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
@@ -196,21 +197,23 @@ public interface MessageReader {
/**
* Reads nested message.
*
+ * @param ctx Serialization context.
* @param <T> Type of the message.
* @return Message.
*/
- public default <T extends Message> T readMessage() {
- return readMessage(false);
+ public default <T extends Message> T readMessage(MessageSerializationContext ctx) {
+ return readMessage(false, ctx);
}
/**
* Reads nested message.
*
* @param compress Whether message should be decompressed.
+ * @param ctx Serialization context.
* @param <T> Type of the message.
* @return Message.
*/
- public <T extends Message> T readMessage(boolean compress);
+ public <T extends Message> T readMessage(boolean compress, MessageSerializationContext ctx);
/**
* Reads {@link CacheObject}.
@@ -237,29 +240,32 @@ public interface MessageReader {
* Reads array of objects.
*
* @param type Array component type.
+ * @param ctx Serialization context.
* @param <T> Type of the read object.
* @return Array of objects.
*/
- public <T> T[] readObjectArray(MessageArrayType type);
+ public <T> T[] readObjectArray(MessageArrayType type, MessageSerializationContext ctx);
/**
* Reads any collection.
*
* @param type Collection item type.
+ * @param ctx Serialization context.
* @param <C> Type of the read collection.
* @return Collection.
*/
- public <C extends Collection<?>> C readCollection(MessageCollectionType type);
+ public <C extends Collection<?>> C readCollection(MessageCollectionType type, MessageSerializationContext ctx);
/**
* Reads map.
*
* @param type Map type.
+ * @param ctx Serialization context.
* @param <M> Type of the read map.
* @return Map.
*/
- public default <M extends Map<?, ?>> M readMap(MessageMapType type) {
- return readMap(type, false);
+ public default <M extends Map<?, ?>> M readMap(MessageMapType type, MessageSerializationContext ctx) {
+ return readMap(type, false, ctx);
}
/**
@@ -267,10 +273,11 @@ public interface MessageReader {
*
* @param type Map type.
* @param compress Whether map should be compressed.
+ * @param ctx Serialization context.
* @param <M> Type of the read map.
* @return Map.
*/
- public <M extends Map<?, ?>> M readMap(MessageMapType type, boolean compress);
+ public <M extends Map<?, ?>> M readMap(MessageMapType type, boolean compress, MessageSerializationContext ctx);
/** @return Ignite product version. */
IgniteProductVersion readIgniteProductVersion();
diff --git a/modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageSerializer.java b/modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageSerializer.java
index 86ce0d170a0..ab66f296ed6 100644
--- a/modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageSerializer.java
+++ b/modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageSerializer.java
@@ -17,6 +17,8 @@
package org.apache.ignite.plugin.extensions.communication;
+import org.apache.ignite.internal.MessageSerializationContext;
+
/**
* Interface for message serialization logic. Resolve-and-dispatch entry points that look the serializer up from the
* message factory live in {@code MessageSerialization}.
@@ -27,18 +29,20 @@ public interface MessageSerializer<M extends Message> {
*
* @param msg Message instance.
* @param writer Writer.
+ * @param ctx Serialization context.
* @return Whether message was fully written.
*/
- public boolean writeTo(M msg, MessageWriter writer);
+ public boolean writeTo(M msg, MessageWriter writer, MessageSerializationContext ctx);
/**
* Reads this message from provided byte buffer.
*
* @param msg Message instance.
* @param reader Reader.
+ * @param ctx Serialization context.
* @return Whether message was fully read.
*/
- public boolean readFrom(M msg, MessageReader reader);
+ public boolean readFrom(M msg, MessageReader reader, MessageSerializationContext ctx);
/**
* @return New instance of message.
diff --git a/modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageWriter.java b/modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageWriter.java
index 019bab8fa9d..4f281cec3a8 100644
--- a/modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageWriter.java
+++ b/modules/nio/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageWriter.java
@@ -22,6 +22,7 @@ import java.util.BitSet;
import java.util.Collection;
import java.util.Map;
import java.util.UUID;
+import org.apache.ignite.internal.MessageSerializationContext;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
@@ -246,10 +247,11 @@ public interface MessageWriter {
* Writes nested message.
*
* @param val Message.
+ * @param ctx Serialization context.
* @return Whether value was fully written.
*/
- public default boolean writeMessage(Message val) {
- return writeMessage(val, false);
+ public default boolean writeMessage(Message val, MessageSerializationContext ctx) {
+ return writeMessage(val, false, ctx);
}
/**
@@ -257,9 +259,10 @@ public interface MessageWriter {
*
* @param val Message.
* @param compress Whether message should be compressed.
+ * @param ctx Serialization context.
* @return Whether value was fully written.
*/
- public boolean writeMessage(Message val, boolean compress);
+ public boolean writeMessage(Message val, boolean compress, MessageSerializationContext ctx);
/**
* Writes {@link CacheObject}.
@@ -290,32 +293,35 @@ public interface MessageWriter {
*
* @param arr Array of objects.
* @param type Array component type.
+ * @param ctx Serialization context.
* @param <T> Type of the objects that array contains.
* @return Whether array was fully written.
*/
- public <T> boolean writeObjectArray(T[] arr, MessageArrayType type);
+ public <T> boolean writeObjectArray(T[] arr, MessageArrayType type, MessageSerializationContext ctx);
/**
* Writes collection with its elements order.
*
* @param col Collection.
* @param type Collection item type.
+ * @param ctx Serialization context.
* @param <T> Type of the objects that collection contains.
* @return Whether value was fully written.
*/
- public <T> boolean writeCollection(Collection<T> col, MessageCollectionType type);
+ public <T> boolean writeCollection(Collection<T> col, MessageCollectionType type, MessageSerializationContext ctx);
/**
* Writes map.
*
* @param map Map.
* @param type Map type.
+ * @param ctx Serialization context.
* @param <K> Initial key types of the map to write.
* @param <V> Initial value types of the map to write.
* @return Whether value was fully written.
*/
- public default <K, V> boolean writeMap(Map<K, V> map, MessageMapType type) {
- return writeMap(map, type, false);
+ public default <K, V> boolean writeMap(Map<K, V> map, MessageMapType type, MessageSerializationContext ctx) {
+ return writeMap(map, type, false, ctx);
}
/**
@@ -324,11 +330,12 @@ public interface MessageWriter {
* @param map Map.
* @param type Map type.
* @param compress Whether map should be compressed.
+ * @param ctx Serialization context.
* @param <K> Initial key types of the map to write.
* @param <V> Initial value types of the map to write.
* @return Whether value was fully written.
*/
- public <K, V> boolean writeMap(Map<K, V> map, MessageMapType type, boolean compress);
+ public <K, V> boolean writeMap(Map<K, V> map, MessageMapType type, boolean compress, MessageSerializationContext ctx);
/**
* Writes ignite product version.
diff --git a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/DiscoveryMessageParser.java b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/DiscoveryMessageParser.java
index aeb7c1be885..06a14735271 100644
--- a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/DiscoveryMessageParser.java
+++ b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/DiscoveryMessageParser.java
@@ -37,6 +37,8 @@ import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
import org.apache.ignite.spi.IgniteSpiException;
+import static org.apache.ignite.internal.MessageSerializationContext.LATEST_SCHEMA;
+
/**
* Class is responsible for serializing discovery messages using RU-ready {@link MessageSerializer} mechanism.
*/
@@ -105,7 +107,7 @@ public class DiscoveryMessageParser {
do {
msgBuf.clear();
- finished = MessageSerialization.writeTo(msgFactory, m, msgWriter);
+ finished = MessageSerialization.writeTo(msgFactory, m, msgWriter, LATEST_SCHEMA);
out.write(msgBuf.array(), 0, msgBuf.position());
}
@@ -131,7 +133,7 @@ public class DiscoveryMessageParser {
msgBuf.rewind();
}
- finished = MessageSerialization.readFrom(msgFactory, msg, msgReader);
+ finished = MessageSerialization.readFrom(msgFactory, msg, msgReader, LATEST_SCHEMA);
assert read != -1 || finished : "Stream closed before message was fully read.";