IGNITE-28607 WIP

Author: shishkovilja

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

@@ -40,8 +40,10 @@
 import org.apache.ignite.internal.managers.encryption.GenerateEncryptionKeyRequest;
 import org.apache.ignite.internal.managers.encryption.GenerateEncryptionKeyResponse;
 import org.apache.ignite.internal.managers.encryption.GroupKeyEncrypted;
+import org.apache.ignite.internal.managers.encryption.KnownEncryptionKeys;
 import org.apache.ignite.internal.managers.encryption.MasterKeyChangeRequest;
 import org.apache.ignite.internal.managers.encryption.NodeEncryptionKeys;
+import org.apache.ignite.internal.managers.eventstorage.EnabledEvents;
 import org.apache.ignite.internal.managers.eventstorage.GridEventStorageMessage;
 import org.apache.ignite.internal.plugin.AbstractMarshallableMessageFactoryProvider;
 import org.apache.ignite.internal.processors.authentication.User;
@@ -181,6 +183,8 @@
 import org.apache.ignite.internal.processors.cluster.CacheMetricsMessage;
 import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessage;
 import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
+import org.apache.ignite.internal.processors.cluster.ClusterFlags;
+import org.apache.ignite.internal.processors.cluster.ClusterIdAndTag;
 import org.apache.ignite.internal.processors.cluster.ClusterMetricsUpdateMessage;
 import org.apache.ignite.internal.processors.cluster.NodeFullMetricsMessage;
 import org.apache.ignite.internal.processors.cluster.NodeMetricsMessage;
@@ -205,6 +209,7 @@
 import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageCasMessage;
 import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageUpdateAckMessage;
 import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageUpdateMessage;
+import org.apache.ignite.internal.processors.plugin.PluginsData;
 import org.apache.ignite.internal.processors.query.InlineSizesData;
 import org.apache.ignite.internal.processors.query.QueryField;
 import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryCancelRequest;
@@ -253,7 +258,7 @@
 import org.apache.ignite.spi.communication.tcp.messages.HandshakeWaitMessage;
 import org.apache.ignite.spi.communication.tcp.messages.NodeIdMessage;
 import org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage;
-import org.apache.ignite.spi.discovery.ObjectData;
+import org.apache.ignite.spi.discovery.DataBagItem;
 import org.apache.ignite.spi.discovery.tcp.internal.DiscoveryDataPacket;
 import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode;
 import org.apache.ignite.spi.discovery.tcp.messages.InetAddressMessage;
@@ -353,7 +358,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchema(GridCacheVersion.class);
         withNoSchema(GridCacheVersionEx.class);
         withNoSchema(WALPointer.class);
-        withNoSchemaResolvedClassLoader(ObjectData.class);
+        withNoSchemaResolvedClassLoader(DataBagItem.class);
         withSchemaResolvedClassLoader(GridTopicMessage.class);
 
         // [5700 - 5900]: Discovery originated messages.
@@ -652,10 +657,12 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchema(MasterKeyChangeRequest.class);
         withNoSchema(GroupKeyEncrypted.class);
         withNoSchema(NodeEncryptionKeys.class);
+        withNoSchema(KnownEncryptionKeys.class);
 
         // [13000 - 13300]: Control, configuration, diagnostincs and other messages.
         msgIdx = 13000;
         withSchema(GridEventStorageMessage.class);
+        withSchema(EnabledEvents.class);
         withNoSchema(ChangeGlobalStateMessage.class);
         withNoSchema(GridChangeGlobalStateMessageResponse.class);
         withSchema(IgniteDiagnosticRequest.class);
@@ -665,6 +672,9 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchemaResolvedClassLoader(DynamicCacheChangeRequest.class);
         withNoSchema(PartitionHashRecord.class);
         withNoSchema(TransactionsHashRecord.class);
+        withNoSchema(ClusterIdAndTag.class);
+        withNoSchema(ClusterFlags.class);
+        withNoSchemaResolvedClassLoader(PluginsData.class);
 
         assert msgIdx <= MAX_MESSAGE_ID;
     }

modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java

@@ -563,28 +563,23 @@ public void onLocalJoin() {
             }
         }
 
-        dataBag.addGridCommonData(ENCRYPTION_MGR.ordinal(), knownEncKeys);
+        dataBag.addGridCommonData(ENCRYPTION_MGR.ordinal(), new KnownEncryptionKeys(knownEncKeys));
     }
 
     /** {@inheritDoc} */
     @Override public void onGridDataReceived(GridDiscoveryData data) {
         if (ctx.clientNode())
             return;
 
-        Map<Integer, Object> encKeysFromCluster = (Map<Integer, Object>)data.commonData();
+        KnownEncryptionKeys encKeysFromCluster = data.commonData();
 
-        if (F.isEmpty(encKeysFromCluster))
+        if (encKeysFromCluster == null || F.isEmpty(encKeysFromCluster.keys))
             return;
 
-        for (Map.Entry<Integer, Object> entry : encKeysFromCluster.entrySet()) {
+        for (Map.Entry<Integer, GroupKeyEncrypted> entry : encKeysFromCluster.keys.entrySet()) {
             int grpId = entry.getKey();
 
-            GroupKeyEncrypted rmtKey;
-
-            if (entry.getValue() instanceof GroupKeyEncrypted)
-                rmtKey = (GroupKeyEncrypted)entry.getValue();
-            else
-                rmtKey = new GroupKeyEncrypted(INITIAL_KEY_ID, (byte[])entry.getValue());
+            GroupKeyEncrypted rmtKey = entry.getValue();
 
             GroupKey locGrpKey = getActiveKey(grpId);
 

modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/KnownEncryptionKeys.java

@@ -0,0 +1,37 @@
+/*
+ * 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.managers.encryption;
+
+import java.util.Map;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.plugin.extensions.communication.Message;
+
+/** */
+public class KnownEncryptionKeys implements Message {
+    /** */
+    @Order(0)
+    Map<Integer, GroupKeyEncrypted> keys;
+
+    /** */
+    public KnownEncryptionKeys() {}
+
+    /** */
+    KnownEncryptionKeys(Map<Integer, GroupKeyEncrypted> keys) {
+        this.keys = keys;
+    }
+}

modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/EnabledEvents.java

@@ -0,0 +1,40 @@
+/*
+ * 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.managers.eventstorage;
+
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.plugin.extensions.communication.Message;
+
+/**
+ *
+ */
+public class EnabledEvents implements Message {
+    /** */
+    @Order(0)
+    int[] evts;
+
+    /** */
+    public EnabledEvents() { }
+
+    /**
+     * @param events Enabled events.
+     */
+    public EnabledEvents(int[] events) {
+        this.evts = events;
+    }
+}

modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java

@@ -1173,13 +1173,15 @@ private int[] copy(int[] arr) {
 
     /** {@inheritDoc} */
     @Override public void onGridDataReceived(DiscoveryDataBag.GridDiscoveryData data) {
-        if (data.commonData() == null)
+        EnabledEvents enabled = data.commonData();
+
+        if (enabled == null)
             return;
 
         if (ctx.clientNode())
             return;
 
-        GridIntList clusterData = new GridIntList((int[])data.commonData());
+        GridIntList clusterData = new GridIntList(enabled.evts);
         GridIntList nodeData = new GridIntList(enabledEvents());
 
         GridIntList toEnable = new GridIntList(clusterData.size());
@@ -1207,9 +1209,7 @@ private int[] copy(int[] arr) {
         if (dataBag.isJoiningNodeClient() && dataBag.commonDataCollectedFor(EVENT_MGR.ordinal()))
             return;
 
-        int[] clusterData = enabledEvents();
-
-        dataBag.addGridCommonData(EVENT_MGR.ordinal(), clusterData);
+        dataBag.addGridCommonData(EVENT_MGR.ordinal(), new EnabledEvents(enabledEvents()));
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/authentication/IgniteAuthenticationProcessor.java

@@ -430,7 +430,7 @@ private boolean isLocalNodeCoordinator() {
 
     /** {@inheritDoc} */
     @Override public void onGridDataReceived(DiscoveryDataBag.GridDiscoveryData data) {
-        initUsrs = (InitialUsersData)data.commonData();
+        initUsrs = data.commonData();
     }
 
     /** {@inheritDoc} */

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

@@ -1504,10 +1504,9 @@ public void onGridDataReceived(DiscoveryDataBag.GridDiscoveryData data) {
         if (data.commonData() == null)
             return;
 
-        assert joinDiscoData != null || disconnectedState();
-        assert data.commonData() instanceof CacheNodeCommonDiscoveryData : data;
+        CacheNodeCommonDiscoveryData cachesData = data.commonData();
 
-        CacheNodeCommonDiscoveryData cachesData = (CacheNodeCommonDiscoveryData)data.commonData();
+        assert joinDiscoData != null || disconnectedState();
 
         // CacheGroup configurations that were created from local node configuration.
         Map<Integer, CacheGroupDescriptor> locCacheGrps = new HashMap<>(registeredCacheGroups());

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

@@ -18,7 +18,6 @@
 package org.apache.ignite.internal.processors.cache.binary;
 
 import java.io.File;
-import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Collection;
@@ -1464,7 +1463,7 @@ private IgniteNodeValidationResult validateBinaryMetadata(UUID rmtNodeId, Binary
                     res.put(e.getKey(), e.getValue());
             }
 
-            dataBag.addGridCommonData(BINARY_PROC.ordinal(), (Serializable)res);
+            dataBag.addGridCommonData(BINARY_PROC.ordinal(), new BinaryMetadataVersionsData(res));
         }
     }
 
@@ -1530,10 +1529,10 @@ private IgniteNodeValidationResult validateBinaryMetadata(UUID rmtNodeId, Binary
 
     /** {@inheritDoc} */
     @Override public void onGridDataReceived(GridDiscoveryData data) {
-        Map<Integer, BinaryMetadataVersionInfo> receivedData = (Map<Integer, BinaryMetadataVersionInfo>)data.commonData();
+        BinaryMetadataVersionsData receivedData = data.commonData();
 
-        if (receivedData != null) {
-            for (Map.Entry<Integer, BinaryMetadataVersionInfo> e : receivedData.entrySet()) {
+        if (receivedData != null && !F.isEmpty(receivedData.data)) {
+            for (Map.Entry<Integer, BinaryMetadataVersionInfo> e : receivedData.data.entrySet()) {
                 BinaryMetadataVersionInfo metaVerInfo = e.getValue();
 
                 BinaryMetadataVersionInfo locMetaVerInfo = new BinaryMetadataVersionInfo(metaVerInfo.metadata(),

modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterFlags.java

@@ -0,0 +1,38 @@
+/*
+ * 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.processors.cluster;
+
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.plugin.extensions.communication.Message;
+
+/** */
+public class ClusterFlags implements Message {
+    /** Update notifier enabled flag. */
+    @Order(0)
+    boolean updateNotifierEnabled;
+
+    /** */
+    public ClusterFlags() { }
+
+    /**
+     * @param updateNotifierEnabled Update notifier enabled flag.
+     */
+    public ClusterFlags(boolean updateNotifierEnabled) {
+        this.updateNotifierEnabled = updateNotifierEnabled;
+    }
+}

modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/ClusterIdAndTag.java

@@ -20,20 +20,27 @@
 import java.io.Serializable;
 import java.util.Objects;
 import java.util.UUID;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.Message;
 
 /**
  * Container class to send cluster ID and tag in disco data and to write them atomically to metastorage.
  */
-public class ClusterIdAndTag implements Serializable {
+public class ClusterIdAndTag implements Serializable, Message {
     /** */
     private static final long serialVersionUID = 0L;
 
     /** */
-    private final UUID id;
+    @Order(0)
+    UUID id;
 
     /** */
-    private final String tag;
+    @Order(1)
+    String tag;
+
+    /** */
+    public ClusterIdAndTag() { }
 
     /**
      * @param id Cluster ID.