IGNITE-28520 Move prepareMarshal / finishUnmarshal out of NIO communication thread

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java

@@ -91,6 +91,7 @@
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteBiInClosure;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
 import org.apache.ignite.thread.IgniteThread;
 import org.jetbrains.annotations.Nullable;
 
@@ -1150,13 +1151,32 @@ private boolean onSend(GridCacheMessage msg, @Nullable UUID destNodeId) throws I
         if (destNodeId == null || !cctx.localNodeId().equals(destNodeId)) {
             msg.prepareMarshal(cctx);
 
+            prepareMarshalGeneratedCacheObjects(msg);
+
             if (msg instanceof GridCacheDeployable && msg.addDeploymentInfo())
                 cctx.deploy().prepare((GridCacheDeployable)msg);
         }
 
         return true;
     }
 
+    /** */
+    @SuppressWarnings({"unchecked", "rawtypes"})
+    private void prepareMarshalGeneratedCacheObjects(GridCacheMessage msg) throws IgniteCheckedException {
+        if (!(msg instanceof GridCacheIdMessage))
+            return;
+
+        CacheObjectContext cacheObjCtx = cctx.cacheObjectContext(((GridCacheIdMessage)msg).cacheId());
+
+        if (cacheObjCtx == null)
+            return;
+
+        MessageSerializer ser = cctx.gridIO().messageFactory().serializer(msg.directType());
+
+        if (ser != null)
+            ser.prepareMarshalCacheObjects(msg, cacheObjCtx, cctx);
+    }
+
     /**
      * @param nodeId Node ID.
      * @param sndErr Send error.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTtlUpdateRequest.java

@@ -155,17 +155,6 @@ public List<GridCacheVersion> nearVersions() {
         return nearVers;
     }
 
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
-        super.prepareMarshal(ctx);
-
-        GridCacheContext cctx = ctx.cacheContext(cacheId);
-
-        prepareMarshalCacheObjects(keys, cctx);
-
-        prepareMarshalCacheObjects(nearKeys, cctx);
-    }
-
     /** {@inheritDoc} */
     @Override public void finishUnmarshal(GridCacheSharedContext ctx, ClassLoader ldr)
         throws IgniteCheckedException {

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java

@@ -346,16 +346,6 @@ public long timeout() {
         return ctx.txLockMessageLogger();
     }
 
-    /** {@inheritDoc}
-     * @param ctx*/
-    @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
-        super.prepareMarshal(ctx);
-
-        GridCacheContext<?, ?> cctx = ctx.cacheContext(cacheId);
-
-        prepareMarshalCacheObjects(keys, cctx);
-    }
-
     /** {@inheritDoc} */
     @Override public void finishUnmarshal(GridCacheSharedContext<?, ?> ctx, ClassLoader ldr) throws IgniteCheckedException {
         super.finishUnmarshal(ctx, ldr);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java

@@ -159,14 +159,6 @@ protected int valuesSize() {
         return ctx.txLockMessageLogger();
     }
 
-    /** {@inheritDoc}
-     * @param ctx*/
-    @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
-        super.prepareMarshal(ctx);
-
-        prepareMarshalCacheObjects(vals, ctx.cacheContext(cacheId));
-    }
-
     /** {@inheritDoc} */
     @Override public void finishUnmarshal(GridCacheSharedContext<?, ?> ctx, ClassLoader ldr) throws IgniteCheckedException {
         super.finishUnmarshal(ctx, ldr);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridNearUnlockRequest.java

@@ -75,14 +75,6 @@ public void addKey(KeyCacheObject key) {
         return keys != null && !keys.isEmpty() ? keys.get(0).partition() : -1;
     }
 
-    /** {@inheritDoc}
-     * @param ctx*/
-    @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
-        super.prepareMarshal(ctx);
-
-        prepareMarshalCacheObjects(keys, ctx.cacheContext(cacheId));
-    }
-
     /** {@inheritDoc} */
     @Override public void finishUnmarshal(GridCacheSharedContext<?, ?> ctx, ClassLoader ldr) throws IgniteCheckedException {
         super.finishUnmarshal(ctx, ldr);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtUnlockRequest.java

@@ -70,13 +70,6 @@ public void addNearKey(KeyCacheObject key)
         nearKeys.add(key);
     }
 
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
-        super.prepareMarshal(ctx);
-
-        prepareMarshalCacheObjects(nearKeys, ctx.cacheContext(cacheId));
-    }
-
     /** {@inheritDoc} */
     @Override public void finishUnmarshal(GridCacheSharedContext<?, ?> ctx, ClassLoader ldr) throws IgniteCheckedException {
         super.finishUnmarshal(ctx, ldr);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicNearResponse.java

@@ -171,14 +171,6 @@ public long futureId() {
         return false;
     }
 
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
-        super.prepareMarshal(ctx);
-
-        if (errs != null)
-            errs.prepareMarshal(this, ctx.cacheContext(cacheId));
-    }
-
     /** {@inheritDoc} */
     @Override public void finishUnmarshal(GridCacheSharedContext<?, ?> ctx, ClassLoader ldr) throws IgniteCheckedException {
         super.finishUnmarshal(ctx, ldr);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java

@@ -467,16 +467,6 @@ else if (conflictVers != null)
 
         GridCacheContext<?, ?> cctx = ctx.cacheContext(cacheId);
 
-        prepareMarshalCacheObjects(keys, cctx);
-
-        prepareMarshalCacheObjects(vals, cctx);
-
-        prepareMarshalCacheObjects(nearKeys, cctx);
-
-        prepareMarshalCacheObjects(nearVals, cctx);
-
-        prepareMarshalCacheObjects(prevVals, cctx);
-
         if (forceTransformBackups) {
             // force addition of deployment info for entry processors if P2P is enabled globally.
             if (!addDepInfo && ctx.deploymentEnabled())

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java

@@ -118,21 +118,6 @@ public void nearEvicted(List<KeyCacheObject> nearEvicted) {
         return partId;
     }
 
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
-        super.prepareMarshal(ctx);
-
-        GridCacheContext cctx = ctx.cacheContext(cacheId);
-
-        // Can be null if client near cache was removed, in this case assume do not need prepareMarshal.
-        if (cctx != null) {
-            prepareMarshalCacheObjects(nearEvicted, cctx);
-
-            if (errs != null)
-                errs.prepareMarshal(this, cctx);
-        }
-    }
-
     /** {@inheritDoc} */
     @Override public void finishUnmarshal(GridCacheSharedContext ctx, ClassLoader ldr) throws IgniteCheckedException {
         super.finishUnmarshal(ctx, ldr);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicFullUpdateRequest.java

@@ -336,8 +336,6 @@ else if (conflictVers != null)
         if (expiryPlc != null && expiryPlcBytes == null)
             expiryPlcBytes = CU.marshal(cctx, new IgniteExternalizableExpiryPolicy(expiryPlc));
 
-        prepareMarshalCacheObjects(keys, cctx);
-
         if (filter != null) {
             boolean hasFilter = false;
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateRequest.java

@@ -216,18 +216,6 @@ public GridNearAtomicSingleUpdateRequest() {
         return CU.EXPIRE_TIME_CALCULATE;
     }
 
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
-        super.prepareMarshal(ctx);
-
-        GridCacheContext cctx = ctx.cacheContext(cacheId);
-
-        prepareMarshalCacheObject(key, cctx);
-
-        if (val != null)
-            prepareMarshalCacheObject(val, cctx);
-    }
-
     /** {@inheritDoc} */
     @Override public void finishUnmarshal(GridCacheSharedContext ctx, ClassLoader ldr) throws IgniteCheckedException {
         super.finishUnmarshal(ctx, ldr);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java

@@ -344,14 +344,8 @@ synchronized void addFailedKeys(Collection<KeyCacheObject> keys, Throwable e) {
 
         GridCacheContext cctx = ctx.cacheContext(cacheId);
 
-        if (errs != null)
-            errs.prepareMarshal(this, cctx);
-
         if (nearUpdates != null)
             prepareMarshalCacheObjects(nearUpdates.nearValues(), cctx);
-
-        if (ret != null)
-            ret.prepareMarshal(cctx);
     }
 
     /** {@inheritDoc} */

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/UpdateErrors.java

@@ -115,11 +115,6 @@ void addFailedKeys(Collection<KeyCacheObject> keys, Throwable e) {
         errMsg.error().addSuppressed(e);
     }
 
-    /** */
-    void prepareMarshal(GridCacheMessage msg, GridCacheContext<?, ?> cctx) throws IgniteCheckedException {
-        msg.prepareMarshalCacheObjects(failedKeys, cctx);
-    }
-
     /** */
     void finishUnmarshal(GridCacheMessage msg, GridCacheContext<?, ?> cctx, ClassLoader ldr) throws IgniteCheckedException {
         msg.finishUnmarshalCacheObjects(failedKeys, cctx, ldr);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysRequest.java

@@ -113,15 +113,6 @@ public Collection<KeyCacheObject> keys() {
         return topVer;
     }
 
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
-        super.prepareMarshal(ctx);
-
-        GridCacheContext cctx = ctx.cacheContext(cacheId);
-
-        prepareMarshalCacheObjects(keys, cctx);
-    }
-
     /** {@inheritDoc} */
     @Override public void finishUnmarshal(GridCacheSharedContext ctx, ClassLoader ldr) throws IgniteCheckedException {
         super.finishUnmarshal(ctx, ldr);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysResponse.java

@@ -137,9 +137,6 @@ public void addInfo(GridCacheEntryInfo info) {
 
         GridCacheContext<?, ?> cctx = ctx.cacheContext(cacheId);
 
-        if (missedKeys != null)
-            prepareMarshalCacheObjects(missedKeys, cctx);
-
         if (infos != null) {
             for (GridCacheEntryInfo info : infos)
                 info.marshal(cctx.cacheObjectContext());

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java

@@ -280,22 +280,6 @@ public long accessTtl() {
         return txLbl;
     }
 
-    /**
-     * @param ctx Cache context.
-     * @throws IgniteCheckedException If failed.
-     */
-    @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
-        super.prepareMarshal(ctx);
-
-        assert ctx != null;
-        assert !F.isEmpty(keys);
-        assert readersFlags == null || keys.size() == readersFlags.size();
-
-        GridCacheContext<?, ?> cctx = ctx.cacheContext(cacheId);
-
-        prepareMarshalCacheObjects(keys, cctx);
-    }
-
     /**
      * @param ctx Context.
      * @param ldr Loader.

modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageSerializer.java

@@ -17,9 +17,11 @@
 
 package org.apache.ignite.plugin.extensions.communication;
 
-/**
- * Interface for message serialization logic.
- */
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.cache.CacheObjectValueContext;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+
+/** Message serialization logic. */
 public interface MessageSerializer<M extends Message> {
     /**
      * Writes this message to provided byte buffer.
@@ -38,4 +40,20 @@ public interface MessageSerializer<M extends Message> {
      * @return Whether message was fully read.
      */
     public boolean readFrom(M msg, MessageReader reader);
+
+    /**
+     * Runs {@code CacheObject.prepareMarshal} for {@code @Order} cache-object fields on the user thread, so the NIO
+     * worker never does it. Default is a no-op. The caller is responsible for guaranteeing that {@code ctx} is
+     * non-null when invoking this method; resolution-with-null-skip happens at call sites.
+     *
+     * @param msg Message instance.
+     * @param ctx Cache object value context for {@code msg}'s direct {@code CacheObject} fields and non-cacheId-aware
+     *     nested messages. Always non-null.
+     * @param sharedCtx Shared cache context for resolving per-cache contexts of nested cacheId-aware messages.
+     * @throws IgniteCheckedException If marshalling fails.
+     */
+    public default void prepareMarshalCacheObjects(M msg, CacheObjectValueContext ctx, GridCacheSharedContext sharedCtx)
+        throws IgniteCheckedException {
+        // No-op by default.
+    }
 }

modules/core/src/test/java/org/apache/ignite/internal/codegen/MessageProcessorTest.java

@@ -340,6 +340,36 @@ public void testCompressedMessageExplicitUsageFails() {
         assertThat(compilation).hadErrorContaining(errMsg);
     }
 
+    /** Collection-of-entries encoding of a {@code Map<KeyCacheObject, ?>}: generator recurses into each entry's KCO. */
+    @Test
+    public void testKeyCacheObjectInCollectionOfEntries() {
+        Compilation compilation = compile("KeyCacheObjectEntryMsg.java", "TestKeyCacheObjectCollectionMessage.java");
+
+        assertThat(compilation).succeeded();
+
+        assertEquals(2, compilation.generatedSourceFiles().size());
+
+        assertThat(compilation)
+            .generatedSourceFile("org.apache.ignite.internal.KeyCacheObjectEntryMsgSerializer")
+            .hasSourceEquivalentTo(javaFile("KeyCacheObjectEntryMsgSerializer.java"));
+
+        assertThat(compilation)
+            .generatedSourceFile("org.apache.ignite.internal.TestKeyCacheObjectCollectionMessageSerializer")
+            .hasSourceEquivalentTo(javaFile("TestKeyCacheObjectCollectionMessageSerializer.java"));
+    }
+
+/** {@code @Order Map<KeyCacheObject, GridCacheVersion>}: generator walks keys/values via {@code keySet()/values()}. */
+    @Test
+    public void testMapWithKeyCacheObjectAndMessageValue() {
+        Compilation compilation = compile("TestMapKeyCacheObjectMessage.java");
+
+        assertThat(compilation).succeeded();
+
+        assertThat(compilation)
+            .generatedSourceFile("org.apache.ignite.internal.TestMapKeyCacheObjectMessageSerializer")
+            .hasSourceEquivalentTo(javaFile("TestMapKeyCacheObjectMessageSerializer.java"));
+    }
+
     /**
      * Negative test that verifies the compilation failed if the Compress annotation is used for unsupported types.
      */

modules/core/src/test/resources/codegen/KeyCacheObjectEntryMsg.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal;
+
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.plugin.extensions.communication.Message;
+
+/** APT fixture: entry Message for {@link TestKeyCacheObjectCollectionMessage}. */
+public class KeyCacheObjectEntryMsg implements Message {
+    @Order(0)
+    KeyCacheObject key;
+
+    @Order(1)
+    GridCacheVersion val;
+
+    public short directType() {
+        return 0;
+    }
+}