IGNITE-28076 Use watch executor in SchemaSafeTimeTrackerImpl (#7729)

Author: rpuch

modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java

@@ -1505,7 +1505,7 @@ private class Node {
 
             schemaManager = new SchemaManager(registry, catalogManager);
 
-            schemaSafeTimeTracker = new SchemaSafeTimeTrackerImpl(metaStorageManager.clusterTime());
+            schemaSafeTimeTracker = new SchemaSafeTimeTrackerImpl(metaStorageManager.clusterTime(), metaStorageManager.watchExecutor());
             metaStorageManager.registerNotificationEnqueuedListener(schemaSafeTimeTracker);
 
             schemaSyncService = new SchemaSyncServiceImpl(schemaSafeTimeTracker, delayDurationMsSupplier);

modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java

@@ -128,7 +128,7 @@
  *     <li>Providing corresponding Meta storage service proxy interface</li>
  * </ul>
  */
-public class MetaStorageManagerImpl implements MetaStorageManager, MetastorageGroupMaintenance {
+public class MetaStorageManagerImpl implements MetaStorageManager, MetastorageGroupMaintenance, WatchProcessorAccess {
     private static final IgniteLogger LOG = Loggers.forClass(MetaStorageManagerImpl.class);
 
     private final ClusterService clusterService;
@@ -1409,4 +1409,9 @@ CompletableFuture<Void> sendCompactionCommand(long compactionRevision) {
     public void markAsStopping() {
         metaStorageSvcFut.thenAccept(MetaStorageServiceImpl::markAsStopping);
     }
+
+    @Override
+    public Executor watchExecutor() {
+        return storage.watchExecutor();
+    }
 }

modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/WatchProcessorAccess.java

@@ -0,0 +1,30 @@
+/*
+ * 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.metastorage.impl;
+
+import java.util.concurrent.Executor;
+
+/**
+ * Provides access to watch processing required by internal components.
+ */
+public interface WatchProcessorAccess {
+    /**
+     * Returns executor for watch processing.
+     */
+    Executor watchExecutor();
+}

modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/AbstractKeyValueStorage.java

@@ -31,6 +31,7 @@
 import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Predicate;
 import org.apache.ignite.internal.failure.FailureProcessor;
@@ -396,4 +397,9 @@ protected void drainNotifyWatchProcessorEventsBeforeStartingWatches() {
             notifyWatchProcessorEventsBeforeStartingWatches = null;
         }
     }
+
+    @Override
+    public Executor watchExecutor() {
+        return watchProcessor.watchExecutor();
+    }
 }

modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/KeyValueStorage.java

@@ -21,6 +21,7 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 import org.apache.ignite.internal.close.ManuallyCloseable;
 import org.apache.ignite.internal.hlc.HybridTimestamp;
 import org.apache.ignite.internal.metastorage.CommandId;
@@ -558,4 +559,9 @@ boolean invoke(
      * @return Future that's completed when flushing of the data is completed.
      */
     CompletableFuture<Void> flush();
+
+    /**
+     * Returns executor used to execute watches.
+     */
+    Executor watchExecutor();
 }

modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/WatchProcessor.java

@@ -39,6 +39,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadFactory;
@@ -513,4 +514,8 @@ private static boolean isNotIdempotentCacheCommand(Entry entry) {
                 IDEMPOTENT_COMMAND_PREFIX_BYTES, 0, prefixLength
         );
     }
+
+    Executor watchExecutor() {
+        return watchExecutor;
+    }
 }

modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java

@@ -639,7 +639,7 @@ public CompletableFuture<Boolean> invoke(Condition condition, Operation success,
 
         volatileLogStorageManagerCreator = new VolatileLogStorageManagerCreator(name, workDir.resolve("volatile-log-spillout-" + name));
 
-        schemaSafeTimeTracker = new SchemaSafeTimeTrackerImpl(metaStorageManager.clusterTime());
+        schemaSafeTimeTracker = new SchemaSafeTimeTrackerImpl(metaStorageManager.clusterTime(), metaStorageManager.watchExecutor());
         metaStorageManager.registerNotificationEnqueuedListener(schemaSafeTimeTracker);
 
         LongSupplier delayDurationMsSupplier = () -> DELAY_DURATION_MS;

modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java

@@ -602,7 +602,7 @@ public CompletableFuture<Boolean> invoke(Condition condition, List<Operation> su
                 zoneId -> completedFuture(Set.of())
         );
 
-        var schemaSafeTimeTracker = new SchemaSafeTimeTrackerImpl(metaStorageMgr.clusterTime());
+        var schemaSafeTimeTracker = new SchemaSafeTimeTrackerImpl(metaStorageMgr.clusterTime(), metaStorageMgr.watchExecutor());
         metaStorageMgr.registerNotificationEnqueuedListener(schemaSafeTimeTracker);
 
         LongSupplier delayDurationMsSupplier = () -> TestIgnitionManager.DEFAULT_DELAY_DURATION_MS;

modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java

@@ -935,7 +935,7 @@ public class IgniteImpl implements Ignite {
                 volatileLogStorageManagerCreator
         );
 
-        schemaSafeTimeTracker = new SchemaSafeTimeTrackerImpl(metaStorageMgr.clusterTime());
+        schemaSafeTimeTracker = new SchemaSafeTimeTrackerImpl(metaStorageMgr.clusterTime(), metaStorageMgr.watchExecutor());
         metaStorageMgr.registerNotificationEnqueuedListener(schemaSafeTimeTracker);
 
         SchemaSyncService schemaSyncService = new SchemaSyncServiceImpl(schemaSafeTimeTracker, delayDurationMsSupplier);

modules/runner/src/main/java/org/apache/ignite/internal/schema/SchemaSafeTimeTrackerImpl.java

@@ -22,6 +22,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 import org.apache.ignite.internal.catalog.storage.UpdateLogImpl;
 import org.apache.ignite.internal.hlc.HybridTimestamp;
 import org.apache.ignite.internal.lang.NodeStoppingException;
@@ -39,15 +40,18 @@
 public class SchemaSafeTimeTrackerImpl implements SchemaSafeTimeTracker, IgniteComponent, NotificationEnqueuedListener {
     private final ClusterTime clusterTime;
 
+    private final Executor watchExecutor;
+
     private final PendingComparableValuesTracker<HybridTimestamp, Void> schemaSafeTime =
             new PendingComparableValuesTracker<>(HybridTimestamp.MIN_VALUE);
 
     private CompletableFuture<Void> schemaSafeTimeUpdateFuture = nullCompletedFuture();
 
     private final Object futureMutex = new Object();
 
-    public SchemaSafeTimeTrackerImpl(ClusterTime clusterTime) {
+    public SchemaSafeTimeTrackerImpl(ClusterTime clusterTime, Executor watchExecutor) {
         this.clusterTime = clusterTime;
+        this.watchExecutor = watchExecutor;
     }
 
     @Override
@@ -74,16 +78,16 @@ public void onEnqueued(CompletableFuture<Void> newNotificationFuture, List<Entry
                 // The update touches the Catalog (i.e. schemas), so we must chain with the core notification future
                 // as Catalog listeners will be included in it (because we need to wait for those listeners to finish execution
                 // before updating the schema safe time).
-                newSchemaSafeTimeUpdateFuture = schemaSafeTimeUpdateFuture.thenCompose(unused -> newNotificationFuture);
+                newSchemaSafeTimeUpdateFuture = schemaSafeTimeUpdateFuture.thenComposeAsync(unused -> newNotificationFuture, watchExecutor);
             } else {
                 // The update does not concern the Catalog (schemas), so we can update schema safe time as soon as previous updates to it
                 // get applied.
                 newSchemaSafeTimeUpdateFuture = schemaSafeTimeUpdateFuture;
             }
 
-            newSchemaSafeTimeUpdateFuture = newSchemaSafeTimeUpdateFuture.thenRun(() -> {
+            newSchemaSafeTimeUpdateFuture = newSchemaSafeTimeUpdateFuture.thenRunAsync(() -> {
                 schemaSafeTime.update(timestamp, null);
-            });
+            }, watchExecutor);
 
             schemaSafeTimeUpdateFuture = newSchemaSafeTimeUpdateFuture;
         }