IGNITE-28528 Revise marshalling of the node filter in StartRequestData

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/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, nodeId);
 
                 filter = U.unmarshal(ctx, filterBytes, U.resolveClassLoader(dep.classLoader(), ctx.config()));
 

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

@@ -24,10 +24,10 @@
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.compute.ComputeJob;
 import org.apache.ignite.compute.ComputeJobSibling;
-import org.apache.ignite.configuration.DeploymentMode;
+import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
+import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoMessage;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.util.tostring.GridToStringExclude;
-import org.apache.ignite.internal.util.tostring.GridToStringInclude;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgnitePredicate;
@@ -70,22 +70,17 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage, DeferredUnma
     @Order(5)
     String taskName;
 
-    /** */
+    /** Deployment of the task classes. */
     @Order(6)
-    String userVer;
+    GridDeploymentInfoMessage depInfo;
 
     /** */
     @Order(7)
     String taskClsName;
 
-    /** Node class loader participants. */
-    @GridToStringInclude
-    @Order(8)
-    Map<UUID, IgniteUuid> ldrParticipants;
-
     /** */
     @GridToStringExclude
-    @Order(9)
+    @Order(8)
     byte[] sesAttrsBytes;
 
     /** */
@@ -95,7 +90,7 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage, DeferredUnma
 
     /** */
     @GridToStringExclude
-    @Order(10)
+    @Order(9)
     byte[] jobAttrsBytes;
 
     /** */
@@ -104,70 +99,62 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage, DeferredUnma
     Map<? extends Serializable, ? extends Serializable> jobAttrs;
 
     /** Checkpoint SPI name. */
-    @Order(11)
+    @Order(10)
     String cpSpi;
 
     /** Left unset for a continuous task: such a job requests its siblings from the task node instead. */
     @Marshalled("siblingsBytes")
     Collection<ComputeJobSibling> siblings;
 
     /** */
-    @Order(12)
+    @Order(11)
     byte[] siblingsBytes;
 
     /** Transient since needs to hold local creation time. */
     private final long createTime = U.currentTimeMillis();
 
     /** */
-    @Order(13)
-    IgniteUuid clsLdrId;
-
-    /** */
-    @Order(14)
-    DeploymentMode depMode;
-
-    /** */
-    @Order(15)
+    @Order(12)
     boolean dynamicSiblings;
 
     /** */
-    @Order(16)
+    @Order(13)
     boolean forceLocDep;
 
     /** */
-    @Order(17)
+    @Order(14)
     boolean sesFullSup;
 
     /** */
-    @Order(18)
+    @Order(15)
     boolean internal;
 
     /** */
-    @Order(19)
+    @Order(16)
     Collection<UUID> top;
 
     /** */
     @Marshalled("topPredBytes")
     IgnitePredicate<ClusterNode> topPred;
 
     /** */
-    @Order(20)
+    @Order(17)
     byte[] topPredBytes;
 
     /** */
-    @Order(21)
+    @Order(18)
     int[] cacheIds;
 
     /** */
-    @Order(22)
+    @Order(19)
     int part;
 
     /** */
-    @Order(23)
+    @Order(20)
     AffinityTopologyVersion topVer;
 
     /** */
-    @Order(24)
+    @Order(21)
     String execName;
 
     /**
@@ -181,7 +168,7 @@ public GridJobExecuteRequest() {
      * @param sesId Task session ID.
      * @param jobId Job ID.
      * @param taskName Task name.
-     * @param userVer Code version.
+     * @param depInfo Deployment of the task classes.
      * @param taskClsName Fully qualified task name.
      * @param job Job.
      * @param startTaskTime Task execution start time.
@@ -192,10 +179,7 @@ public GridJobExecuteRequest() {
      * @param sesAttrs Session attributes.
      * @param jobAttrs Job attributes.
      * @param cpSpi Collision SPI.
-     * @param clsLdrId Task local class loader id.
-     * @param depMode Task deployment mode.
      * @param dynamicSiblings {@code True} if siblings are dynamic.
-     * @param ldrParticipants Other node class loader IDs that can also load classes.
      * @param forceLocDep {@code True} If remote node should ignore deployment settings.
      * @param sesFullSup {@code True} if session attributes are disabled.
      * @param internal {@code True} if internal job.
@@ -208,7 +192,7 @@ public GridJobExecuteRequest(
             IgniteUuid sesId,
             IgniteUuid jobId,
             String taskName,
-            String userVer,
+            GridDeploymentInfo depInfo,
             String taskClsName,
             ComputeJob job,
             long startTaskTime,
@@ -219,10 +203,7 @@ public GridJobExecuteRequest(
             Map<Object, Object> sesAttrs,
             Map<? extends Serializable, ? extends Serializable> jobAttrs,
             String cpSpi,
-            IgniteUuid clsLdrId,
-            DeploymentMode depMode,
             boolean dynamicSiblings,
-            Map<UUID, IgniteUuid> ldrParticipants,
             boolean forceLocDep,
             boolean sesFullSup,
             boolean internal,
@@ -238,14 +219,12 @@ public GridJobExecuteRequest(
         assert sesAttrs != null || !sesFullSup;
         assert jobAttrs != null;
         assert top != null || topPred != null;
-        assert clsLdrId != null;
-        assert userVer != null;
-        assert depMode != null;
+        assert depInfo != null;
 
         this.sesId = sesId;
         this.jobId = jobId;
         this.taskName = taskName;
-        this.userVer = userVer;
+        this.depInfo = new GridDeploymentInfoMessage(depInfo);
         this.taskClsName = taskClsName;
         this.job = job;
         this.startTaskTime = startTaskTime;
@@ -256,10 +235,7 @@ public GridJobExecuteRequest(
         this.siblings = dynamicSiblings ? null : siblings;
         this.sesAttrs = sesAttrs;
         this.jobAttrs = jobAttrs;
-        this.clsLdrId = clsLdrId;
-        this.depMode = depMode;
         this.dynamicSiblings = dynamicSiblings;
-        this.ldrParticipants = ldrParticipants;
         this.forceLocDep = forceLocDep;
         this.sesFullSup = sesFullSup;
         this.internal = internal;
@@ -285,6 +261,11 @@ public IgniteUuid jobId() {
         return jobId;
     }
 
+    /** @return Deployment of the task classes. */
+    public GridDeploymentInfo deploymentInfo() {
+        return depInfo;
+    }
+
     /**
      * @return Task class name.
      */
@@ -299,13 +280,6 @@ public String taskName() {
         return taskName;
     }
 
-    /**
-     * @return Task version.
-     */
-    public String userVersion() {
-        return userVer;
-    }
-
     /**
      * @return Grid job.
      */
@@ -364,27 +338,6 @@ public String checkpointSpi() {
         return cpSpi;
     }
 
-    /**
-     * @return Task local class loader id.
-     */
-    public IgniteUuid classLoaderId() {
-        return clsLdrId;
-    }
-
-    /**
-     * @return Deployment mode.
-     */
-    public DeploymentMode deploymentMode() {
-        return depMode;
-    }
-
-    /**
-     * @return Node class loader participant map.
-     */
-    public Map<UUID, IgniteUuid> loaderParticipants() {
-        return ldrParticipants;
-    }
-
     /**
      * @return Returns {@code true} if deployment should always be used.
      */
@@ -446,7 +399,6 @@ public AffinityTopologyVersion topologyVersion() {
         return topVer;
     }
 
-
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(GridJobExecuteRequest.class, this);

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

@@ -28,7 +28,7 @@
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.internal.managers.deployment.GridDeployment;
-import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
+import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoMessage;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.continuous.GridContinuousBatch;
 import org.apache.ignite.internal.processors.continuous.GridContinuousBatchAdapter;
@@ -64,7 +64,7 @@ public class GridMessageListenHandler implements GridContinuousHandler {
     private String clsName;
 
     /** */
-    private GridDeploymentInfoBean depInfo;
+    private GridDeploymentInfoMessage depInfo;
 
     /** */
     private boolean depEnabled;
@@ -166,7 +166,7 @@ public GridMessageListenHandler(@Nullable Object topic, IgniteBiPredicate<UUID,
         if (dep == null)
             throw new IgniteDeploymentCheckedException("Failed to deploy message listener.");
 
-        depInfo = new GridDeploymentInfoBean(dep);
+        depInfo = new GridDeploymentInfoMessage(dep);
 
         depEnabled = true;
     }
@@ -178,13 +178,7 @@ public GridMessageListenHandler(@Nullable Object topic, IgniteBiPredicate<UUID,
         assert ctx.config().isPeerClassLoadingEnabled();
 
         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);
-
-            ClassLoader ldr = dep.classLoader();
+            ClassLoader ldr = ctx.deploy().globalDeployment(depInfo, clsName, nodeId).classLoader();
 
             if (topicBytes != null)
                 topic = U.unmarshal(ctx, topicBytes, U.resolveClassLoader(ldr, ctx.config()));
@@ -260,7 +254,7 @@ public GridMessageListenHandler(@Nullable Object topic, IgniteBiPredicate<UUID,
             topicBytes = U.readByteArray(in);
             predBytes = U.readByteArray(in);
             clsName = U.readString(in);
-            depInfo = (GridDeploymentInfoBean)in.readObject();
+            depInfo = (GridDeploymentInfoMessage)in.readObject();
         }
         else {
             topic = in.readObject();

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java

@@ -2422,10 +2422,7 @@ public void sendUserMessage(Collection<? extends ClusterNode> nodes,
             depClsName,
             topic,
             serTopic,
-            dep != null ? dep.classLoaderId() : null,
-            dep != null ? dep.deployMode() : null,
-            dep != null ? dep.userVersion() : null,
-            dep != null ? dep.participants() : null);
+            dep);
 
         if (ordered)
             sendOrderedMessageToGridTopic(nodes, TOPIC_COMM_USER, ioMsg, PUBLIC_POOL, timeout, true);
@@ -3635,21 +3632,15 @@ private class GridUserMessageListener implements GridMessageListener {
 
                     if (dep == null && ctx.config().isPeerClassLoadingEnabled() &&
                         ioMsg.deploymentClassName() != null) {
-                        dep = ctx.deploy().getGlobalDeployment(
-                            ioMsg.deploymentMode(),
-                            ioMsg.deploymentClassName(),
-                            ioMsg.deploymentClassName(),
-                            ioMsg.userVersion(),
-                            nodeId,
-                            ioMsg.classLoaderId(),
-                            ioMsg.loaderParticipants(),
-                            null);
+                        dep = ctx.deploy().globalDeployment(ioMsg.deploymentInfo(), ioMsg.deploymentClassName(),
+                            ioMsg.deploymentClassName(), nodeId);
 
-                        if (dep == null)
+                        if (dep == null) {
                             throw new IgniteDeploymentCheckedException(
                                 "Failed to obtain deployment information for user message. " +
                                     "If you are using custom message or topic class, try implementing " +
                                     "GridPeerDeployAware interface. [msg=" + ioMsg + ']');
+                        }
 
                         ioMsg.deployment(dep); // Cache deployment.
                     }