IGNITE-28528 Revise marshalling of the node filter in StartRequestData

modules/codegen/src/main/java/org/apache/ignite/internal/MessageMarshallerGenerator.java

@@ -46,6 +46,7 @@
 import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.internal.MessageProcessor.CACHE_OBJECT_CLS;
+import static org.apache.ignite.internal.MessageProcessor.DEPLOYMENT_AWARE_MESSAGE_INTERFACE;
 import static org.apache.ignite.internal.MessageProcessor.IGNITE_CHECKED_EXCEPTION_CLS;
 import static org.apache.ignite.internal.MessageProcessor.KEY_CACHE_OBJECT_CLS;
 import static org.apache.ignite.internal.MessageProcessor.MARSHALLABLE_MESSAGE_INTERFACE;
@@ -96,6 +97,9 @@ public class MessageMarshallerGenerator extends MessageCompanionGenerator {
     /** */
     private final TypeMirror nonMarshallableType;
 
+    /** */
+    private final TypeMirror deploymentAwareMsgType;
+
     /** */
     private final TypeMirror selfMarshallingMsgType;
 
@@ -140,6 +144,7 @@ public class MessageMarshallerGenerator extends MessageCompanionGenerator {
         msgType = type(MESSAGE_INTERFACE);
         cacheObjType = type(CACHE_OBJECT_CLS);
         nonMarshallableType = type(NON_MARSHALLABLE_MESSAGE_INTERFACE);
+        deploymentAwareMsgType = type(DEPLOYMENT_AWARE_MESSAGE_INTERFACE);
         selfMarshallingMsgType = type(SELF_MARSHALLING_MESSAGE_INTERFACE);
         cacheGrpIdMsgType = type(GRID_CACHE_GROUP_ID_MESSAGE_CLS);
         mapType = type(Map.class.getName());
@@ -306,6 +311,9 @@ private void generateUnmarshalMethod(String params, List<VariableElement> fields
             if (needsCtx(fields) || !wireFieldSkip.isEmpty())
                 appendBlock(body, List.of(ctxResolutionLine()));
 
+            if (isDeploymentAware())
+                appendBlock(body, List.of(deploymentResolutionLine()));
+
             appendFields(body, fields, MarshalMode.UNMARSHAL, wireFieldSkip);
 
             if (marshallable)
@@ -1039,6 +1047,20 @@ else if (isCacheGroupIdMessage(type))
             return indentedLine("CacheObjectContext ctx = cacheObjCtx;");
     }
 
+    /**
+     * Returns the line resolving the class loader of a {@code DeploymentAware} message. The deployment the message
+     * carries wins over whatever the caller passed: the caller cannot know the loader of classes deployed elsewhere,
+     * and the overloads defaulting to the local one would silently read them without peer class loading.
+     */
+    private String deploymentResolutionLine() {
+        return indentedLine("clsLdr = kctx.deploy().classLoader(msg);");
+    }
+
+    /** @return {@code true} if the message carries the deployment of its classes. */
+    private boolean isDeploymentAware() {
+        return deploymentAwareMsgType != null && assignableFrom(type.asType(), deploymentAwareMsgType);
+    }
+
     /** Returns {@code true} if any field requires {@code ctx} in generated marshal/unmarshal code. */
     private boolean needsCtx(List<VariableElement> fields) {
         return fields.stream().anyMatch(f -> needsCtxType(f.asType()));

modules/codegen/src/main/java/org/apache/ignite/internal/MessageProcessor.java

@@ -87,6 +87,9 @@ public class MessageProcessor extends AbstractProcessor {
     /** Message that reshapes its own fields before they go on the wire. */
     static final String SELF_MARSHALLING_MESSAGE_INTERFACE = "org.apache.ignite.internal.SelfMarshallingMessage";
 
+    /** Message that carries the deployment of the classes inside it. */
+    static final String DEPLOYMENT_AWARE_MESSAGE_INTERFACE = "org.apache.ignite.internal.DeploymentAware";
+
     /** Marker of messages with no marshaller. */
     static final String NON_MARSHALLABLE_MESSAGE_INTERFACE = "org.apache.ignite.plugin.extensions.communication.NonMarshallableMessage";
 

modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java

@@ -30,7 +30,7 @@
 import org.apache.ignite.internal.managers.communication.IgniteIoTestMessage;
 import org.apache.ignite.internal.managers.communication.IgniteMessageFactory;
 import org.apache.ignite.internal.managers.communication.SessionChannelMessage;
-import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
+import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoMessage;
 import org.apache.ignite.internal.managers.deployment.GridDeploymentRequest;
 import org.apache.ignite.internal.managers.deployment.GridDeploymentResponse;
 import org.apache.ignite.internal.managers.encryption.ChangeCacheEncryptionRequest;
@@ -686,7 +686,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh) {
 
         // [12200 - 12300]: Binary, classloading and marshalling messages.
         msgIdx = 12200;
-        register(GridDeploymentInfoBean.class);
+        register(GridDeploymentInfoMessage.class);
         register(GridDeploymentRequest.class);
         register(GridDeploymentResponse.class);
         register(MissingMappingRequestMessage.class);

modules/core/src/main/java/org/apache/ignite/internal/DeploymentAware.java

@@ -0,0 +1,43 @@
+/*
+ * 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.managers.deployment.GridDeploymentInfo;
+import org.apache.ignite.plugin.extensions.communication.Message;
+
+/**
+ * Implemented by messages that carry classes deployed from another node. The deployment lets the generated marshaller
+ * resolve the class loader those classes are read with, the same way {@code CacheIdAware} lets it resolve the cache
+ * object context.
+ * <p>
+ * Resolving may have to request the deployment from its owner and block, and it fails when the classes are gone, so a
+ * message stating this must be unmarshalled where blocking is allowed and where the failure reaches whoever waits for
+ * it. A message read on a socket-reading thread promises neither: a discovery custom message, for one, is a nested
+ * field of its envelope, so the envelope's marshaller reads the whole tree there, and a missing class is swallowed
+ * with a warning. Such a message keeps its deployment as a plain field and asks {@code GridDeploymentManager} for the
+ * loader where it is read, as {@code StartRequestData} does.
+ *
+ * @see MarshallableMessage
+ */
+public interface DeploymentAware extends Message {
+    /** @return Deployment of the classes the message carries. */
+    public GridDeploymentInfo deploymentInfo();
+
+    /** @return Name of a class the deployment must be able to load. */
+    public String deployedClassName();
+}

modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java

@@ -34,7 +34,7 @@
 import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
 import org.apache.ignite.internal.managers.deployment.GridDeployment;
 import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
-import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
+import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoMessage;
 import org.apache.ignite.internal.managers.deployment.P2PClassLoadingIssues;
 import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -403,7 +403,7 @@ private boolean filterDropsEvent(Event evt) {
             if (dep == null)
                 throw new IgniteDeploymentCheckedException("Failed to deploy event filter: " + filter);
 
-            depInfo = new GridDeploymentInfoBean(dep);
+            depInfo = new GridDeploymentInfoMessage(dep);
 
             filterBytes = U.marshal(ctx.marshaller(), filter);
         }
@@ -417,11 +417,7 @@ private boolean filterDropsEvent(Event evt) {
 
         if (filterBytes != null) {
             try {
-                GridDeployment dep = ctx.deploy().getGlobalDeployment(depInfo.deployMode(), clsName, clsName,
-                    depInfo.userVersion(), nodeId, depInfo.classLoaderId(), depInfo.participants(), null);
-
-                if (dep == null)
-                    throw new IgniteDeploymentCheckedException("Failed to obtain deployment for class: " + clsName);
+                GridDeployment dep = ctx.deploy().globalDeployment(depInfo, clsName);
 
                 filter = U.unmarshal(ctx, filterBytes, U.resolveClassLoader(dep.classLoader(), ctx.config()));