IGNITE-28516 Use MessageSerializer for DynamicCacheChangeRequest (#13032)

Author: Vladsz83

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

@@ -48,6 +48,7 @@
 import org.apache.ignite.internal.processors.authentication.UserManagementOperationFinishedMessage;
 import org.apache.ignite.internal.processors.authentication.UserProposedMessage;
 import org.apache.ignite.internal.processors.cache.CacheAffinityChangeMessage;
+import org.apache.ignite.internal.processors.cache.CacheConfigurationEnrichment;
 import org.apache.ignite.internal.processors.cache.CacheEntryPredicateAdapter;
 import org.apache.ignite.internal.processors.cache.CacheEvictionEntry;
 import org.apache.ignite.internal.processors.cache.CacheInvokeDirectResult;
@@ -56,6 +57,7 @@
 import org.apache.ignite.internal.processors.cache.ClientCacheChangeDiscoveryMessage;
 import org.apache.ignite.internal.processors.cache.ClientCacheChangeDummyDiscoveryMessage;
 import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
+import org.apache.ignite.internal.processors.cache.DynamicCacheChangeRequest;
 import org.apache.ignite.internal.processors.cache.ExchangeFailureMessage;
 import org.apache.ignite.internal.processors.cache.GridCacheEntryInfo;
 import org.apache.ignite.internal.processors.cache.GridCacheReturn;
@@ -628,14 +630,16 @@ public CoreMessagesProvider(Marshaller schemaAwareMarhaller, Marshaller schemaLe
         withNoSchema(ChangeCacheEncryptionRequest.class);
         withNoSchema(MasterKeyChangeRequest.class);
 
-        // [13000 - 13300]: Control, diagnostincs and other messages.
+        // [13000 - 13300]: Control, configuration, diagnostincs and other messages.
         msgIdx = 13000;
         withSchema(GridEventStorageMessage.class);
         withNoSchema(ChangeGlobalStateMessage.class);
         withNoSchema(GridChangeGlobalStateMessageResponse.class);
         withSchema(IgniteDiagnosticRequest.class);
         withNoSchema(IgniteDiagnosticResponse.class);
         withNoSchema(WalStateAckMessage.class);
+        withNoSchema(CacheConfigurationEnrichment.class);
+        withNoSchemaResolvedClassLoader(DynamicCacheChangeRequest.class);
 
         assert msgIdx <= MAX_MESSAGE_ID;
     }

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

@@ -20,25 +20,35 @@
 import java.io.Serializable;
 import java.util.Map;
 import java.util.Set;
+import org.apache.ignite.internal.CoreMessagesProvider;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
 import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.Message;
 
 /**
  * Object that contains serialized values for fields marked with {@link org.apache.ignite.configuration.SerializeSeparately}
  * in {@link org.apache.ignite.configuration.CacheConfiguration}.
  * This object is needed to exchange and store shrinked cache configurations to avoid possible {@link ClassNotFoundException} errors
  * during deserialization on nodes where some specific class may not exist.
  */
-public class CacheConfigurationEnrichment implements Serializable {
+public class CacheConfigurationEnrichment implements Message, Serializable {
     /** */
     private static final long serialVersionUID = 0L;
 
     /** Field name -> Field serialized value. */
-    private final Map<String, byte[]> enrichFields;
+    @Order(0)
+    Map<String, byte[]> enrichFields;
 
     /** Field name -> Field value class name. */
     @GridToStringInclude
-    private final Map<String, String> fieldClassNames;
+    @Order(1)
+    Map<String, String> fieldClassNames;
+
+    /** Empty constructor for {@link CoreMessagesProvider}. */
+    public CacheConfigurationEnrichment() {
+        // No-op.
+    }
 
     /**
      * Creates a new instance of CacheConfigurationEnrichment.

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

@@ -1161,7 +1161,7 @@ else if (encMgr.masterKeyDigest() != null &&
             req.initiatingNodeId(),
             req.deploymentId(),
             req.encryptionKey(),
-            req.encryptionKeyId(),
+            req.encryptionKeyId() < 0 ? null : req.encryptionKeyId(),
             req.cacheConfigurationEnrichment()
         );
 

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

@@ -19,8 +19,6 @@
 
 import java.util.Collection;
 import java.util.Set;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.MarshallableMessage;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.managers.discovery.DiscoCache;
 import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
@@ -31,26 +29,22 @@
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteUuid;
-import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.Message;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * Cache change batch.
  */
-public class DynamicCacheChangeBatch implements DiscoveryCustomMessage, MarshallableMessage {
+public class DynamicCacheChangeBatch implements DiscoveryCustomMessage, Message {
     /** Discovery custom message ID. */
     @Order(0)
     IgniteUuid id;
 
     /** Change requests. */
     @GridToStringInclude
-    Collection<DynamicCacheChangeRequest> reqs;
-
-    /** JDK Serialized version of reqs. */
     @Order(1)
-    byte[] requestsBytes;
+    Collection<DynamicCacheChangeRequest> reqs;
 
     /** Cache updates to be executed on exchange. */
     private ExchangeActions exchangeActions;
@@ -172,18 +166,6 @@ public void startCaches(boolean startCaches) {
         this.startCaches = startCaches;
     }
 
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
-        if (reqs != null)
-            requestsBytes = U.marshal(marsh, reqs);
-    }
-
-    /** {@inheritDoc} */
-    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
-        if (requestsBytes != null)
-            reqs = U.unmarshal(marsh, requestsBytes, clsLdr);
-    }
-
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(DynamicCacheChangeBatch.class, this);

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

@@ -19,96 +19,139 @@
 
 import java.io.Serializable;
 import java.util.UUID;
+import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.CoreMessagesProvider;
 import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.MarshallableMessage;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.processors.query.QuerySchema;
 import org.apache.ignite.internal.util.tostring.GridToStringExclude;
 import org.apache.ignite.internal.util.typedef.T2;
+import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.marshaller.Marshaller;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * Cache start/stop request.
  */
-public class DynamicCacheChangeRequest implements Serializable {
+public class DynamicCacheChangeRequest implements MarshallableMessage, Serializable {
     /** */
     private static final long serialVersionUID = 0L;
 
     /** */
-    private UUID reqId;
+    @Order(0)
+    UUID reqId;
 
     /** Start ID. */
-    private IgniteUuid deploymentId;
+    @Order(1)
+    IgniteUuid deploymentId;
 
     /** Stop cache name. */
     @GridToStringExclude
-    private String cacheName;
+    @Order(2)
+    String cacheName;
 
     /** Cache start configuration. */
     @GridToStringExclude
-    private CacheConfiguration startCfg;
+    private CacheConfiguration<?, ?> startCfg;
+
+    /** Bytes of {@link #startCfg}. */
+    @Order(3)
+    byte[] cfgBytes;
 
     /** Cache type. */
-    private CacheType cacheType;
+    @Order(4)
+    CacheType cacheType;
 
     /** Near node ID in case if near cache is being started. */
-    private UUID initiatingNodeId;
+    @Order(5)
+    UUID initiatingNodeId;
 
     /** Near cache configuration. */
     @GridToStringExclude
-    private NearCacheConfiguration nearCacheCfg;
+    private NearCacheConfiguration<?, ?> nearCacheCfg;
+
+    /** Bytes of {@link #nearCacheCfg}. */
+    @Order(6)
+    byte[] nearCfgBytes;
 
     /** Start only client cache, do not start data nodes. */
-    private boolean clientStartOnly;
+    @Order(7)
+    boolean clientStartOnly;
 
     /** Stop flag. */
-    private boolean stop;
+    @Order(8)
+    boolean stop;
 
     /** Restart flag. */
-    private boolean restart;
+    @Order(9)
+    boolean restart;
 
     /** Finalize update counters flag. */
-    private boolean finalizePartitionCounters;
+    @Order(10)
+    boolean finalizePartitionCounters;
 
     /** Restart operation id. */
-    private IgniteUuid restartId;
+    @Order(11)
+    IgniteUuid restartId;
 
     /** Cache active on start or not*/
-    private boolean disabledAfterStart;
+    @Order(12)
+    boolean disabledAfterStart;
 
     /** Cache data destroy flag. Setting to <code>true</code> will cause removing all cache data.*/
-    private boolean destroy;
+    @Order(13)
+    boolean destroy;
 
     /** Whether cache was created through SQL. */
-    private boolean sql;
+    @Order(14)
+    boolean sql;
 
     /** Fail if exists flag. */
-    private boolean failIfExists;
+    @Order(15)
+    boolean failIfExists;
 
     /** Template configuration flag. */
-    private boolean template;
+    @Order(16)
+    boolean template;
 
     /** Reset lost partitions flag. */
-    private boolean resetLostPartitions;
+    @Order(17)
+    boolean resetLostPartitions;
 
     /** Dynamic schema. */
-    private QuerySchema schema;
+    QuerySchema schema;
 
-    /** */
-    private transient boolean locallyConfigured;
+    /** Bytes of {@link #schema}. */
+    @Order(18)
+    byte[] schemaBytes;
+
+    /** Is transient. */
+    private boolean locallyConfigured;
 
     /** Encryption key. */
-    @Nullable private byte[] encKey;
+    @Order(19)
+    @Nullable byte[] encKey;
 
     /** Id of encryption key. */
-    @Nullable private Integer encKeyId;
+    @Order(20)
+    int encKeyId;
 
     /** Master key digest. */
-    @Nullable private byte[] masterKeyDigest;
+    @Order(21)
+    @Nullable byte[] masterKeyDigest;
 
     /** Cache configuration enrichment. */
-    private CacheConfigurationEnrichment cacheCfgEnrichment;
+    @Order(22)
+    CacheConfigurationEnrichment cacheCfgEnrichment;
+
+    /** Empty constructor for {@link CoreMessagesProvider}. */
+    public DynamicCacheChangeRequest() {
+        // No-op.
+    }
 
     /**
      * @param reqId Unique request ID.
@@ -123,6 +166,32 @@ public DynamicCacheChangeRequest(UUID reqId, String cacheName, UUID initiatingNo
         this.initiatingNodeId = initiatingNodeId;
     }
 
+    /** {@inheritDoc} */
+    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+        cfgBytes = U.marshal(marsh, startCfg);
+
+        if (nearCacheCfg != null)
+            nearCfgBytes = U.marshal(marsh, nearCacheCfg);
+
+        if (schema != null)
+            schemaBytes = U.marshal(marsh, schema);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+        startCfg = U.unmarshal(marsh, cfgBytes, clsLdr);
+
+        if (nearCfgBytes != null)
+            nearCacheCfg = U.unmarshal(marsh, nearCfgBytes, clsLdr);
+
+        if (schemaBytes != null)
+            schema = U.unmarshal(marsh, schemaBytes, clsLdr);
+
+        cfgBytes = null;
+        nearCfgBytes = null;
+        schemaBytes = null;
+    }
+
     /**
      * @param ctx Context.
      * @param cacheName Cache name.
@@ -338,28 +407,28 @@ public UUID initiatingNodeId() {
     /**
      * @return Near cache configuration.
      */
-    public NearCacheConfiguration nearCacheConfiguration() {
+    public NearCacheConfiguration<?, ?> nearCacheConfiguration() {
         return nearCacheCfg;
     }
 
     /**
      * @param nearCacheCfg Near cache configuration.
      */
-    public void nearCacheConfiguration(NearCacheConfiguration nearCacheCfg) {
+    public void nearCacheConfiguration(NearCacheConfiguration<?, ?> nearCacheCfg) {
         this.nearCacheCfg = nearCacheCfg;
     }
 
     /**
      * @return Cache configuration.
      */
-    public CacheConfiguration startCacheConfiguration() {
+    public CacheConfiguration<?, ?> startCacheConfiguration() {
         return startCfg;
     }
 
     /**
      * @param startCfg Cache configuration.
      */
-    public void startCacheConfiguration(CacheConfiguration startCfg) {
+    public void startCacheConfiguration(CacheConfiguration<?, ?> startCfg) {
         this.startCfg = startCfg;
 
         if (startCfg.getNearConfiguration() != null)
@@ -485,14 +554,14 @@ public void encryptionKey(@Nullable byte[] encKey) {
      *
      * @param encKeyId Encryption key id.
      */
-    public void encryptionKeyId(@Nullable Integer encKeyId) {
+    public void encryptionKeyId(int encKeyId) {
         this.encKeyId = encKeyId;
     }
 
     /**
      * @return Encryption key id.
      */
-    @Nullable public Integer encryptionKeyId() {
+    @Nullable public int encryptionKeyId() {
         return encKeyId;
     }
 

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

@@ -5087,7 +5087,9 @@ private DynamicCacheChangeRequest prepareCacheChangeRequest(
 
         req.encryptionKey(encKey);
 
-        req.encryptionKeyId(encKeyId);
+        assert encKeyId == null || encKeyId >= 0;
+
+        req.encryptionKeyId(encKeyId == null ? -1 : encKeyId);
 
         req.restartId(restartId);