IGNITE-28793 Use MessageSerializer to transfer BaselineStateAndHistoryData

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

@@ -189,12 +189,16 @@
 import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
 import org.apache.ignite.internal.processors.cache.verify.TransactionsHashRecord;
 import org.apache.ignite.internal.processors.cache.version.GridCacheRawVersionedEntry;
+import org.apache.ignite.internal.processors.cluster.BaselineStateAndHistoryData;
+import org.apache.ignite.internal.processors.cluster.BaselineTopologyHistory;
+import org.apache.ignite.internal.processors.cluster.BaselineTopologyHistoryItem;
 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.ClusterIdAndTag;
 import org.apache.ignite.internal.processors.cluster.ClusterMetricsUpdateMessage;
 import org.apache.ignite.internal.processors.cluster.ClusterUpdateNotifierDataBagItem;
+import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
 import org.apache.ignite.internal.processors.cluster.NodeFullMetricsMessage;
 import org.apache.ignite.internal.processors.cluster.NodeMetricsMessage;
 import org.apache.ignite.internal.processors.continuous.ContinuousRoutineStartResultMessage;
@@ -719,6 +723,10 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh) {
         register(ClusterUpdateNotifierDataBagItem.class);
         register(PluginsDataBagItem.class);
         register(EventsDataBagItem.class);
+        register(BaselineStateAndHistoryData.class);
+        register(BaselineTopologyHistory.class);
+        register(BaselineTopologyHistoryItem.class);
+        register(DiscoveryDataClusterState.class);
 
         // [13400 - 13500]: Operation context messages.
         msgIdx = 13400;

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java

@@ -106,7 +106,6 @@
 import org.apache.ignite.internal.processors.cluster.BaselineTopology;
 import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessage;
 import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
-import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
 import org.apache.ignite.internal.processors.cluster.IgniteChangeGlobalStateSupport;
 import org.apache.ignite.internal.processors.metric.GridMetricManager;
 import org.apache.ignite.internal.processors.subscription.GridInternalSubscriptionProcessor;
@@ -1275,11 +1274,6 @@ private ExchangeType onClusterStateChangeRequest(boolean crd) {
 
         GridKernalContext kctx = cctx.kernalContext();
 
-        DiscoveryDataClusterState state = kctx.state().clusterState();
-
-        if (state.transitionError() != null)
-            exchangeLocE = state.transitionError();
-
         if (req.activeChanged()) {
             if (req.state().active()) {
                 if (log.isInfoEnabled()) {

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

@@ -0,0 +1,53 @@
+/*
+ * 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;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+
+/**
+ * Carries cluster state and recent baseline topology history to joining nodes.
+ *
+ * <p>Instances are sent through discovery data collection during node join.
+ * A joining node receives this message, extracts the current {@link DiscoveryDataClusterState},
+ * and replays the {@link BaselineTopologyHistory} items into its local history.</p>
+ */
+public class BaselineStateAndHistoryData implements Message {
+    /** Current cluster state (active/inactive, baseline topology, transition info). */
+    @Order(0)
+    DiscoveryDataClusterState globalState;
+
+    /** Recent baseline topology history items for replay on the joining node. */
+    @Order(1)
+    BaselineTopologyHistory recentHistory;
+
+    /** Default constructor for {@link MessageFactory}. */
+    public BaselineStateAndHistoryData() {
+        // No-op.
+    }
+
+    /**
+     * @param globalState Current cluster state.
+     * @param recentHistory  Recent baseline topology history to transfer.
+     */
+    BaselineStateAndHistoryData(DiscoveryDataClusterState globalState, BaselineTopologyHistory recentHistory) {
+        this.globalState = globalState;
+        this.recentHistory = recentHistory;
+    }
+}

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

@@ -16,32 +16,28 @@
  */
 package org.apache.ignite.internal.processors.cluster;
 
-import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.processors.cache.persistence.metastorage.ReadOnlyMetastorage;
 import org.apache.ignite.internal.processors.cache.persistence.metastorage.ReadWriteMetastorage;
+import org.apache.ignite.plugin.extensions.communication.Message;
 
-/**
- *
- */
-public class BaselineTopologyHistory implements Serializable {
-    /** */
-    private static final long serialVersionUID = 0L;
-
+/** */
+public class BaselineTopologyHistory implements Message {
     /** */
     private static final String METASTORE_BLT_HIST_PREFIX = "bltHist-";
 
     /** */
-    private final Queue<BaselineTopologyHistoryItem> bufferedForStore =
-        new ConcurrentLinkedQueue<>();
+    private final Queue<BaselineTopologyHistoryItem> bufferedForStore = new ConcurrentLinkedQueue<>();
 
     /** */
-    private final List<BaselineTopologyHistoryItem> hist = new ArrayList<>();
+    @Order(0)
+    List<BaselineTopologyHistoryItem> hist = new ArrayList<>();
 
     /** */
     void restoreHistory(ReadOnlyMetastorage metastorage, int lastId) throws IgniteCheckedException {

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

@@ -18,40 +18,39 @@
 
 import java.io.Serializable;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.List;
-import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 
-/**
- *
- */
-public class BaselineTopologyHistoryItem implements Serializable {
+/** */
+public class BaselineTopologyHistoryItem implements Serializable, Message {
     /** */
     private static final long serialVersionUID = 0L;
 
     /** */
-    private final int id;
+    @Order(0)
+    int id;
 
     /** */
-    private final Collection<Object> consIds;
+    @Order(1)
+    List<Long> branchingHistory;
 
-    /** */
-    private final List<Long> branchingHistory;
+    /** Default constructor for {@link MessageFactory}. */
+    public BaselineTopologyHistoryItem() {
+        // No-op.
+    }
 
     /**
      * @param id Id.
-     * @param consIds Consistent IDs.
      * @param branchingHistory Activation history.
      */
-    private BaselineTopologyHistoryItem(int id, Collection<Object> consIds, List<Long> branchingHistory) {
+    private BaselineTopologyHistoryItem(int id, List<Long> branchingHistory) {
         this.id = id;
-        this.consIds = consIds;
         this.branchingHistory = branchingHistory;
     }
 
-    /**
-     * @param blt Baseline Topology.
-     */
+    /** @param blt Baseline Topology. */
     public static BaselineTopologyHistoryItem fromBaseline(BaselineTopology blt) {
         if (blt == null)
             return null;
@@ -60,19 +59,15 @@ public static BaselineTopologyHistoryItem fromBaseline(BaselineTopology blt) {
 
         fullActivationHistory.addAll(blt.branchingHistory());
 
-        return new BaselineTopologyHistoryItem(blt.id(), U.arrayList(blt.consistentIds()), fullActivationHistory);
+        return new BaselineTopologyHistoryItem(blt.id(), fullActivationHistory);
     }
 
-    /**
-     * @return ID.
-     */
+    /** @return ID. */
     public int id() {
         return id;
     }
 
-    /**
-     *
-     */
+    /** */
     public List<Long> branchingHistory() {
         return branchingHistory;
     }