IGNITE-28477 Merge StartRoutineDiscoveryMessageV2 into StartRoutineDiscoveryMessage (#13100)

Author: chesnokoff

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

@@ -189,7 +189,6 @@
 import org.apache.ignite.internal.processors.continuous.StartRequestData;
 import org.apache.ignite.internal.processors.continuous.StartRoutineAckDiscoveryMessage;
 import org.apache.ignite.internal.processors.continuous.StartRoutineDiscoveryMessage;
-import org.apache.ignite.internal.processors.continuous.StartRoutineDiscoveryMessageV2;
 import org.apache.ignite.internal.processors.continuous.StopRoutineAckDiscoveryMessage;
 import org.apache.ignite.internal.processors.continuous.StopRoutineDiscoveryMessage;
 import org.apache.ignite.internal.processors.datastreamer.DataStreamerEntry;
@@ -519,9 +518,8 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchema(LatchAckMessage.class);
         withSchema(AtomicApplicationAttributesAwareRequest.class);
         withNoSchema(StartRequestData.class);
-        withNoSchema(StartRoutineDiscoveryMessage.class);
         withNoSchema(StartRoutineAckDiscoveryMessage.class);
-        withNoSchema(StartRoutineDiscoveryMessageV2.class);
+        withNoSchema(StartRoutineDiscoveryMessage.class);
         withNoSchema(StoredCacheData.class);
 
         // [10600-10800]: Affinity & partition maps.

modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java

@@ -72,6 +72,7 @@
 import org.apache.ignite.internal.processors.cache.GridCacheProcessor;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionPartialCountersMap;
 import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler;
+import org.apache.ignite.internal.processors.continuous.StartRoutineDiscoveryMessage.Mode;
 import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
 import org.apache.ignite.internal.systemview.ContinuousQueryViewWalker;
 import org.apache.ignite.internal.thread.OomExceptionHandler;
@@ -211,26 +212,13 @@ public GridContinuousProcessor(GridKernalContext ctx) {
                 @Override public void onCustomEvent(AffinityTopologyVersion topVer,
                     ClusterNode snd,
                     StartRoutineDiscoveryMessage msg) {
-                    assert !immutableDiscoCustomMsg;
-
                     if (ctx.isStopping())
                         return;
 
-                    processStartRequestMutable(snd, msg);
-                }
-            });
-
-        ctx.discovery().setCustomEventListener(StartRoutineDiscoveryMessageV2.class,
-            new CustomEventListener<StartRoutineDiscoveryMessageV2>() {
-                @Override public void onCustomEvent(AffinityTopologyVersion topVer,
-                    ClusterNode snd,
-                    StartRoutineDiscoveryMessageV2 msg) {
-                    assert immutableDiscoCustomMsg;
-
-                    if (ctx.isStopping())
-                        return;
-
-                    processStartRequestImmutable(topVer, snd, msg);
+                    if (immutableDiscoCustomMsg)
+                        processStartRequestImmutable(topVer, snd, msg);
+                    else
+                        processStartRequestMutable(snd, msg);
                 }
             });
 
@@ -991,17 +979,15 @@ private AbstractContinuousMessage createStartMessage(UUID routineId,
         reqData.prepareMarshal(ctx);
 
         if (!immutableDiscoCustomMsg) {
-            StartRoutineDiscoveryMessage msg = new StartRoutineDiscoveryMessage(
-                    routineId,
-                    reqData);
+            StartRoutineDiscoveryMessage msg = new StartRoutineDiscoveryMessage(routineId, reqData, Mode.MUTABLE);
 
             if (hnd.updateCounters() != null)
                 msg.addUpdateCounters(ctx.localNodeId(), hnd.updateCounters());
 
             return msg;
         }
         else
-            return new StartRoutineDiscoveryMessageV2(routineId, reqData);
+            return new StartRoutineDiscoveryMessage(routineId, reqData, Mode.IMMUTABLE);
     }
 
     /**
@@ -1467,7 +1453,7 @@ private void processRoutineStartResultMessage(UUID sndId, ContinuousRoutineStart
      */
     private void processStartRequestImmutable(final AffinityTopologyVersion topVer,
         final ClusterNode snd,
-        final StartRoutineDiscoveryMessageV2 msg) {
+        final StartRoutineDiscoveryMessage msg) {
         StartRequestData reqData = msg.startRequestData();
 
         ContinuousRoutineInfo routineInfo = new ContinuousRoutineInfo(snd.id(),

modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRoutineDiscoveryMessage.java

@@ -26,15 +26,26 @@
 import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
 import org.apache.ignite.internal.util.typedef.internal.S;
 
+import static org.apache.ignite.internal.processors.continuous.StartRoutineDiscoveryMessage.Mode.MUTABLE;
+
 /**
  * Discovery message used for Continuous Query registration.
  */
 public class StartRoutineDiscoveryMessage extends AbstractContinuousMessage {
+    /** Discovery message mode. */
+    enum Mode {
+        /** Mutable discovery mode. */
+        MUTABLE,
+
+        /** Immutable discovery mode. */
+        IMMUTABLE
+    }
+
     /** */
     @Order(0)
     StartRequestData startReqData;
 
-    /** */
+    /** Errors collected by mutable discovery. */
     @Order(1)
     Map<UUID, ErrorMessage> errs = new HashMap<>();
 
@@ -46,14 +57,20 @@ public class StartRoutineDiscoveryMessage extends AbstractContinuousMessage {
     @Order(3)
     Map<UUID, Map<Integer, Long>> updateCntrsPerNode;
 
+    /** Discovery message mode. */
+    @Order(4)
+    Mode mode;
+
     /**
      * @param routineId Routine id.
      * @param startReqData Start request data.
+     * @param mode Discovery message mode.
      */
-    public StartRoutineDiscoveryMessage(UUID routineId, StartRequestData startReqData) {
+    StartRoutineDiscoveryMessage(UUID routineId, StartRequestData startReqData, Mode mode) {
         super(routineId);
 
         this.startReqData = startReqData;
+        this.mode = mode;
     }
 
     /** */
@@ -110,11 +127,14 @@ public void addUpdateCounters(UUID nodeId, Map<Integer, Long> cntrs) {
 
     /** {@inheritDoc} */
     @Override public boolean isMutable() {
-        return true;
+        return mode == MUTABLE;
     }
 
     /** {@inheritDoc} */
     @Override public DiscoveryCustomMessage ackMessage() {
+        if (!isMutable())
+            return null;
+
         return new StartRoutineAckDiscoveryMessage(routineId, errs, updateCntrs, updateCntrsPerNode);
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRoutineDiscoveryMessageV2.java

@@ -1518,7 +1518,6 @@ org.apache.ignite.internal.processors.continuous.GridContinuousProcessor$LocalRo
 org.apache.ignite.internal.processors.continuous.StartRequestData
 org.apache.ignite.internal.processors.continuous.StartRoutineAckDiscoveryMessage
 org.apache.ignite.internal.processors.continuous.StartRoutineDiscoveryMessage
-org.apache.ignite.internal.processors.continuous.StartRoutineDiscoveryMessageV2
 org.apache.ignite.internal.processors.continuous.StopRoutineAckDiscoveryMessage
 org.apache.ignite.internal.processors.continuous.StopRoutineDiscoveryMessage
 org.apache.ignite.internal.processors.datastreamer.DataStreamProcessor$3

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

@@ -40,7 +40,6 @@
 import org.apache.ignite.cluster.ClusterGroup;
 import org.apache.ignite.internal.DiscoverySpiTestListener;
 import org.apache.ignite.internal.processors.continuous.StartRoutineDiscoveryMessage;
-import org.apache.ignite.internal.processors.continuous.StartRoutineDiscoveryMessageV2;
 import org.apache.ignite.internal.processors.continuous.StopRoutineDiscoveryMessage;
 import org.apache.ignite.internal.util.GridConcurrentHashSet;
 import org.apache.ignite.internal.util.typedef.P2;
@@ -1051,7 +1050,7 @@ public void testAsyncOld() throws Exception {
             }
         }, IllegalStateException.class, null);
 
-        lsnr.blockCustomEvent(StartRoutineDiscoveryMessage.class, StartRoutineDiscoveryMessageV2.class);
+        lsnr.blockCustomEvent(StartRoutineDiscoveryMessage.class);
 
         final String topic = "topic";
 
@@ -1149,7 +1148,7 @@ public void testAsync() throws Exception {
 
         discoSpi.setInternalListener(lsnr);
 
-        lsnr.blockCustomEvent(StartRoutineDiscoveryMessage.class, StartRoutineDiscoveryMessageV2.class);
+        lsnr.blockCustomEvent(StartRoutineDiscoveryMessage.class);
 
         final String topic = "topic";