reimpl

Vladsz83 · Vladsz83 · commit c9390c50a430 · 2026-04-09T15:07:54.000+03:00
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java
@@ -19,6 +19,7 @@
 
 import java.lang.reflect.Constructor;
 import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.binary.BinaryMarshaller;
 import org.apache.ignite.internal.cache.query.index.IndexQueryResultMeta;
 import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyDefinition;
 import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyTypeSettings;
@@ -227,8 +228,10 @@
 import org.apache.ignite.internal.util.distributed.FullMessage;
 import org.apache.ignite.internal.util.distributed.InitMessage;
 import org.apache.ignite.internal.util.distributed.SingleNodeMessage;
+import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteProductVersion;
 import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.marshaller.jdk.JdkMarshaller;
 import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
 import org.apache.ignite.plugin.extensions.communication.Message;
 import org.apache.ignite.plugin.extensions.communication.MessageFactory;
@@ -285,11 +288,14 @@ public class CoreMessagesProvider implements MessageFactoryProvider {
     /** Handshake wait message type. */
     public static final short HANDSHAKE_WAIT_MSG_TYPE = HANDSHAKE_MSG_TYPE + 1;
 
-    /** Custom data marshaller. */
-    private final Marshaller marsh;
+    /** Binary marshaller. */
+    private final Marshaller schemaAwareMarhaller;
 
-    /** Class loader for the custom data marshalling. */
-    private final ClassLoader clsLdr;
+    /** Binary marshaller. */
+    private final Marshaller schemaLessMarshaller;
+
+    /** Resolved classloader. */
+    private final ClassLoader resolvedClsLdr;
 
     /** */
     private short msgIdx;
@@ -298,12 +304,14 @@ public class CoreMessagesProvider implements MessageFactoryProvider {
     private @Nullable MessageFactory factory;
 
     /**
-     * @param marsh Custom data marshaller.
-     * @param clsLdr Class loader for the custom data marshalling.
+     * @param schemaAwareMarhaller Schema-aware marshaller like {@link BinaryMarshaller}.
+     * @param schemaLessMarshaller Pure, schemaless marshaller like {@link JdkMarshaller}.
+     * @param resolvedClsLdr Resolved classloader.
      */
-    public CoreMessagesProvider(Marshaller marsh, ClassLoader clsLdr) {
-        this.marsh = marsh;
-        this.clsLdr = clsLdr;
+    public CoreMessagesProvider(Marshaller schemaAwareMarhaller, Marshaller schemaLessMarshaller, ClassLoader resolvedClsLdr) {
+        this.schemaAwareMarhaller = schemaAwareMarhaller;
+        this.schemaLessMarshaller = schemaLessMarshaller;
+        this.resolvedClsLdr = resolvedClsLdr;
     }
 
     /** {@inheritDoc} */
@@ -632,8 +640,13 @@ public CoreMessagesProvider(Marshaller marsh, ClassLoader clsLdr) {
         register(WalStateAckMessage.class);
     }
 
-    /** Registers message incrementing {@link #msgIdx}. */
+    /** */
     private <T extends Message> void register(Class<T> cls) {
+        register(cls, schemaLessMarshaller, U.gridClassLoader());
+    }
+
+    /** Registers message using ane incrementing {@link #msgIdx} as the message id/type. */
+    private <T extends Message> void register(Class<T> cls, Marshaller marsh, ClassLoader clsLdr) {
         Constructor<T> ctor;
         MessageSerializer<T> serializer;
 
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -1320,7 +1320,8 @@ private void initMessageFactory() throws IgniteCheckedException {
 
         List<MessageFactoryProvider> compMsgs = new ArrayList<>();
 
-        compMsgs.add(new CoreMessagesProvider(ctx.marshaller(), U.resolveClassLoader(ctx.config())));
+        compMsgs.add(new CoreMessagesProvider(ctx.marshaller(), ctx.marshallerContext().jdkMarshaller(),
+            U.resolveClassLoader(ctx.config())));
 
         for (IgniteComponentType compType : IgniteComponentType.values()) {
             MessageFactoryProvider f = compType.messageFactory();
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
@@ -46,7 +46,7 @@ public class DirectMarshallingMessagesTest extends GridCommonAbstractTest {
     /** Message factory. */
     private final MessageFactory msgFactory =
         new IgniteMessageFactoryImpl(new MessageFactoryProvider[] {
-            new CoreMessagesProvider(jdk(), U.gridClassLoader()),
+            new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader()),
             factory -> factory.register(
                 TestNestedContainersMessage.TYPE,
                 TestNestedContainersMessage::new,
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
@@ -47,7 +47,7 @@ public class CompressedMessageTest {
     @Test
     public void testWriteReadHugeMessage() {
         MessageFactory msgFactory = new IgniteMessageFactoryImpl(new MessageFactoryProvider[]{
-            new CoreMessagesProvider(jdk(), U.gridClassLoader())});
+            new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader())});
 
         DirectMessageWriter writer = new DirectMessageWriter(msgFactory);
 
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/IgniteIoCommunicationMessageSerializationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/IgniteIoCommunicationMessageSerializationTest.java
@@ -34,7 +34,7 @@
 public class IgniteIoCommunicationMessageSerializationTest extends AbstractMessageSerializationTest {
     /** {@inheritDoc} */
     @Override protected MessageFactoryProvider messageFactory() {
-        return new CoreMessagesProvider(jdk(), U.gridClassLoader());
+        return new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader());
     }
 
     /** {@inheritDoc} */
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/IgniteDiscoveryMessageSerializationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/IgniteDiscoveryMessageSerializationTest.java
@@ -28,6 +28,6 @@
 public class IgniteDiscoveryMessageSerializationTest extends AbstractMessageSerializationTest {
     /** {@inheritDoc} */
     @Override protected MessageFactoryProvider messageFactory() {
-        return new CoreMessagesProvider(jdk(), U.gridClassLoader());
+        return new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader());
     }
 }
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
@@ -150,7 +150,7 @@ public void testCacheContinuousQueryEntrySerialization() {
         e0.markFiltered();
 
         IgniteMessageFactoryImpl msgFactory =
-            new IgniteMessageFactoryImpl(new MessageFactoryProvider[]{new CoreMessagesProvider(jdk(), U.gridClassLoader())});
+            new IgniteMessageFactoryImpl(new MessageFactoryProvider[]{new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader())});
 
         ByteBuffer buf = ByteBuffer.allocate(4096);
         DirectMessageWriter writer = new DirectMessageWriter(msgFactory);
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java
@@ -154,7 +154,7 @@ private void startSpis() throws Exception {
             GridSpiTestContext ctx = initSpiContext();
 
             ctx.messageFactory(new IgniteMessageFactoryImpl(new MessageFactoryProvider[] {
-                new CoreMessagesProvider(jdk(), U.gridClassLoader()), customMessageFactory()}));
+                new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader()), customMessageFactory()}));
 
             ctx.setLocalNode(node);
 
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java
@@ -434,7 +434,7 @@ private void startSpis(MessageListener lsnr) throws Exception {
             GridSpiTestContext ctx = initSpiContext();
 
             ctx.messageFactory(new IgniteMessageFactoryImpl(
-                new MessageFactoryProvider[] {new CoreMessagesProvider(jdk(), U.gridClassLoader()), GRID_TEST_MESSAGE_FACTORY})
+                new MessageFactoryProvider[] {new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader()), GRID_TEST_MESSAGE_FACTORY})
             );
 
             ctx.setLocalNode(node);
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConfigSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConfigSelfTest.java
@@ -252,7 +252,7 @@ private TcpCommunicationSpi initializeSpi(GridSpiTestContext ctx,
         node.setId(rsrcs.getNodeId());
 
         ctx.messageFactory(new IgniteMessageFactoryImpl(new MessageFactoryProvider[]{
-            new CoreMessagesProvider(jdk(), U.gridClassLoader()), GRID_TEST_MESSAGE_FACTORY}));
+            new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader()), GRID_TEST_MESSAGE_FACTORY}));
 
         ctx.setLocalNode(node);
 
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java
@@ -468,7 +468,7 @@ private int getSpiCount() {
             GridSpiTestContext ctx = initSpiContext();
 
             ctx.messageFactory(new IgniteMessageFactoryImpl(
-                    new MessageFactoryProvider[] {new CoreMessagesProvider(jdk(), U.gridClassLoader()), GRID_TEST_MESSAGE_FACTORY})
+                    new MessageFactoryProvider[] {new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader()), GRID_TEST_MESSAGE_FACTORY})
             );
 
             ctx.timeoutProcessor(timeoutProcessor);
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryAckSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryAckSelfTest.java
@@ -400,7 +400,7 @@ private void startSpis(int ackCnt, int idleTimeout, int queueLimit) throws Excep
             GridSpiTestContext ctx = initSpiContext();
 
             ctx.messageFactory(new IgniteMessageFactoryImpl(
-                    new MessageFactoryProvider[] {new CoreMessagesProvider(jdk(), U.gridClassLoader()), GRID_TEST_MESSAGE_FACTORY})
+                    new MessageFactoryProvider[] {new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader()), GRID_TEST_MESSAGE_FACTORY})
             );
 
             ctx.setLocalNode(node);
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoverySelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoverySelfTest.java
@@ -729,7 +729,7 @@ private void startSpis() throws Exception {
             GridSpiTestContext ctx = initSpiContext();
 
             ctx.messageFactory(new IgniteMessageFactoryImpl(
-                new MessageFactoryProvider[] {new CoreMessagesProvider(jdk(), U.gridClassLoader()), GRID_TEST_MESSAGE_FACTORY})
+                new MessageFactoryProvider[] {new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader()), GRID_TEST_MESSAGE_FACTORY})
             );
 
             ctx.setLocalNode(node);
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java
@@ -453,7 +453,7 @@ private void startSpis(int ackCnt, int idleTimeout, int queueLimit) throws Excep
             GridSpiTestContext ctx = initSpiContext();
 
             ctx.messageFactory(new IgniteMessageFactoryImpl(
-                new MessageFactoryProvider[] {new CoreMessagesProvider(jdk(), U.gridClassLoader()), GRID_TEST_MESSAGE_FACTORY})
+                new MessageFactoryProvider[] {new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader()), GRID_TEST_MESSAGE_FACTORY})
             );
 
             ctx.setLocalNode(node);
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
@@ -121,7 +121,7 @@ public void messageFactory(MessageFactoryProvider msgFactoryProvider) {
         assert !started();
 
         this.msgFactory = new IgniteMessageFactoryImpl(new MessageFactoryProvider[] {
-            new CoreMessagesProvider(jdk(), U.resolveClassLoader(ignite().configuration())),
+            new CoreMessagesProvider(jdk(), jdk(), U.resolveClassLoader(ignite().configuration())),
             msgFactoryProvider
         });
     }
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/GridSpiTestContext.java b/modules/core/src/test/java/org/apache/ignite/testframework/GridSpiTestContext.java
@@ -554,7 +554,8 @@ public void triggerEvent(Event evt) {
     /** {@inheritDoc} */
     @Override public MessageFactory messageFactory() {
         if (factory == null)
-            factory = new IgniteMessageFactoryImpl(new MessageFactoryProvider[]{new CoreMessagesProvider(jdk(), U.gridClassLoader())});
+            factory = new IgniteMessageFactoryImpl(new MessageFactoryProvider[]{new CoreMessagesProvider(jdk(), jdk(),
+                U.gridClassLoader())});
 
         return factory;
     }
diff --git a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpi.java b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/ZookeeperDiscoverySpi.java
@@ -460,7 +460,9 @@ public DiscoverySpiNodeAuthenticator getAuthenticator() {
             lsnr,
             exchange,
             stats,
-            ((IgniteEx)ignite).context().marshallerContext().jdkMarshaller());
+            ((IgniteEx)ignite).context().marshallerContext().jdkMarshaller(),
+            ((IgniteEx)ignite).context().messageFactory()
+        );
 
         registerMBean(igniteInstanceName, new ZookeeperDiscoverySpiMBeanImpl(this), ZookeeperDiscoverySpiMBean.class);
 
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
@@ -26,16 +26,12 @@
 import java.nio.ByteBuffer;
 import java.util.zip.DeflaterOutputStream;
 import java.util.zip.InflaterInputStream;
-import org.apache.ignite.internal.CoreMessagesProvider;
 import org.apache.ignite.internal.direct.DirectMessageReader;
 import org.apache.ignite.internal.direct.DirectMessageWriter;
-import org.apache.ignite.internal.managers.communication.IgniteMessageFactoryImpl;
 import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.marshaller.Marshaller;
 import org.apache.ignite.marshaller.jdk.JdkMarshaller;
 import org.apache.ignite.plugin.extensions.communication.Message;
 import org.apache.ignite.plugin.extensions.communication.MessageFactory;
-import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
 import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
 import org.apache.ignite.spi.IgniteSpiException;
 import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
@@ -57,15 +53,15 @@ public class DiscoveryMessageParser {
     private static final int MSG_BUFFER_SIZE = 100;
 
     /** */
-    private final MessageFactory msgFactory;
+    private final JdkMarshaller jdkMarshaller;
 
     /** */
-    private final Marshaller marsh;
+    private final MessageFactory msgFactory;
 
     /** */
-    public DiscoveryMessageParser(Marshaller marsh) {
-        this.marsh = marsh;
-        msgFactory = new IgniteMessageFactoryImpl(new MessageFactoryProvider[] { new CoreMessagesProvider(marsh, U.gridClassLoader()) });
+    public DiscoveryMessageParser(JdkMarshaller jdkMarshaller, MessageFactory msgFactory) {
+        this.jdkMarshaller = jdkMarshaller;
+        this.msgFactory = msgFactory;
     }
 
     /** Marshals discovery message to bytes array. */
@@ -81,7 +77,7 @@ public byte[] marshalZip(DiscoverySpiCustomMessage msg) {
             else {
                 out.write(JAVA_SERIALIZATION);
 
-                U.marshal(marsh, msg, out);
+                U.marshal(jdkMarshaller, msg, out);
             }
         }
         catch (Exception e) {
@@ -100,7 +96,7 @@ public DiscoverySpiCustomMessage unmarshalZip(byte[] bytes) {
             byte mode = (byte)in.read();
 
             if (mode == JAVA_SERIALIZATION)
-                return U.unmarshal(marsh, in, U.gridClassLoader());
+                return U.unmarshal(jdkMarshaller, in, U.gridClassLoader());
 
             if (MESSAGE_SERIALIZATION != mode)
                 throw new IOException("Received unexpected byte while reading discovery message: " + mode);
diff --git a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
@@ -77,6 +77,7 @@
 import org.apache.ignite.lang.IgniteRunnable;
 import org.apache.ignite.lang.IgniteUuid;
 import org.apache.ignite.marshaller.jdk.JdkMarshaller;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.apache.ignite.plugin.security.SecurityCredentials;
 import org.apache.ignite.spi.IgniteNodeValidationResult;
 import org.apache.ignite.spi.IgniteSpiException;
@@ -228,7 +229,8 @@ public class ZookeeperDiscoveryImpl {
      * @param lsnr Discovery events listener.
      * @param exchange Discovery data exchange.
      * @param stats Zookeeper DiscoverySpi statistics collector.
-     * @param marsh Marshaller.
+     * @param jdkMarshaller JDK Marshaller.
+     * @param msgFactory Message Factory.
      */
     public ZookeeperDiscoveryImpl(
         ZookeeperDiscoverySpi spi,
@@ -239,7 +241,8 @@ public ZookeeperDiscoveryImpl(
         DiscoverySpiListener lsnr,
         DiscoverySpiDataExchange exchange,
         ZookeeperDiscoveryStatistics stats,
-        JdkMarshaller marsh
+        JdkMarshaller jdkMarshaller,
+        MessageFactory msgFactory
     ) {
         assert locNode.id() != null && locNode.isLocal() : locNode;
 
@@ -254,7 +257,7 @@ public ZookeeperDiscoveryImpl(
         this.lsnr = lsnr;
         this.exchange = exchange;
         this.clientReconnectEnabled = locNode.isClient() && !spi.isClientReconnectDisabled();
-        this.marsh = marsh;
+        this.marsh = jdkMarshaller;
 
         int evtsAckThreshold = IgniteSystemProperties.getInteger(IGNITE_ZOOKEEPER_DISCOVERY_SPI_ACK_THRESHOLD,
             DFLT_ZOOKEEPER_DISCOVERY_SPI_ACK_THRESHOLD);
@@ -266,7 +269,7 @@ public ZookeeperDiscoveryImpl(
 
         this.stats = stats;
 
-        msgParser = new DiscoveryMessageParser(marsh);
+        msgParser = new DiscoveryMessageParser(jdkMarshaller, msgFactory);
     }
 
     /**

Original file line number	Diff line number	Diff line change
`@@ -34,7 +34,7 @@`
`34`	`34`	`public class IgniteIoCommunicationMessageSerializationTest extends AbstractMessageSerializationTest {`
`35`	`35`	`/** {@inheritDoc} */`
`36`	`36`	`@Override protected MessageFactoryProvider messageFactory() {`
`37`		`- return new CoreMessagesProvider(jdk(), U.gridClassLoader());`
	`37`	`+ return new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader());`
`38`	`38`	`}`
`39`	`39`
`40`	`40`	`/** {@inheritDoc} */`
Original file line number	Diff line number	Diff line change
`@@ -28,6 +28,6 @@`
`28`	`28`	`public class IgniteDiscoveryMessageSerializationTest extends AbstractMessageSerializationTest {`
`29`	`29`	`/** {@inheritDoc} */`
`30`	`30`	`@Override protected MessageFactoryProvider messageFactory() {`
`31`		`- return new CoreMessagesProvider(jdk(), U.gridClassLoader());`
	`31`	`+ return new CoreMessagesProvider(jdk(), jdk(), U.gridClassLoader());`
`32`	`32`	`}`
`33`	`33`	`}`