T readMessage() throws IgniteCheckedException, IOException {
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
@@ -262,39 +275,13 @@ public Socket socket() {
return sock;
}
- /**
- * Serializes a discovery message into given output stream.
- *
- * @param m Discovery message to serialize.
- * @param out Output stream to write serialized message.
- * @throws IOException If serialization fails.
- */
- void serializeMessage(Message m, OutputStream out) throws IOException, IgniteCheckedException {
- DiscoveryMarshalling.marshal(m, ctx, null);
-
- msgWriter.reset();
- msgWriter.setBuffer(writeBuf);
-
- boolean finished;
-
- do {
- // Should be cleared before first operation.
- writeBuf.clear();
-
- finished = MessageSerialization.writeTo(msgFactory, m, msgWriter);
-
- out.write(writeBuf.array(), 0, writeBuf.position());
- }
- while (!finished);
- }
-
/**
* Writes raw data to the underlying socket output stream.
*
* @param data Raw data to write.
* @throws IOException If failed.
*/
- void write(byte[] data) throws IOException {
+ synchronized void write(byte[] data) throws IOException {
out.write(data);
out.flush();
@@ -306,7 +293,7 @@ void write(byte[] data) throws IOException {
* @param b Integer response.
* @throws IOException If failed.
*/
- void write(int b) throws IOException {
+ synchronized void write(int b) throws IOException {
out.write(b);
out.flush();
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryMessageSerializer.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryMessageSerializer.java
deleted file mode 100644
index ec7cdc569f0c7..0000000000000
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryMessageSerializer.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * 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.spi.discovery.tcp;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.Socket;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.GridKernalContext;
-import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
-import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
-
-/**
- * Class is responsible for serializing discovery messages using RU-ready {@link MessageSerializer} mechanism.
- *
- * It is used in a special case: when server wants to send discovery messages to clients, it may not have a {@link TcpDiscoveryIoSession}
- * to serialize the messages.
- * This class enables server to serialize discovery messages anyway, duplicating serialization code from {@link TcpDiscoveryIoSession}.
- */
-class TcpDiscoveryMessageSerializer extends TcpDiscoveryIoSession {
- /**
- * @param ctx Kernal context.
- */
- public TcpDiscoveryMessageSerializer(GridKernalContext ctx) {
- super(ctx, new Socket() {
- @Override public OutputStream getOutputStream() throws IOException {
- return null;
- }
-
- @Override public InputStream getInputStream() throws IOException {
- return null;
- }
- });
- }
-
- /**
- * Serializes a discovery message into a byte array.
- *
- * @param msg Discovery message to serialize.
- * @return Serialized byte array containing the message data.
- * @throws IgniteCheckedException If serialization fails.
- * @throws IOException If serialization fails.
- */
- byte[] serializeMessage(TcpDiscoveryAbstractMessage msg) throws IgniteCheckedException, IOException {
- try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
- serializeMessage((Message)msg, out);
-
- return out.toByteArray();
- }
- }
-}
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 7b8490ded8385..c2e1f64295681 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.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.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;
@@ -1680,36 +1677,6 @@ Socket createSocket() throws IOException {
}
}
- /** */
- 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
new file mode 100644
index 0000000000000..e663ab47795fc
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/ClientMessageHolder.java
@@ -0,0 +1,62 @@
+/*
+ * 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.spi.discovery.tcp.internal;
+
+import java.util.HashMap;
+import java.util.Map;
+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;
+
+/** */
+public class ClientMessageHolder {
+ /** */
+ private final TcpDiscoveryAbstractMessage msg;
+
+ /** */
+ private final Map bytesByCtx = new HashMap<>(1);
+
+ /** */
+ public ClientMessageHolder(TcpDiscoveryAbstractMessage msg) {
+ assert msg != null;
+
+ this.msg = msg;
+ }
+
+ /** */
+ public TcpDiscoveryAbstractMessage message() {
+ return msg;
+ }
+
+ /** */
+ public synchronized byte @Nullable [] messageBytes(MessageSerializationContext ctx) {
+ return bytesByCtx.get(ctx);
+ }
+
+ /** */
+ public synchronized void serialize(TcpDiscoveryMessageSerializer ser, MessageSerializationContext ctx) throws IgniteCheckedException {
+ if (!bytesByCtx.containsKey(ctx))
+ bytesByCtx.put(ctx, ser.serialize(msg, ctx));
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return msg.toString();
+ }
+}
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
new file mode 100644
index 0000000000000..9102876206bd4
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryMessageSerializer.java
@@ -0,0 +1,106 @@
+/*
+ * 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.spi.discovery.tcp.internal;
+
+import java.io.IOException;
+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;
+import org.apache.ignite.internal.util.nio.MessageSerialization;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
+
+/** */
+public class TcpDiscoveryMessageSerializer {
+ /** Size of the intermediate buffer a message is serialized through. */
+ private static final int BUFFER_SIZE = 100;
+
+ /** */
+ private final GridKernalContext ctx;
+
+ /** */
+ private final DirectMessageWriter writer;
+
+ /** */
+ private final ByteBuffer buf = ByteBuffer.allocate(BUFFER_SIZE);
+
+ /** @param ctx Kernal context. */
+ public TcpDiscoveryMessageSerializer(GridKernalContext ctx) {
+ this.ctx = ctx;
+
+ writer = new DirectMessageWriter(ctx.messageFactory());
+ }
+
+ /**
+ * Serializes a discovery message into given output stream.
+ *
+ * @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,
+ MessageSerializationContext serCtx
+ ) throws IgniteCheckedException, IOException {
+ DiscoveryMarshalling.marshal(msg, ctx, null);
+
+ writer.reset();
+ writer.setBuffer(buf);
+
+ boolean finished;
+
+ do {
+ // Should be cleared before first operation.
+ buf.clear();
+
+ finished = MessageSerialization.writeTo(ctx.messageFactory(), msg, writer, serCtx);
+
+ out.write(buf.array(), 0, buf.position());
+ }
+ while (!finished);
+ }
+
+ /**
+ * 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,
+ MessageSerializationContext serCtx
+ ) throws IgniteCheckedException {
+ try (GridByteArrayOutputStream out = new GridByteArrayOutputStream()) {
+ writeTo(msg, out, serCtx);
+
+ return out.toByteArray();
+ }
+ catch (IOException e) {
+ throw new IgniteCheckedException("Failed to serialize a discovery message: " + msg, e);
+ }
+ }
+}
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 1105f85cf3fd9..1f8a288af4d3f 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 void testProcessorGeneratesSerializer() {
.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 testSameFeatureInBothGuardsRejected() {
+ 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 e9874e03055fc..a3e66c91819bd 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.transactions.TransactionIsolation;
import org.junit.Test;
+import static org.apache.ignite.internal.MessageSerializationContext.IGNORED;
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 @@ private T doMarshalUnmarshalChunked(T srcMsg) {
writer.setBuffer(chunk);
- fullyWritten = writer.writeMessage(srcMsg, false);
+ fullyWritten = writer.writeMessage(srcMsg, false, IGNORED);
chunk.flip();
@@ -168,7 +169,7 @@ private T doMarshalUnmarshalChunked(T srcMsg) {
reader.setBuffer(chunk);
- resMsg = reader.readMessage(false);
+ resMsg = reader.readMessage(false, IGNORED);
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 bfa3c8e314204..2472a1f440db9 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,11 +25,13 @@
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;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.thread.context.OperationContextSnapshotMessage;
import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.nio.MessageSerialization;
import org.apache.ignite.lang.IgniteProductVersion;
@@ -47,6 +49,7 @@
import org.junit.Test;
import static java.lang.Integer.MAX_VALUE;
+import static org.apache.ignite.internal.MessageSerializationContext.IGNORED;
import static org.apache.ignite.plugin.extensions.communication.CollectionImplementationType.HASH_SET;
import static org.junit.Assert.assertEquals;
@@ -104,15 +107,19 @@ private void checkSerializationAndDeserializationConsistency(
Message msg = msgFactory.create(msgType);
+ // OperationContextSnapshotMessage uses custom serialization that is incompatible with test reader and writer implementation.
+ if (msg instanceof OperationContextSnapshotMessage)
+ return;
+
initializeMessage(msg);
- while (!MessageSerialization.writeTo(msgFactory, msg, writer)) {
+ while (!MessageSerialization.writeTo(msgFactory, msg, writer, IGNORED)) {
// No-op.
}
msg = msgFactory.create(msgType);
- while (!MessageSerialization.readFrom(msgFactory, msg, reader)) {
+ while (!MessageSerialization.readFrom(msgFactory, msg, reader, IGNORED)) {
// No-op.
}
@@ -295,22 +302,22 @@ private boolean writeField(Class type) {
}
/** {@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 boolean writeObjectArray(T[] arr, MessageArrayType type) {
+ @Override public boolean writeObjectArray(T[] arr, MessageArrayType type, MessageSerializationContext ctx) {
return writeField(Object[].class);
}
/** {@inheritDoc} */
- @Override public boolean writeCollection(Collection col, MessageCollectionType type) {
+ @Override public boolean writeCollection(Collection col, MessageCollectionType type, MessageSerializationContext ctx) {
return writeField(type.collectionImplementationType() == HASH_SET ? Set.class : Collection.class);
}
/** {@inheritDoc} */
- @Override public boolean writeMap(Map map, MessageMapType type, boolean compress) {
+ @Override public boolean writeMap(Map map, MessageMapType type, boolean compress, MessageSerializationContext ctx) {
return writeField(type.linked() ? LinkedHashMap.class : HashMap.class);
}
@@ -537,7 +544,7 @@ private void readField(Class type) {
}
/** {@inheritDoc} */
- @Override public T readMessage(boolean compress) {
+ @Override public T readMessage(boolean compress, MessageSerializationContext ctx) {
readField(Message.class);
return null;
@@ -565,21 +572,21 @@ private void readField(Class type) {
}
/** {@inheritDoc} */
- @Override public T[] readObjectArray(MessageArrayType type) {
+ @Override public T[] readObjectArray(MessageArrayType type, MessageSerializationContext ctx) {
readField(Object[].class);
return null;
}
/** {@inheritDoc} */
- @Override public > C readCollection(MessageCollectionType type) {
+ @Override public > C readCollection(MessageCollectionType type, MessageSerializationContext ctx) {
readField(type.collectionImplementationType() == HASH_SET ? Set.class : Collection.class);
return null;
}
/** {@inheritDoc} */
- @Override public > M readMap(MessageMapType type, boolean compress) {
+ @Override public > 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 046dbbb6f55b0..28c5b23e347be 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.testframework.GridTestUtils;
import org.junit.Test;
+import static org.apache.ignite.internal.MessageSerializationContext.IGNORED;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -70,7 +71,7 @@ public void testWriteReadHugeMessage() {
ByteBuffer msgBuf = ByteBuffer.allocate(40_960);
while (!finished) {
- finished = writer.writeMessage(fullMsg, true);
+ finished = writer.writeMessage(fullMsg, true, IGNORED);
if (checkChunkCnt) {
DirectMessageState state = U.field(writer, "state");
@@ -104,7 +105,7 @@ public void testWriteReadHugeMessage() {
reader.setBuffer(msgBuf);
- Message readMsg = reader.readMessage(true);
+ Message readMsg = reader.readMessage(true, IGNORED);
assertTrue(readMsg instanceof GridDhtPartitionsFullMessage);
@@ -133,7 +134,7 @@ public void testReadFailsOnNullChunk() {
reader.setBuffer(buf);
GridTestUtils.assertThrows(null,
- () -> MessageSerialization.readFrom(MSG_FACTORY, new CompressedMessage(), reader),
+ () -> MessageSerialization.readFrom(MSG_FACTORY, new CompressedMessage(), reader, IGNORED),
IgniteException.class,
"unexpected null chunk");
}
@@ -156,7 +157,7 @@ public void testReadFailsOnNegativeDataSize() {
reader.setBuffer(buf);
GridTestUtils.assertThrows(null,
- () -> MessageSerialization.readFrom(MSG_FACTORY, new CompressedMessage(), reader),
+ () -> MessageSerialization.readFrom(MSG_FACTORY, new CompressedMessage(), reader, IGNORED),
IgniteException.class,
"Invalid compressed message data size");
}
@@ -217,7 +218,7 @@ public void testReadFailsOnTruncatedPayload() {
writer.setBuffer(tmpBuf);
- assertTrue(writer.writeMessage(fullMessage(), false));
+ assertTrue(writer.writeMessage(fullMessage(), false, IGNORED));
tmpBuf.flip();
@@ -231,7 +232,7 @@ public void testReadFailsOnTruncatedPayload() {
wireWriter.setBuffer(wire);
- assertTrue(wireWriter.writeMessage(compressedMsg, false));
+ assertTrue(wireWriter.writeMessage(compressedMsg, false, IGNORED));
wire.flip();
@@ -239,7 +240,7 @@ public void testReadFailsOnTruncatedPayload() {
reader.setBuffer(wire);
- GridTestUtils.assertThrows(null, () -> reader.readMessage(true), IgniteException.class, "ended unexpectedly");
+ GridTestUtils.assertThrows(null, () -> reader.readMessage(true, IGNORED), 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 4b51337eac099..2a5ce91b96302 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.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 @@ private static class FailingUnmarshalMessage implements Message {
/** Writes the two fields behind the header. */
private static class Serializer implements MessageSerializer {
/** {@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 @@ private static class Serializer implements MessageSerializer
@@ -121,7 +123,7 @@ public void testCacheSize() throws Exception {
// 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, IGNORED));
assertTrue(msgWritter.getBuffer().hasRemaining());
@@ -133,7 +135,7 @@ public void testCacheSize() throws Exception {
TcpDiscoveryMetricsUpdateMessage msg2 = new TcpDiscoveryMetricsUpdateMessage();
- assertTrue(MessageSerialization.readFrom(msgFactory, msg2, msgReader));
+ assertTrue(MessageSerialization.readFrom(msgFactory, msg2, msgReader, IGNORED));
Map 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 45a5fb3381fd5..4f6851c31a6e6 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 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.IGNORED;
/**
*
@@ -204,7 +205,7 @@ private CacheContinuousQueryEntry roundTrip(CacheContinuousQueryEntry e) throws
// Skip write class header.
writer.onHeaderWritten();
- MessageSerialization.writeTo(msgFactory, e, writer);
+ MessageSerialization.writeTo(msgFactory, e, writer, IGNORED);
CacheContinuousQueryEntry res = new CacheContinuousQueryEntry();
@@ -212,7 +213,7 @@ private CacheContinuousQueryEntry roundTrip(CacheContinuousQueryEntry e) throws
reader.setBuffer(ByteBuffer.wrap(buf.array()));
- MessageSerialization.readFrom(msgFactory, res, reader);
+ MessageSerialization.readFrom(msgFactory, res, reader, IGNORED);
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 f7fea391425dd..60d05ca3feddf 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.common.GridCommonAbstractTest;
import org.junit.Test;
+import static org.apache.ignite.internal.MessageSerializationContext.IGNORED;
import static org.apache.ignite.internal.util.CommonUtils.makeMessageType;
/** Test for serialization round-trip of {@link QueryEntityMessage} and {@link QueryEntityExMessage}. */
@@ -173,7 +174,7 @@ private T writeAndReadBack(T msg, long expReadsWritesCnt) th
DirectMessageWriter writer = new DirectMessageWriter(msgFactory);
writer.setBuffer(buf);
- assertTrue(MessageSerialization.writeTo(msgFactory, msg, writer));
+ assertTrue(MessageSerialization.writeTo(msgFactory, msg, writer, IGNORED));
assertEquals("Writes" + ERROR_SUFFIX,
expReadsWritesCnt, writer.state());
@@ -184,7 +185,7 @@ private T writeAndReadBack(T msg, long expReadsWritesCnt) th
T res = (T)msgFactory.create(makeMessageType(buf.get(), buf.get()));
- assertTrue(MessageSerialization.readFrom(msgFactory, res, reader));
+ assertTrue(MessageSerialization.readFrom(msgFactory, res, reader, IGNORED));
assertEquals("Reads" + ERROR_SUFFIX,
expReadsWritesCnt, reader.state());
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/AbstractRollingUpgradeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/AbstractRollingUpgradeTest.java
index 0bbdb9824a57d..ef1e198be1257 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/AbstractRollingUpgradeTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/AbstractRollingUpgradeTest.java
@@ -58,6 +58,7 @@
import org.apache.ignite.internal.processors.rollingupgrade.feature.TestPluginComponentFeatureSetProvider;
import org.apache.ignite.internal.processors.rollingupgrade.feature.TestPluginFeature;
import org.apache.ignite.internal.processors.rollingupgrade.feature.TestPluginReleaseFeatures_1_0_0;
+import org.apache.ignite.internal.thread.context.AbstractDistributedAttributeTest;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.lang.ConsumerX;
import org.apache.ignite.internal.util.typedef.F;
@@ -73,7 +74,6 @@
import org.apache.ignite.spi.discovery.tcp.TestBlockingTcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.internal.UnsupportedNodeVersionException;
import org.apache.ignite.testframework.GridTestUtils;
-import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.jspecify.annotations.Nullable;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -132,7 +132,7 @@
*
*
*/
-public abstract class AbstractRollingUpgradeTest extends GridCommonAbstractTest {
+public abstract class AbstractRollingUpgradeTest extends AbstractDistributedAttributeTest {
/** */
protected static final String TEST_DEFAULT_VER = "2.19.0";
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 d661f5cff66bc..066d0403917df 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 @@
/** */
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 edb4b3b3646ac..385f7ded6a900 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 @@
/** */
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/AbstractRollingUpgradeMessageTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/AbstractRollingUpgradeMessageTest.java
new file mode 100644
index 0000000000000..da30b7662ad62
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/AbstractRollingUpgradeMessageTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.processors.rollingupgrade.AbstractRollingUpgradeTest;
+import org.apache.ignite.internal.thread.context.OperationContext;
+import org.apache.ignite.internal.thread.context.OperationContextAttribute;
+import org.apache.ignite.internal.thread.context.OperationContextSnapshot;
+import org.apache.ignite.internal.thread.context.Scope;
+import org.apache.ignite.plugin.extensions.communication.Message;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.ignite.internal.managers.communication.GridIoPolicy.PUBLIC_POOL;
+
+/** */
+public abstract class AbstractRollingUpgradeMessageTest extends AbstractRollingUpgradeTest {
+ /** */
+ protected void startServerNodes(String... vers) throws Exception {
+ IgniteEx first = startGrid(0, vers[0]);
+
+ if (Arrays.stream(vers).distinct().count() > 1)
+ ru(first).enableVersionUpgrade();
+
+ for (int idx = 1; idx < vers.length; idx++)
+ startGrid(idx, vers[idx]);
+ }
+
+ /** */
+ protected Received send(IgniteEx from, IgniteEx to, T msg) throws Exception {
+ AtomicReference> got = new AtomicReference<>();
+ CountDownLatch latch = new CountDownLatch(1);
+
+ String topic = msg.getClass().getName();
+
+ to.context().io().addMessageListener(topic, (nodeId, rcvd, plc) -> {
+ got.set(new Received<>((T)rcvd));
+
+ latch.countDown();
+ });
+
+ ClusterNode rcvNode = from.context().discovery().node(to.localNode().id());
+
+ from.context().io().sendToCustomTopic(rcvNode, topic, msg, PUBLIC_POOL);
+
+ assertTrue(latch.await(getTestTimeout(), MILLISECONDS));
+
+ return got.get();
+ }
+
+ /** */
+ protected Map> sendOverDiscovery(IgniteEx from, T msg) throws Exception {
+ List clusterNodes = Ignition.allGrids();
+
+ Map> receivedMsgs = new ConcurrentHashMap<>();
+
+ CountDownLatch latch = new CountDownLatch(clusterNodes.size());
+
+ for (Ignite rcv : clusterNodes) {
+ String name = rcv.name();
+
+ ((IgniteEx)rcv).context().discovery().setCustomEventListener((Class)msg.getClass(), (v, n, m) -> {
+ receivedMsgs.put(name, new Received<>(m));
+
+ latch.countDown();
+ });
+ }
+
+ from.context().discovery().sendCustomEvent(msg);
+
+ assertTrue(latch.await(getTestTimeout(), MILLISECONDS));
+
+ receivedMsgs.remove(from.name());
+
+ return receivedMsgs;
+ }
+
+ /** */
+ protected static class Received {
+ /** */
+ final T msg;
+
+ /** */
+ private final OperationContextSnapshot opCtx;
+
+ /** */
+ private Received(T msg) {
+ this.msg = msg;
+
+ opCtx = OperationContext.createSnapshot();
+ }
+
+ /** */
+ V attribute(OperationContextAttribute attr) {
+ try (Scope ignored = OperationContext.restoreSnapshot(opCtx)) {
+ return OperationContext.get(attr);
+ }
+ }
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/RollingUpgradeDistributedAttributeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/RollingUpgradeDistributedAttributeTest.java
new file mode 100644
index 0000000000000..0d00ee7fa128e
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/RollingUpgradeDistributedAttributeTest.java
@@ -0,0 +1,193 @@
+/*
+ * 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.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.processors.authentication.User;
+import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
+import org.apache.ignite.internal.thread.context.DistributedAttributeKey;
+import org.apache.ignite.internal.thread.context.OperationContext;
+import org.apache.ignite.internal.thread.context.OperationContextAttribute;
+import org.apache.ignite.internal.thread.context.Scope;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.plugin.AbstractTestPluginProvider;
+import org.apache.ignite.plugin.PluginContext;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.MessagesPluginProvider;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+
+import static org.apache.ignite.internal.processors.rollingupgrade.feature.TestIgniteReleaseFeatures_2_20_0.VER_2_20_0_ID_3_FEATURE;
+import static org.apache.ignite.internal.processors.rollingupgrade.message.RollingUpgradeDistributedAttributeTest.TestIgniteComponent.PTR_VAL;
+import static org.apache.ignite.internal.processors.rollingupgrade.message.RollingUpgradeDistributedAttributeTest.TestIgniteComponent.USR_VAL;
+import static org.apache.ignite.internal.processors.rollingupgrade.message.RollingUpgradeDistributedAttributeTest.TestIgniteComponent.VER_2_19_PTR_ATTR;
+import static org.apache.ignite.internal.processors.rollingupgrade.message.RollingUpgradeDistributedAttributeTest.TestIgniteComponent.VER_2_20_USR_ATTR;
+
+/** */
+public class RollingUpgradeDistributedAttributeTest extends AbstractRollingUpgradeMessageTest {
+ /** */
+ private static final DistributedAttributeKey VER_2_19_ATTR_KEY = createTestKey(0);
+
+ /** */
+ private static final DistributedAttributeKey VER_2_20_ATTR_KEY = createTestKey(7, VER_2_20_0_ID_3_FEATURE);
+
+ /** {@inheritDoc} */
+ @Override protected Collection distributedAttributeKeys() {
+ return Arrays.asList(VER_2_19_ATTR_KEY, VER_2_20_ATTR_KEY);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName, String ver) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName, ver);
+
+ cfg.setPluginProviders(F.concat(
+ cfg.getPluginProviders(),
+ new MessagesPluginProvider(TestCoreMessage.class),
+ new TestIgniteComponent()));
+
+ return cfg;
+ }
+
+ /** */
+ @Test
+ public void testNewAttributeIsCutForOldPeer() throws Exception {
+ startServerNodes("2.19.0", "2.20.0");
+
+ checkMutualSend(grid(0), grid(1), PTR_VAL, null);
+ }
+
+ /** */
+ @Test
+ public void testNewAttributeIsCutForPeerReadingCompactLayout() throws Exception {
+ startServerNodes("2.19.2", "2.20.0");
+
+ checkMutualSend(grid(0), grid(1), PTR_VAL, null);
+ }
+
+ /** */
+ @Test
+ public void testBothAttributesReachPeerOfSameNewRelease() throws Exception {
+ startServerNodes("2.20.0", "2.20.0");
+
+ checkMutualSend(grid(0), grid(1), PTR_VAL, USR_VAL);
+ }
+
+ /** */
+ @Test
+ public void testNewAttributeIsCutAroundRingWithOldCoordinator() throws Exception {
+ startServerNodes("2.19.0", "2.20.0", "2.20.0");
+
+ Map> rcvd = sendOverDiscovery(grid(1), TestCoreMessage.build());
+
+ assertAttributes(PTR_VAL, null, rcvd.get(grid(0).name()));
+ assertAttributes(PTR_VAL, null, rcvd.get(grid(2).name()));
+
+ assertAttributes(PTR_VAL, USR_VAL, send(grid(1), grid(2), TestCoreMessage.build()));
+ }
+
+ /** */
+ @Test
+ public void testNewAttributeReachesOnlyClientOfNewRelease() 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> rcvd = sendOverDiscovery(grid(1), TestCoreMessage.build());
+
+ assertAttributes(PTR_VAL, USR_VAL, rcvd.get(newVerCli.name()));
+ assertAttributes(PTR_VAL, null, rcvd.get(oldVerCli.name()));
+
+ assertAttributes(PTR_VAL, USR_VAL, send(grid(1), newVerCli, TestCoreMessage.build()));
+ assertAttributes(PTR_VAL, null, send(grid(1), oldVerCli, TestCoreMessage.build()));
+ }
+
+ /** */
+ private void checkMutualSend(IgniteEx first, IgniteEx second, WALPointer expPtr, @Nullable User expUsr) throws Exception {
+ assertAttributes(expPtr, expUsr, send(first, second, TestCoreMessage.build()));
+ assertAttributes(expPtr, expUsr, send(second, first, TestCoreMessage.build()));
+
+ assertAttributes(expPtr, expUsr, sendOverDiscovery(first, TestCoreMessage.build()).get(second.name()));
+ assertAttributes(expPtr, expUsr, sendOverDiscovery(second, TestCoreMessage.build()).get(first.name()));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected Map> sendOverDiscovery(
+ IgniteEx from,
+ T msg
+ ) throws Exception {
+ try (Scope ignored = OperationContext.set(VER_2_19_PTR_ATTR, PTR_VAL, VER_2_20_USR_ATTR, USR_VAL)) {
+ return super.sendOverDiscovery(from, msg);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected Received send(IgniteEx from, IgniteEx to, T msg) throws Exception {
+ try (Scope ignored = OperationContext.set(VER_2_19_PTR_ATTR, PTR_VAL, VER_2_20_USR_ATTR, USR_VAL)) {
+ return super.send(from, to, msg);
+ }
+ }
+
+ /** */
+ private static void assertAttributes(WALPointer expBase, @Nullable User expNew, Received rcvd) {
+ assertEquals(expBase, rcvd.attribute(VER_2_19_PTR_ATTR));
+ assertEquals(expNew, rcvd.attribute(VER_2_20_USR_ATTR));
+ }
+
+ /** */
+ static class TestIgniteComponent extends AbstractTestPluginProvider {
+ /** */
+ public static final OperationContextAttribute VER_2_19_PTR_ATTR = OperationContextAttribute.newInstance();
+
+ /** */
+ public static final OperationContextAttribute VER_2_20_USR_ATTR = OperationContextAttribute.newInstance();
+
+ /** */
+ public static final WALPointer PTR_VAL = new WALPointer(1, 1, 1);
+
+ /** */
+ public static final User USR_VAL = User.create("1", "1");
+
+ /** {@inheritDoc} */
+ @Override public String name() {
+ return "TestIgniteComponent";
+ }
+
+ /** {@inheritDoc} */
+ @Override public void start(PluginContext ctx) {
+ GridKernalContext kctx = ((IgniteEx)ctx.grid()).context();
+
+ kctx.operationContextDispatcher().registerDistributedAttribute(VER_2_19_ATTR_KEY, VER_2_19_PTR_ATTR);
+
+ if (kctx.localNodeFeatures().contains(VER_2_20_0_ID_3_FEATURE))
+ kctx.operationContextDispatcher().registerDistributedAttribute(VER_2_20_ATTR_KEY, VER_2_20_USR_ATTR);
+ }
+ }
+}
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 0000000000000..d42fb160864db
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/RollingUpgradeMessageSerializationTest.java
@@ -0,0 +1,441 @@
+/*
+ * 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.function.Supplier;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+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 AbstractRollingUpgradeMessageTest {
+ /** {@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> receivedMsgs = sendOverDiscovery(grid(1), TestCoreMessage.build());
+
+ assertReceived(A, B, C, D, E, null, receivedMsgs.get(newVerCli.name()));
+ assertReceived(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 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> receivedMsgs = sendOverDiscovery(from, TestCoreMessage.build()).values();
+
+ for (Received rcvd : receivedMsgs)
+ assertReceived(expA, expB, expC, expD, expE, expF, rcvd);
+ }
+
+ /** */
+ private void checkMutualMessageSend(
+ IgniteEx first,
+ IgniteEx second,
+ Supplier 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 void checkReceivedMessageFields(
+ IgniteEx from,
+ IgniteEx to,
+ Supplier msgFactory,
+ String expA,
+ String expB,
+ String expC,
+ String expD,
+ String expE,
+ String expF
+ ) throws Exception {
+ assertReceived(expA, expB, expC, expD, expE, expF, send(from, to, msgFactory.get()));
+
+ assertReceived(expA, expB, expC, expD, expE, expF, sendOverDiscovery(from, msgFactory.get()).get(to.name()));
+ }
+
+ /** */
+ private static void assertReceived(
+ String expA,
+ String expB,
+ String expC,
+ String expD,
+ String expE,
+ String expF,
+ Received rcvd
+ ) {
+ TestMessage msg = rcvd.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 0000000000000..843c928ad1ca0
--- /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.FeatureRegistry;
+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;
+
+/** */
+@FeatureRegistry(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 0000000000000..02e6173990b11
--- /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/message/TestMessage.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestMessage.java
new file mode 100644
index 0000000000000..15c639bf6fdaa
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/message/TestMessage.java
@@ -0,0 +1,57 @@
+/*
+ * 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;
+
+/** */
+public interface TestMessage {
+ /** */
+ public static final String A = "A";
+
+ /** */
+ 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 0000000000000..4bdf8ebe80ac3
--- /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.FeatureRegistry;
+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;
+
+/** */
+@FeatureRegistry(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 1536b53e9926d..a978a506abb89 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.common.GridCommonAbstractTest;
import org.junit.Test;
+import static org.apache.ignite.internal.MessageSerializationContext.IGNORED;
import static org.apache.ignite.internal.util.CommonUtils.makeMessageType;
import static org.junit.Assert.assertArrayEquals;
@@ -126,7 +127,7 @@ private T writeAndReadBack(T msg, long expReadsWritesCnt) th
DirectMessageWriter writer = new DirectMessageWriter(msgFactory);
writer.setBuffer(buf);
- assertTrue(MessageSerialization.writeTo(msgFactory, msg, writer));
+ assertTrue(MessageSerialization.writeTo(msgFactory, msg, writer, IGNORED));
assertEquals("Writes" + ERROR_SUFFIX, expReadsWritesCnt, writer.state());
buf.flip();
@@ -136,7 +137,7 @@ private T writeAndReadBack(T msg, long expReadsWritesCnt) th
T res = (T)msgFactory.create(makeMessageType(buf.get(), buf.get()));
- assertTrue(MessageSerialization.readFrom(msgFactory, res, reader));
+ assertTrue(MessageSerialization.readFrom(msgFactory, res, reader, IGNORED));
assertEquals("Reads" + ERROR_SUFFIX, expReadsWritesCnt, reader.state());
DiscoveryMarshalling.unmarshal(res, kctx);
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/thread/context/AbstractDistributedAttributeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/AbstractDistributedAttributeTest.java
new file mode 100644
index 0000000000000..33c8a3ca37fec
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/AbstractDistributedAttributeTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.thread.context;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeature;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.internal.thread.context.DistributedAttributeKeyRegistry.VALS;
+
+/** */
+public abstract class AbstractDistributedAttributeTest extends GridCommonAbstractTest {
+ /** */
+ private DistributedAttributeKey[] originalAttrKeys;
+
+ /** */
+ protected Collection distributedAttributeKeys() {
+ return Collections.emptyList();
+ }
+
+ /** */
+ protected static DistributedAttributeKey createTestKey(int id) {
+ return new DistributedAttributeKey(id);
+ }
+
+ /** */
+ protected static DistributedAttributeKey createTestKey(int id, IgniteFeature introducedBy) {
+ return new DistributedAttributeKey(id, introducedBy);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ Collection testKeys = distributedAttributeKeys();
+
+ if (!testKeys.isEmpty()) {
+ originalAttrKeys = Arrays.copyOf(VALS, VALS.length);
+
+ for (DistributedAttributeKey key : testKeys) {
+ VALS[key.id()] = key;
+ }
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+
+ if (originalAttrKeys != null) {
+ System.arraycopy(originalAttrKeys, 0, VALS, 0, originalAttrKeys.length);
+
+ originalAttrKeys = null;
+ }
+
+ super.afterTestsStopped();
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributePropagationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributePropagationTest.java
index ea4978a711ee1..5e1e38f0e8508 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributePropagationTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributePropagationTest.java
@@ -18,6 +18,8 @@
package org.apache.ignite.internal.thread.context;
import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -45,7 +47,6 @@
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.ListeningTestLogger;
import org.apache.ignite.testframework.LogListener;
-import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.jetbrains.annotations.Nullable;
import org.junit.Test;
@@ -67,15 +68,26 @@
import static org.junit.Assume.assumeFalse;
/** */
-public class OperationContextAttributePropagationTest extends GridCommonAbstractTest {
+public class OperationContextAttributePropagationTest extends AbstractDistributedAttributeTest {
/** */
public static final LogListener MSG_DELAYED_LSNR = LogListener.builder().andMatches(
"Delay custom message processing, there are joining nodes"
).build();
+ /** */
+ public static final DistributedAttributeKey PTR_KEY = createTestKey(0);
+
+ /** */
+ public static final DistributedAttributeKey USR_KEY = createTestKey(MAX_ATTRS_CNT - 1);
+
/** */
private volatile Consumer discoMsgLsnr;
+ /** {@inheritDoc} */
+ @Override protected Collection distributedAttributeKeys() {
+ return Arrays.asList(PTR_KEY, USR_KEY);
+ }
+
/** {@inheritDoc} */
@Override protected void beforeTest() throws Exception {
super.beforeTest();
@@ -223,7 +235,7 @@ private void prepareCluster() throws Exception {
assertThrows(
null,
- () -> grid(0).context().operationContextDispatcher().registerDistributedAttribute(1, null),
+ () -> grid(0).context().operationContextDispatcher().registerDistributedAttribute(PTR_KEY, null),
IgniteException.class,
"Initialization of distributed operation context attributes has already finished"
);
@@ -348,7 +360,7 @@ private void checkOperationContextCommunicationTransmission(
}
}
- /** Prevents {@link ClusterNode#isLocal()} to be negative. */
+ /** */
private ClusterNode node(Ignite from, Ignite to) {
return from.cluster().node(((IgniteEx)to).localNode().id());
}
@@ -404,13 +416,13 @@ static class TestIgniteComponent extends AbstractTestPluginProvider {
@Override public void start(PluginContext ctx) {
kctx = ((IgniteEx)ctx.grid()).context();
- kctx.operationContextDispatcher().registerDistributedAttribute(MAX_ATTRS_CNT - 1, USR_ATTR);
- kctx.operationContextDispatcher().registerDistributedAttribute(0, PTR_ATTR);
+ kctx.operationContextDispatcher().registerDistributedAttribute(USR_KEY, USR_ATTR);
+ kctx.operationContextDispatcher().registerDistributedAttribute(PTR_KEY, PTR_ATTR);
assertThrowsAnyCause(
log,
() -> {
- kctx.operationContextDispatcher().registerDistributedAttribute(MAX_ATTRS_CNT - 1, PTR_ATTR);
+ kctx.operationContextDispatcher().registerDistributedAttribute(USR_KEY, PTR_ATTR);
return null;
}, IgniteException.class,
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 0286297db11e4..8a8f0521739c4 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.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 PayloadMessage() {
/** */
private static class PayloadSerializer implements MessageSerializer {
/** {@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 @@ private static class PayloadSerializer implements MessageSerializer {
/** {@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 @@ private static class Serializer implements MessageSerializer {
/** {@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 @@ private static class RetrySerializer implements MessageSerializer T writeAndReadBack(T msg) throws IgniteCheckedExcept
DirectMessageWriter writer = new DirectMessageWriter(msgFactory);
writer.setBuffer(buf);
- assertTrue(MessageSerialization.writeTo(msgFactory, msg, writer));
+ assertTrue(MessageSerialization.writeTo(msgFactory, msg, writer, IGNORED));
buf.flip();
@@ -126,7 +127,7 @@ private T writeAndReadBack(T msg) throws IgniteCheckedExcept
T res = (T)msgFactory.create(makeMessageType(buf.get(), buf.get()));
- assertTrue(MessageSerialization.readFrom(msgFactory, res, reader));
+ assertTrue(MessageSerialization.readFrom(msgFactory, res, reader, IGNORED));
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 ce905b9806f70..1d61f44f57fa7 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 @@
/** Serializer for {@link TestDelayMessage} that injects an optional write delay for testing. */
public class TestDelayMessageSerializer implements MessageSerializer {
/** {@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 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/spi/discovery/tcp/TestTcpDiscoverySpi.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestTcpDiscoverySpi.java
index feedf6a4c6ed5..99eae878e2df1 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestTcpDiscoverySpi.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestTcpDiscoverySpi.java
@@ -125,7 +125,11 @@ public void discoveryHook(DiscoveryHook discoHook) {
};
try (dataSock) {
- return new TcpDiscoveryIoSession(ctx, dataSock).readMessage();
+ TcpDiscoveryIoSession ses = new TcpDiscoveryIoSession(ctx, dataSock);
+
+ ses.applyMessageSerializationContext(ctx.localNodeFeatures());
+
+ return ses.readMessage();
}
catch (Exception e) {
throw new IgniteException("Failed to decode a message", e);
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 cd76954a1a274..e90cc86847564 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 @@
/**
* 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 GridTestNode(UUID id, ClusterMetrics metrics) {
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 void order(long order) {
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 void setMetrics(ClusterMetrics metrics) {
this.metrics = metrics;
}
+ /** {@inheritDoc} */
+ @Override public Map cacheMetrics() {
+ return Collections.emptyMap();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void setCacheMetrics(Map 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 55e53c7d44cb1..4767e5fdc587d 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
@@ -1414,7 +1414,7 @@ protected final List movingKeysAfterJoin(Ignite ign, String cacheName,
if (nodeInitializer != null)
nodeInitializer.apply(fakeNode);
- fakeNode.consistentId(joiningNodeConsistentId == null ? getTestIgniteInstanceName(nodes.size()) :
+ fakeNode.setConsistentId(joiningNodeConsistentId == null ? getTestIgniteInstanceName(nodes.size()) :
joiningNodeConsistentId);
nodes.add(fakeNode);
@@ -1465,7 +1465,7 @@ protected List evictingPartitionsAfterJoin(Ignite ign, IgniteCache {
/** */
- @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 {
/** */
- @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(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 dba15be284691..9e707da48c770 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 bd79c20a48c26..d1814946151ce 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 fb0afa111fa6f..d5cc64a6abde1 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(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 {
/** */
- @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 4f3a7c0e51fc9..52f8785c633e4 100644
--- a/modules/core/src/test/resources/codegen/TestMarshalledArrayMapMessageSerializer.java
+++ b/modules/core/src/test/resources/codegen/TestMarshalledArrayMapMessageSerializer.java
@@ -19,6 +19,7 @@
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 43f7bf2f2e996..7b831f8ec04e8 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 dd0fcb2253364..e3aca2bf684f3 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 fa81057b60403..b924687a0d06f 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 @@
*/
public final class TestMarshalledMessageSerializer implements MessageSerializer {
/** */
- @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 b68578b06a615..68d62d8879123 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 7aa0a054471aa..1811c09f59037 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 {
+ /** */
+ @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/resources/codegen/TestUnknownFeatureMessage.java b/modules/core/src/test/resources/codegen/TestUnknownFeatureMessage.java
new file mode 100644
index 0000000000000..7e1db62bb29e7
--- /dev/null
+++ b/modules/core/src/test/resources/codegen/TestUnknownFeatureMessage.java
@@ -0,0 +1,27 @@
+/*
+ * 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.plugin.extensions.communication.Message;
+
+/** */
+public class TestUnknownFeatureMessage implements Message {
+ /** */
+ @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 f30fbc5cfc0ea..43f89a6ab11c8 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.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 GridDirectParser(IgniteLogger log, MessageFactory msgFactory, GridNioMess
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 a067277648d35..40badc6bcf4f6 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
@@ -85,6 +85,7 @@
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
@@ -1506,7 +1507,7 @@ private void processWriteSsl(SelectionKey key) throws IOException {
List pendingRequests = new ArrayList<>(2);
if (req != null)
- finished = writeToBuffer(writer, buf, req, pendingRequests);
+ finished = writeToBuffer(ses, writer, buf, req, pendingRequests);
// Fill up as many messages as possible to write buffer.
while (finished) {
@@ -1518,7 +1519,7 @@ private void processWriteSsl(SelectionKey key) throws IOException {
if (req == null)
break;
- finished = writeToBuffer(writer, buf, req, pendingRequests);
+ finished = writeToBuffer(ses, writer, buf, req, pendingRequests);
}
int sesBufLimit = buf.limit();
@@ -1588,6 +1589,7 @@ private void processWriteSsl(SelectionKey key) throws IOException {
}
/**
+ * @param ses Session the message is written to.
* @param writer Customizer of writing.
* @param buf Buffer to write.
* @param req Source of data.
@@ -1595,6 +1597,7 @@ private void processWriteSsl(SelectionKey key) throws IOException {
* @return {@code true} if message successfully written to buffer and {@code false} otherwise.
*/
private boolean writeToBuffer(
+ GridSelectorNioSessionImpl ses,
MessageWriter writer,
ByteBuffer buf,
SessionWriteRequest req,
@@ -1614,7 +1617,7 @@ private boolean writeToBuffer(
else {
writer.setBuffer(buf);
- finished = MessageSerialization.writeTo(messageFactory(), msg, writer);
+ finished = MessageSerialization.writeTo(messageFactory(), msg, writer, resolveSerializationContext(ses));
}
if (finished) {
@@ -1781,14 +1784,13 @@ private void processWrite0(SelectionKey key) throws IOException {
}
/**
- * @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.
* @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) {
Message msg;
boolean finished;
msg = (Message)req.message();
@@ -1803,7 +1805,7 @@ private boolean writeToBuffer(GridSelectorNioSessionImpl ses, ByteBuffer buf, Se
else {
writer.setBuffer(buf);
- finished = MessageSerialization.writeTo(msgFactory, msg, writer);
+ finished = MessageSerialization.writeTo(msgFactory, msg, writer, resolveSerializationContext(ses));
}
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 fbdcc6d5d34f9..9a5836d14cc4c 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 b436edb764d49..5f888f042ad3a 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,6 +17,7 @@
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;
@@ -40,11 +41,17 @@ private MessageSerialization() {
* @param factory Message factory.
* @param msg Message instance.
* @param writer Writer.
+ * @param ctx Serialization context.
* @param Message type.
* @return Whether message was fully written.
*/
- public static boolean writeTo(MessageFactory factory, M msg, MessageWriter writer) {
- return resolve(factory, msg).writeTo(msg, writer);
+ public static boolean writeTo(
+ MessageFactory factory,
+ M msg,
+ MessageWriter writer,
+ MessageSerializationContext ctx
+ ) {
+ return resolveMessageserializer(factory, msg).writeTo(msg, writer, ctx);
}
/**
@@ -53,16 +60,31 @@ public static boolean writeTo(MessageFactory factory, M msg,
* @param factory Message factory.
* @param msg Message instance.
* @param reader Reader.
+ * @param ctx Serialization context.
* @param Message type.
* @return Whether message was fully read.
*/
- public static boolean readFrom(MessageFactory factory, M msg, MessageReader reader) {
- return resolve(factory, msg).readFrom(msg, reader);
+ public static boolean readFrom(
+ MessageFactory factory,
+ M msg,
+ MessageReader reader,
+ MessageSerializationContext ctx
+ ) {
+ return resolveMessageserializer(factory, msg).readFrom(msg, reader, ctx);
+ }
+
+ /** */
+ public static MessageSerializationContext resolveSerializationContext(GridNioSession ses) {
+ MessageSerializationContext ctx = ses.meta(GridNioSessionMetaKey.MSG_SER_CTX.ordinal());
+
+ assert ctx != null : "Session has no serialization context: " + ses;
+
+ return ctx;
}
/** @return the serializer registered for {@code msg}'s direct type. */
@SuppressWarnings("unchecked")
- private static MessageSerializer resolve(MessageFactory factory, M msg) {
+ private static MessageSerializer resolveMessageserializer(MessageFactory factory, M msg) {
return (MessageSerializer)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 796acc40a191d..faf50669796dd 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.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 default void setBuffer(ByteBuffer buf) {
/**
* Reads nested message.
*
+ * @param ctx Serialization context.
* @param Type of the message.
* @return Message.
*/
- public default T readMessage() {
- return readMessage(false);
+ public default T readMessage(MessageSerializationContext ctx) {
+ return readMessage(false, ctx);
}
/**
* Reads nested message.
*
* @param compress Whether message should be decompressed.
+ * @param ctx Serialization context.
* @param Type of the message.
* @return Message.
*/
- public T readMessage(boolean compress);
+ public T readMessage(boolean compress, MessageSerializationContext ctx);
/**
* Reads {@link CacheObject}.
@@ -237,29 +240,32 @@ public default T readMessage() {
* Reads array of objects.
*
* @param type Array component type.
+ * @param ctx Serialization context.
* @param Type of the read object.
* @return Array of objects.
*/
- public T[] readObjectArray(MessageArrayType type);
+ public T[] readObjectArray(MessageArrayType type, MessageSerializationContext ctx);
/**
* Reads any collection.
*
* @param type Collection item type.
+ * @param ctx Serialization context.
* @param Type of the read collection.
* @return Collection.
*/
- public > C readCollection(MessageCollectionType type);
+ public > C readCollection(MessageCollectionType type, MessageSerializationContext ctx);
/**
* Reads map.
*
* @param type Map type.
+ * @param ctx Serialization context.
* @param Type of the read map.
* @return Map.
*/
- public default > M readMap(MessageMapType type) {
- return readMap(type, false);
+ public default > M readMap(MessageMapType type, MessageSerializationContext ctx) {
+ return readMap(type, false, ctx);
}
/**
@@ -267,10 +273,11 @@ public default T readMessage() {
*
* @param type Map type.
* @param compress Whether map should be compressed.
+ * @param ctx Serialization context.
* @param Type of the read map.
* @return Map.
*/
- public > M readMap(MessageMapType type, boolean compress);
+ public > 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 86ce0d170a09f..ab66f296ed6b0 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 {
*
* @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 019bab8fa9db9..4f281cec3a86a 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.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 default void setBuffer(ByteBuffer buf) {
* 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 default boolean writeMessage(Message val) {
*
* @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 default boolean writeMessage(Message val) {
*
* @param arr Array of objects.
* @param type Array component type.
+ * @param ctx Serialization context.
* @param Type of the objects that array contains.
* @return Whether array was fully written.
*/
- public boolean writeObjectArray(T[] arr, MessageArrayType type);
+ public 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 Type of the objects that collection contains.
* @return Whether value was fully written.
*/
- public boolean writeCollection(Collection col, MessageCollectionType type);
+ public boolean writeCollection(Collection col, MessageCollectionType type, MessageSerializationContext ctx);
/**
* Writes map.
*
* @param map Map.
* @param type Map type.
+ * @param ctx Serialization context.
* @param Initial key types of the map to write.
* @param Initial value types of the map to write.
* @return Whether value was fully written.
*/
- public default boolean writeMap(Map map, MessageMapType type) {
- return writeMap(map, type, false);
+ public default boolean writeMap(Map map, MessageMapType type, MessageSerializationContext ctx) {
+ return writeMap(map, type, false, ctx);
}
/**
@@ -324,11 +330,12 @@ public default boolean writeMap(Map map, MessageMapType type) {
* @param map Map.
* @param type Map type.
* @param compress Whether map should be compressed.
+ * @param ctx Serialization context.
* @param Initial key types of the map to write.
* @param Initial value types of the map to write.
* @return Whether value was fully written.
*/
- public boolean writeMap(Map map, MessageMapType type, boolean compress);
+ public boolean writeMap(Map 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 aeb7c1be8851b..451fe3d66efb9 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.MessageSerializer;
import org.apache.ignite.spi.IgniteSpiException;
+import static org.apache.ignite.internal.MessageSerializationContext.IGNORED;
+
/**
* Class is responsible for serializing discovery messages using RU-ready {@link MessageSerializer} mechanism.
*/
@@ -105,7 +107,7 @@ private void serializeMessage(Message m, OutputStream out) throws IOException {
do {
msgBuf.clear();
- finished = MessageSerialization.writeTo(msgFactory, m, msgWriter);
+ finished = MessageSerialization.writeTo(msgFactory, m, msgWriter, IGNORED);
out.write(msgBuf.array(), 0, msgBuf.position());
}
@@ -131,7 +133,7 @@ private T deserializeMessage(InputStream in) throws IOExcept
msgBuf.rewind();
}
- finished = MessageSerialization.readFrom(msgFactory, msg, msgReader);
+ finished = MessageSerialization.readFrom(msgFactory, msg, msgReader, IGNORED);
assert read != -1 || finished : "Stream closed before message was fully read.";