IGNITE-28096 Extract TableImplFactory from TableManager (#7898)

Author: dant3

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

@@ -232,6 +232,7 @@
 import org.apache.ignite.internal.table.StreamerReceiverRunner;
 import org.apache.ignite.internal.table.TableTestUtils;
 import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.internal.table.distributed.DefaultMvTableStorageFactory;
 import org.apache.ignite.internal.table.distributed.TableManager;
 import org.apache.ignite.internal.table.distributed.index.IndexMetaStorage;
 import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorService;
@@ -1595,7 +1596,8 @@ private class Node {
                     minTimeCollectorService,
                     systemDistributedConfiguration,
                     metricManager,
-                    TableTestUtils.NOOP_PARTITION_MODIFICATION_COUNTER_FACTORY
+                    TableTestUtils.NOOP_PARTITION_MODIFICATION_COUNTER_FACTORY,
+                    new DefaultMvTableStorageFactory(dataStorageMgr, catalogManager, lowWatermark)
             );
 
             tableManager.setStreamerReceiverRunner(mock(StreamerReceiverRunner.class));

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

@@ -57,8 +57,6 @@
 import org.apache.ignite.internal.catalog.CatalogManagerImpl;
 import org.apache.ignite.internal.catalog.CatalogTestUtils;
 import org.apache.ignite.internal.catalog.compaction.CatalogCompactionRunner;
-import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
-import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
 import org.apache.ignite.internal.catalog.storage.UpdateLogImpl;
 import org.apache.ignite.internal.cluster.management.ClusterIdHolder;
 import org.apache.ignite.internal.cluster.management.ClusterInitializer;
@@ -169,6 +167,8 @@
 import org.apache.ignite.internal.systemview.api.SystemViewManager;
 import org.apache.ignite.internal.table.StreamerReceiverRunner;
 import org.apache.ignite.internal.table.TableTestUtils;
+import org.apache.ignite.internal.table.distributed.DefaultMvTableStorageFactory;
+import org.apache.ignite.internal.table.distributed.MvTableStorageFactory;
 import org.apache.ignite.internal.table.distributed.TableManager;
 import org.apache.ignite.internal.table.distributed.index.IndexMetaStorage;
 import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorService;
@@ -800,30 +800,9 @@ public CompletableFuture<Boolean> invoke(Condition condition, Operation success,
                 minTimeCollectorService,
                 systemDistributedConfiguration,
                 metricManager,
-                TableTestUtils.NOOP_PARTITION_MODIFICATION_COUNTER_FACTORY
-        ) {
-
-            @Override
-            protected MvTableStorage createTableStorage(CatalogTableDescriptor tableDescriptor, CatalogZoneDescriptor zoneDescriptor) {
-                MvTableStorage storage = createSpy(super.createTableStorage(tableDescriptor, zoneDescriptor));
-
-                var partitionStorages = new ConcurrentHashMap<Integer, MvPartitionStorage>();
-
-                doAnswer(invocation -> {
-                    Integer partitionId = invocation.getArgument(0);
-
-                    return partitionStorages.computeIfAbsent(partitionId, id -> {
-                        try {
-                            return (MvPartitionStorage) createSpy(invocation.callRealMethod());
-                        } catch (Throwable e) {
-                            throw new RuntimeException(e);
-                        }
-                    });
-                }).when(storage).getMvPartition(anyInt());
-
-                return storage;
-            }
-        };
+                TableTestUtils.NOOP_PARTITION_MODIFICATION_COUNTER_FACTORY,
+                spyingMvTableStorageFactory(new DefaultMvTableStorageFactory(dataStorageMgr, catalogManager, lowWatermark))
+        );
 
         tableManager.setStreamerReceiverRunner(mock(StreamerReceiverRunner.class));
 
@@ -1050,6 +1029,28 @@ public ReplicaMeta getPrimaryReplica(int zoneId) throws InterruptedException {
         return primaryReplicaFuture.join();
     }
 
+    private static MvTableStorageFactory spyingMvTableStorageFactory(MvTableStorageFactory delegate) {
+        return (tableDesc, zoneDesc) -> {
+            MvTableStorage storage = createSpy(delegate.createMvTableStorage(tableDesc, zoneDesc));
+
+            var partitionStorages = new ConcurrentHashMap<Integer, MvPartitionStorage>();
+
+            doAnswer(invocation -> {
+                Integer partitionId = invocation.getArgument(0);
+
+                return partitionStorages.computeIfAbsent(partitionId, id -> {
+                    try {
+                        return (MvPartitionStorage) createSpy(invocation.callRealMethod());
+                    } catch (Throwable e) {
+                        throw new RuntimeException(e);
+                    }
+                });
+            }).when(storage).getMvPartition(anyInt());
+
+            return storage;
+        };
+    }
+
     @Contract("null -> null")
     private static <T> @Nullable T createSpy(@Nullable T object) {
         if (object == null) {

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

@@ -212,6 +212,7 @@
 import org.apache.ignite.internal.table.TableImpl;
 import org.apache.ignite.internal.table.TableTestUtils;
 import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.internal.table.distributed.DefaultMvTableStorageFactory;
 import org.apache.ignite.internal.table.distributed.TableManager;
 import org.apache.ignite.internal.table.distributed.index.IndexMetaStorage;
 import org.apache.ignite.internal.table.distributed.raft.MinimumRequiredTimeCollectorService;
@@ -818,7 +819,8 @@ public CompletableFuture<Set<String>> dataNodes(HybridTimestamp timestamp, int c
                 minTimeCollectorService,
                 systemDistributedConfiguration,
                 metricManager,
-                TableTestUtils.NOOP_PARTITION_MODIFICATION_COUNTER_FACTORY
+                TableTestUtils.NOOP_PARTITION_MODIFICATION_COUNTER_FACTORY,
+                new DefaultMvTableStorageFactory(dataStorageManager, catalogManager, lowWatermark)
         );
 
         var indexManager = new IndexManager(

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

@@ -268,6 +268,7 @@
 import org.apache.ignite.internal.system.JvmCpuInformationProvider;
 import org.apache.ignite.internal.systemview.SystemViewManagerImpl;
 import org.apache.ignite.internal.systemview.api.SystemViewManager;
+import org.apache.ignite.internal.table.distributed.DefaultMvTableStorageFactory;
 import org.apache.ignite.internal.table.distributed.PartitionModificationCounterFactory;
 import org.apache.ignite.internal.table.distributed.PublicApiThreadingIgniteTables;
 import org.apache.ignite.internal.table.distributed.TableManager;
@@ -393,6 +394,9 @@ public class IgniteImpl implements Ignite {
     /** Replica manager. */
     private final ReplicaManager replicaMgr;
 
+    /** Replica service. */
+    private final ReplicaService replicaSvc;
+
     /** Transactions manager. */
     private final TxManagerImpl txManager;
 
@@ -918,7 +922,7 @@ public class IgniteImpl implements Ignite {
 
         TransactionConfiguration txConfig = clusterConfigRegistry.getConfiguration(TransactionExtensionConfiguration.KEY).transaction();
 
-        ReplicaService replicaSvc = new ReplicaService(
+        replicaSvc = new ReplicaService(
                 messagingServiceReturningToStorageOperationsPool,
                 clockService,
                 threadPoolsManager.partitionOperationsExecutor(),
@@ -1195,7 +1199,8 @@ public class IgniteImpl implements Ignite {
                 minTimeCollectorService,
                 systemDistributedConfiguration,
                 metricManager,
-                partitionModificationCounterFactory
+                partitionModificationCounterFactory,
+                new DefaultMvTableStorageFactory(dataStorageMgr, catalogManager, lowWatermark)
         );
 
         disasterRecoveryManager = new DisasterRecoveryManager(
@@ -1837,6 +1842,11 @@ public DisasterRecoveryManager disasterRecoveryManager() {
         return disasterRecoveryManager;
     }
 
+    @TestOnly
+    public ReplicaService replicaService() {
+        return replicaSvc;
+    }
+
     @TestOnly
     public VaultManager vault() {
         return vaultMgr;

modules/table/src/main/java/org/apache/ignite/internal/table/distributed/DefaultMvTableStorageFactory.java

@@ -0,0 +1,71 @@
+/*
+ * 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.table.distributed;
+
+import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.lowwatermark.LowWatermark;
+import org.apache.ignite.internal.storage.DataStorageManager;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
+import org.apache.ignite.internal.storage.engine.StorageEngine;
+import org.apache.ignite.internal.storage.engine.StorageTableDescriptor;
+import org.apache.ignite.internal.table.distributed.storage.NullStorageEngine;
+
+/**
+ * Default implementation of {@link MvTableStorageFactory} that creates table storages via {@link DataStorageManager}.
+ */
+public final class DefaultMvTableStorageFactory implements MvTableStorageFactory {
+    private final DataStorageManager dataStorageMgr;
+
+    private final CatalogService catalogService;
+
+    private final LowWatermark lowWatermark;
+
+    /**
+     * Constructor.
+     *
+     * @param dataStorageMgr Data storage manager.
+     * @param catalogService Catalog service.
+     * @param lowWatermark Low watermark.
+     */
+    public DefaultMvTableStorageFactory(
+            DataStorageManager dataStorageMgr,
+            CatalogService catalogService,
+            LowWatermark lowWatermark
+    ) {
+        this.dataStorageMgr = dataStorageMgr;
+        this.catalogService = catalogService;
+        this.lowWatermark = lowWatermark;
+    }
+
+    @Override
+    public MvTableStorage createMvTableStorage(CatalogTableDescriptor tableDescriptor, CatalogZoneDescriptor zoneDescriptor) {
+        StorageEngine engine = dataStorageMgr.engineByStorageProfile(tableDescriptor.storageProfile());
+
+        if (engine == null) {
+            // Create a placeholder to allow Table object being created.
+            engine = new NullStorageEngine();
+        }
+
+        return engine.createMvTable(
+                new StorageTableDescriptor(tableDescriptor.id(), zoneDescriptor.partitions(), tableDescriptor.storageProfile()),
+                new CatalogStorageIndexDescriptorSupplier(catalogService, lowWatermark)
+        );
+    }
+}

modules/table/src/main/java/org/apache/ignite/internal/table/distributed/MvTableStorageFactory.java

@@ -0,0 +1,36 @@
+/*
+ * 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.table.distributed;
+
+import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
+
+/**
+ * Factory for creating {@link MvTableStorage} instances.
+ */
+public interface MvTableStorageFactory {
+    /**
+     * Creates a new MV table storage for the given table and zone descriptors.
+     *
+     * @param tableDescriptor Catalog table descriptor.
+     * @param zoneDescriptor Catalog zone descriptor.
+     * @return New MV table storage.
+     */
+    MvTableStorage createMvTableStorage(CatalogTableDescriptor tableDescriptor, CatalogZoneDescriptor zoneDescriptor);
+}

modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StreamerFlushExecutorFactory.java

@@ -0,0 +1,105 @@
+/*
+ * 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.table.distributed;
+
+import static org.apache.ignite.internal.thread.ThreadOperation.STORAGE_WRITE;
+import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
+import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.network.InternalClusterNode;
+import org.apache.ignite.internal.thread.IgniteThreadFactory;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Lazily creates and manages the lifecycle of a single streamer flush {@link ScheduledExecutorService}.
+ *
+ * <p>The executor is created on first call to {@link #get()} and shut down on {@link #stop}.
+ * {@link #beforeStop()} must be called before {@link #stop} to prevent new executor creation during shutdown.
+ */
+public final class StreamerFlushExecutorFactory {
+    private static final IgniteLogger LOG = Loggers.forClass(StreamerFlushExecutorFactory.class);
+
+    private final InternalClusterNode localNode;
+
+    private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+    @Nullable
+    private ScheduledExecutorService executor;
+
+    /**
+     * Constructor.
+     *
+     * @param localNode Local node, used for thread naming.
+     */
+    public StreamerFlushExecutorFactory(InternalClusterNode localNode) {
+        this.localNode = localNode;
+    }
+
+    /**
+     * Returns the executor, creating it lazily if needed.
+     *
+     * @return Streamer flush executor.
+     * @throws IgniteException If the node is stopping.
+     */
+    public synchronized ScheduledExecutorService get() {
+        if (!busyLock.enterBusy()) {
+            throw new IgniteException(NODE_STOPPING_ERR, new NodeStoppingException());
+        }
+
+        try {
+            if (executor == null) {
+                executor = Executors.newSingleThreadScheduledExecutor(
+                        IgniteThreadFactory.create(localNode.name(), "streamer-flush-executor", LOG, STORAGE_WRITE));
+            }
+
+            return executor;
+        } finally {
+            busyLock.leaveBusy();
+        }
+    }
+
+    /**
+     * Blocks the busy lock to prevent new executor creation during shutdown.
+     */
+    public void beforeStop() {
+        busyLock.block();
+    }
+
+    /**
+     * Shuts down the executor if it was created.
+     *
+     * @param timeoutSeconds Timeout in seconds to wait for termination.
+     */
+    public void stop(int timeoutSeconds) {
+        ScheduledExecutorService local;
+
+        synchronized (this) {
+            local = this.executor;
+        }
+
+        shutdownAndAwaitTermination(local, timeoutSeconds, TimeUnit.SECONDS);
+    }
+}