IGNITE-24096 Remove GridDhtPartitionSupplyMessageV2#AVAILABLE_SINCE (#11811)

Author: vladnovoren

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

@@ -93,8 +93,8 @@
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtForceKeysResponse;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionDemandMessage;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionExchangeId;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyErrorMessage;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessage;
-import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessageV2;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleRequest;
@@ -315,7 +315,7 @@ public class GridIoMessageFactory implements MessageFactoryProvider {
         factory.register((short)134, ContinuousRoutineStartResultMessage::new);
         factory.register((short)135, LatchAckMessage::new);
         factory.register((short)157, PartitionUpdateCountersMessage::new);
-        factory.register((short)158, GridDhtPartitionSupplyMessageV2::new);
+        factory.register((short)158, GridDhtPartitionSupplyErrorMessage::new);
         factory.register((short)162, GenerateEncryptionKeyRequest::new);
         factory.register((short)163, GenerateEncryptionKeyResponse::new);
         factory.register((short)167, ServiceDeploymentProcessId::new);

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

@@ -444,15 +444,12 @@ else if (iter.isPartitionMissing(p)) {
             if (grp.shared().kernalContext().isStopping())
                 return;
 
-            // Sending supply messages with error requires new protocol.
-            boolean sendErrMsg = demanderNode.version().compareTo(GridDhtPartitionSupplyMessageV2.AVAILABLE_SINCE) >= 0;
+            boolean skipErrMsg = t instanceof IgniteSpiException;
 
-            if (t instanceof IgniteSpiException) {
+            if (skipErrMsg) {
                 if (log.isDebugEnabled())
                     log.debug("Failed to send message to node (current node is stopping?) ["
                         + supplyRoutineInfo(topicId, nodeId, demandMsg) + ", msg=" + t.getMessage() + ']');
-
-                sendErrMsg = false;
             }
             else
                 U.error(log, "Failed to continue supplying ["
@@ -467,7 +464,7 @@ else if (iter.isPartitionMissing(p)) {
                     + supplyRoutineInfo(topicId, nodeId, demandMsg) + ']', t1);
             }
 
-            if (!sendErrMsg)
+            if (skipErrMsg)
                 return;
 
             boolean fallbackToFullRebalance = X.hasCause(t, IgniteHistoricalIteratorException.class);
@@ -491,7 +488,7 @@ else if (iter.isPartitionMissing(p)) {
                     errMsg = supplyMsg;
                 }
                 else {
-                    errMsg = new GridDhtPartitionSupplyMessageV2(
+                    errMsg = new GridDhtPartitionSupplyErrorMessage(
                         demandMsg.rebalanceId(),
                         grp.groupId(),
                         demandMsg.topologyVersion(),

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

@@ -23,21 +23,17 @@
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
 import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteProductVersion;
 import org.apache.ignite.plugin.extensions.communication.MessageReader;
 import org.apache.ignite.plugin.extensions.communication.MessageWriter;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * Supply message with supplier error transfer support.
  */
-public class GridDhtPartitionSupplyMessageV2 extends GridDhtPartitionSupplyMessage {
+public class GridDhtPartitionSupplyErrorMessage extends GridDhtPartitionSupplyMessage {
     /** */
     private static final long serialVersionUID = 0L;
 
-    /** Available since. */
-    public static final IgniteProductVersion AVAILABLE_SINCE = IgniteProductVersion.fromString("2.7.0");
-
     /** Supplying process error. */
     @GridDirectTransient
     private Throwable err;
@@ -48,7 +44,7 @@ public class GridDhtPartitionSupplyMessageV2 extends GridDhtPartitionSupplyMessa
     /**
      * Default constructor.
      */
-    public GridDhtPartitionSupplyMessageV2() {
+    public GridDhtPartitionSupplyErrorMessage() {
     }
 
     /**
@@ -58,7 +54,7 @@ public GridDhtPartitionSupplyMessageV2() {
      * @param addDepInfo Add dep info.
      * @param err Supply process error.
      */
-    public GridDhtPartitionSupplyMessageV2(
+    public GridDhtPartitionSupplyErrorMessage(
         long rebalanceId,
         int grpId,
         AffinityTopologyVersion topVer,
@@ -133,7 +129,7 @@ public GridDhtPartitionSupplyMessageV2(
 
         }
 
-        return reader.afterMessageRead(GridDhtPartitionSupplyMessageV2.class);
+        return reader.afterMessageRead(GridDhtPartitionSupplyErrorMessage.class);
     }
 
     /** {@inheritDoc} */

modules/core/src/main/resources/META-INF/classnames.properties

@@ -1183,7 +1183,7 @@ org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPar
 org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap
 org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap
 org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessage
-org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessageV2
+org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyErrorMessage
 org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsAbstractMessage
 org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture$2
 org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture$3

modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheNoAffinityExchangeTest.java

@@ -46,7 +46,7 @@
 import org.apache.ignite.internal.TestRecordingCommunicationSpi;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtAffinityAssignmentRequest;
-import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessageV2;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyErrorMessage;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage;
 import org.apache.ignite.internal.util.typedef.G;
 import org.apache.ignite.internal.util.typedef.internal.U;
@@ -290,7 +290,7 @@ public void testNoAffinityChangeOnClientLeftWithMergedExchanges() throws Excepti
 
             TestRecordingCommunicationSpi.spi(ig).blockMessages(new IgniteBiPredicate<ClusterNode, Message>() {
                 @Override public boolean apply(ClusterNode node, Message msg) {
-                    return msg instanceof GridDhtPartitionSupplyMessageV2;
+                    return msg instanceof GridDhtPartitionSupplyErrorMessage;
                 }
             });