Add ClusterFlags for update notifier

Author: shishkovilja

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

@@ -183,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;
@@ -669,6 +671,8 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchemaResolvedClassLoader(DynamicCacheChangeRequest.class);
         withNoSchema(PartitionHashRecord.class);
         withNoSchema(TransactionsHashRecord.class);
+        withNoSchema(ClusterIdAndTag.class);
+        withNoSchema(ClusterFlags.class);
 
         assert msgIdx <= MAX_MESSAGE_ID;
     }

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.

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

@@ -19,7 +19,6 @@
 
 import java.io.Serializable;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 import java.util.Timer;
@@ -76,6 +75,7 @@
 import org.apache.ignite.lang.IgniteUuid;
 import org.apache.ignite.metric.MetricRegistry;
 import org.apache.ignite.mxbean.IgniteClusterMXBean;
+import org.apache.ignite.plugin.extensions.communication.Message;
 import org.apache.ignite.spi.discovery.DiscoveryDataBag;
 import org.apache.ignite.spi.discovery.DiscoveryDataBag.GridDiscoveryData;
 import org.apache.ignite.spi.discovery.DiscoveryMetricsProvider;
@@ -465,33 +465,22 @@ public IgniteFuture<?> clientReconnectFuture() {
 
     /** {@inheritDoc} */
     @Override public void collectJoiningNodeData(DiscoveryDataBag dataBag) {
-        dataBag.addJoiningNodeData(CLUSTER_PROC.ordinal(), getDiscoveryData());
+        dataBag.addJoiningNodeData(CLUSTER_PROC.ordinal(), new ClusterFlags(notifyEnabled.get()));
     }
 
     /** {@inheritDoc} */
     @Override public void collectGridNodeData(DiscoveryDataBag dataBag) {
-        dataBag.addNodeSpecificData(CLUSTER_PROC.ordinal(), getDiscoveryData());
+        dataBag.addNodeSpecificData(CLUSTER_PROC.ordinal(), new ClusterFlags(notifyEnabled.get()));
 
-        dataBag.addGridCommonData(CLUSTER_PROC.ordinal(), new ClusterIdAndTag(cluster.id(), cluster.tag()));
-    }
-
-    /**
-     * @return Discovery data.
-     */
-    private Serializable getDiscoveryData() {
-        HashMap<String, Object> map = new HashMap<>(2);
-
-        map.put(ATTR_UPDATE_NOTIFIER_STATUS, notifyEnabled.get());
-
-        return map;
+        dataBag.addGridCommonData(CLUSTER_PROC.ordinal(), (Message)new ClusterIdAndTag(cluster.id(), cluster.tag()));
     }
 
     /** {@inheritDoc} */
     @Override public void onGridDataReceived(GridDiscoveryData data) {
-        Map<UUID, Map<String, Boolean>> nodeSpecData = data.nodeSpecificData();
+        Map<UUID, ClusterFlags> nodeSpecData = data.nodeSpecificData();
 
         if (nodeSpecData != null) {
-            Boolean lstFlag = findLastFlag(nodeSpecData.values());
+            Boolean lstFlag = findLastUpdateNotifierFlag(nodeSpecData.values());
 
             if (lstFlag != null)
                 notifyEnabled.set(lstFlag);
@@ -500,7 +489,7 @@ private Serializable getDiscoveryData() {
         ClusterIdAndTag commonData = data.commonData();
 
         if (commonData != null) {
-            Serializable remoteClusterId = commonData.id();
+            UUID remoteClusterId = commonData.id();
 
             if (remoteClusterId != null) {
                 if (locClusterId != null && !locClusterId.equals(remoteClusterId)) {
@@ -510,7 +499,7 @@ private Serializable getDiscoveryData() {
                         ", local cluster ID: " + locClusterId);
                 }
 
-                locClusterId = (UUID)remoteClusterId;
+                locClusterId = remoteClusterId;
             }
 
             String remoteClusterTag = commonData.tag();
@@ -521,15 +510,15 @@ private Serializable getDiscoveryData() {
     }
 
     /**
-     * @param vals collection to seek through.
+     * @param flags Flags collection to seek through.
      */
-    private Boolean findLastFlag(Collection<Map<String, Boolean>> vals) {
-        for (Map<String, Boolean> map : vals) {
-            if (map != null && map.containsKey(ATTR_UPDATE_NOTIFIER_STATUS))
-                return map.get(ATTR_UPDATE_NOTIFIER_STATUS);
-        }
+    private Boolean findLastUpdateNotifierFlag(Collection<ClusterFlags> flags) {
+        Boolean notifierFlag = null;
 
-        return null;
+        for (ClusterFlags flag : flags)
+            notifierFlag = flag.updateNotifierEnabled;
+
+        return notifierFlag;
     }
 
     /** {@inheritDoc} */