IGNITE-17286 Added missed busy locks to get rid of resources leaking during table... main
authorMirza Aliev <alievmirza@gmail.com>
Fri, 12 Aug 2022 13:23:37 +0000 (16:23 +0300)
committerSlava Koptilin <slava.koptilin@gmail.com>
Fri, 12 Aug 2022 13:23:37 +0000 (16:23 +0300)
Signed-off-by: Slava Koptilin <slava.koptilin@gmail.com>
modules/core/src/main/java/org/apache/ignite/internal/causality/VersionedValue.java
modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java
modules/runner/src/integrationTest/java/org/apache/ignite/internal/AbstractClusterIntegrationTest.java
modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java
modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java

index 1eb90e1528a2e30571fb34608dc501d64a775d23..92a3d2ad6c3b031055ab8646fe33ec4f8cb1f4d7 100644 (file)
@@ -578,7 +578,7 @@ public class VersionedValue<T> {
 
         try {
             for (Long token : history.keySet()) {
-                if (token != lastToken && causalityToken - token >= historySize) {
+                if (!token.equals(lastToken) && causalityToken - token >= historySize) {
                     history.remove(token);
                 }
             }
index 77478a8b692ebed35b85af11f87c8b19a3d26a37..e19d734eba75f84db8fc4de4eda9bbfeb21fe058 100644 (file)
@@ -17,6 +17,8 @@
 
 package org.apache.ignite.internal.util;
 
+import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
+
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.file.AtomicMoveNotSupportedException;
@@ -44,11 +46,14 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 import java.util.stream.Stream;
 import org.apache.ignite.internal.logger.IgniteLogger;
 import org.apache.ignite.internal.util.worker.IgniteWorker;
+import org.apache.ignite.lang.IgniteInternalException;
 import org.apache.ignite.lang.IgniteStringBuilder;
 import org.apache.ignite.lang.IgniteStringFormatter;
+import org.apache.ignite.lang.NodeStoppingException;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -787,4 +792,40 @@ public class IgniteUtils {
             }
         }
     }
+
+    /**
+     * Method that runs the provided {@code fn} in {@code busyLock}.
+     *
+     * @param busyLock Component's busy lock
+     * @param fn Function to run
+     * @param <T> Type of returned value from {@code fn}
+     * @return Result of the provided function
+     */
+    public static <T> T inBusyLock(IgniteSpinBusyLock busyLock, Supplier<T> fn) {
+        if (!busyLock.enterBusy()) {
+            throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
+        }
+        try {
+            return fn.get();
+        } finally {
+            busyLock.leaveBusy();
+        }
+    }
+
+    /**
+     * Method that runs the provided {@code fn} in {@code busyLock}.
+     *
+     * @param busyLock Component's busy lock
+     * @param fn Runnable to run
+     */
+    public static void inBusyLock(IgniteSpinBusyLock busyLock, Runnable fn) {
+        if (!busyLock.enterBusy()) {
+            throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
+        }
+        try {
+            fn.run();
+        } finally {
+            busyLock.leaveBusy();
+        }
+    }
 }
index 8dfe45efcea4f937ed17a38ff7b484ba244423d5..760b2f776c12532988beb6c0c5fa26bf9077a42e 100644 (file)
@@ -254,7 +254,8 @@ public class JraftServerImpl implements RaftServer {
     /** {@inheritDoc} */
     @Override
     public void stop() throws Exception {
-        assert groups.isEmpty() : IgniteStringFormatter.format("Raft groups are still running {}", groups.keySet());
+        assert groups.isEmpty() : IgniteStringFormatter.format("Raft groups {} are still running on the node {}", groups.keySet(),
+                service.topologyService().localMember().name());
 
         rpcServer.shutdown();
 
index 68549f71ee5ed6240e279450d961274cdec6d3fb..e43f683f1a0cda39cccf7f052ddbfdc6b6f1cca4 100644 (file)
@@ -91,7 +91,7 @@ public abstract class AbstractClusterIntegrationTest extends BaseIgniteAbstractT
                 })
                 .collect(toList());
 
-        String metaStorageNodeName = testNodeName(testInfo, 0);
+        String metaStorageNodeName = testNodeName(testInfo, nodes() - 1);
 
         IgnitionManager.init(metaStorageNodeName, List.of(metaStorageNodeName), "cluster");
 
index b03c77eb2253b78495603338d79a4eb940440e11..16a25c5a29a246ad006100f92fe05d6f0aca474d 100644 (file)
@@ -20,6 +20,8 @@ package org.apache.ignite.internal.schema;
 import static java.util.concurrent.CompletableFuture.completedFuture;
 import static java.util.concurrent.CompletableFuture.failedFuture;
 import static org.apache.ignite.internal.configuration.util.ConfigurationUtil.getByInternalId;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -107,22 +109,30 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
      * @return A future.
      */
     private CompletableFuture<?> onSchemaCreate(ConfigurationNotificationEvent<SchemaView> schemasCtx) {
-        long causalityToken = schemasCtx.storageRevision();
+        if (!busyLock.enterBusy()) {
+            return failedFuture(new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()));
+        }
+
+        try {
+            long causalityToken = schemasCtx.storageRevision();
 
-        ExtendedTableConfiguration tblCfg = schemasCtx.config(ExtendedTableConfiguration.class);
+            ExtendedTableConfiguration tblCfg = schemasCtx.config(ExtendedTableConfiguration.class);
 
-        UUID tblId = tblCfg.id().value();
+            UUID tblId = tblCfg.id().value();
 
-        String tableName = tblCfg.name().value();
+            String tableName = tblCfg.name().value();
 
-        SchemaDescriptor schemaDescriptor = SchemaSerializerImpl.INSTANCE.deserialize((schemasCtx.newValue().schema()));
+            SchemaDescriptor schemaDescriptor = SchemaSerializerImpl.INSTANCE.deserialize((schemasCtx.newValue().schema()));
 
-        CompletableFuture<?> createSchemaFut = createSchema(causalityToken, tblId, tableName, schemaDescriptor);
+            CompletableFuture<?> createSchemaFut = createSchema(causalityToken, tblId, tableName, schemaDescriptor);
 
-        registriesVv.get(causalityToken)
-                .thenRun(() -> fireEvent(SchemaEvent.CREATE, new SchemaEventParameters(causalityToken, tblId, schemaDescriptor)));
+            registriesVv.get(causalityToken).thenRun(() -> inBusyLock(busyLock,
+                    () -> fireEvent(SchemaEvent.CREATE, new SchemaEventParameters(causalityToken, tblId, schemaDescriptor))));
 
-        return createSchemaFut;
+            return createSchemaFut;
+        } finally {
+            busyLock.leaveBusy();
+        }
     }
 
     /**
@@ -161,27 +171,29 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
             String tableName,
             SchemaDescriptor schemaDescriptor
     ) {
-        return registriesVv.update(causalityToken, (registries, e) -> {
+        return registriesVv.update(causalityToken, (registries, e) -> inBusyLock(busyLock, () -> {
             if (e != null) {
                 return failedFuture(new IgniteInternalException(IgniteStringFormatter.format(
-                    "Cannot create a schema for the table [tblId={}, ver={}]", tableId, schemaDescriptor.version()), e)
+                        "Cannot create a schema for the table [tblId={}, ver={}]", tableId, schemaDescriptor.version()), e)
                 );
             }
 
-            SchemaRegistryImpl reg = registries.get(tableId);
+            Map<UUID, SchemaRegistryImpl> regs = registries;
+
+            SchemaRegistryImpl reg = regs.get(tableId);
 
             if (reg == null) {
-                registries = new HashMap<>(registries);
+                regs = new HashMap<>(registries);
 
                 SchemaRegistryImpl registry = createSchemaRegistry(tableId, tableName, schemaDescriptor);
 
-                registries.put(tableId, registry);
+                regs.put(tableId, registry);
             } else {
                 reg.onSchemaRegistered(schemaDescriptor);
             }
 
-            return completedFuture(registries);
-        });
+            return completedFuture(regs);
+        }));
     }
 
     /**
@@ -195,7 +207,7 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
     private SchemaRegistryImpl createSchemaRegistry(UUID tableId, String tableName, SchemaDescriptor initialSchema) {
         return new SchemaRegistryImpl(ver -> {
             if (!busyLock.enterBusy()) {
-                throw new IgniteException(new NodeStoppingException());
+                throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
             }
 
             try {
@@ -205,7 +217,7 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
             }
         }, () -> {
             if (!busyLock.enterBusy()) {
-                throw new IgniteException(new NodeStoppingException());
+                throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
             }
 
             try {
@@ -336,11 +348,12 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
      */
     public CompletableFuture<SchemaRegistry> schemaRegistry(long causalityToken, @Nullable UUID tableId) {
         if (!busyLock.enterBusy()) {
-            throw new IgniteException(new NodeStoppingException());
+            throw new IgniteException(NODE_STOPPING_ERR, new NodeStoppingException());
         }
 
         try {
-            return registriesVv.get(causalityToken).thenApply(regs -> tableId == null ? null : regs.get(tableId));
+            return registriesVv.get(causalityToken)
+                    .thenApply(regs -> inBusyLock(busyLock, () -> tableId == null ? null : regs.get(tableId)));
         } finally {
             busyLock.leaveBusy();
         }
@@ -363,20 +376,18 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
      * @param tableId Table id.
      */
     public CompletableFuture<?> dropRegistry(long causalityToken, UUID tableId) {
-        return registriesVv.update(causalityToken, (registries, e) -> {
+        return registriesVv.update(causalityToken, (registries, e) -> inBusyLock(busyLock, () -> {
             if (e != null) {
                 return failedFuture(new IgniteInternalException(
-                        IgniteStringFormatter.format("Cannot remove a schema registry for the table [tblId={}]", tableId), e
-                    )
-                );
+                        IgniteStringFormatter.format("Cannot remove a schema registry for the table [tblId={}]", tableId), e));
             }
 
-            registries = new HashMap<>(registries);
+            Map<UUID, SchemaRegistryImpl> regs = new HashMap<>(registries);
 
-            registries.remove(tableId);
+            regs.remove(tableId);
 
-            return completedFuture(registries);
-        });
+            return completedFuture(regs);
+        }));
     }
 
     /** {@inheritDoc} */
index 1e81423c7c6e330263075d15e8a5973e37dd775d..028ae0cc6b99c067481df0d3e23dea97da0e1d37 100644 (file)
@@ -162,7 +162,7 @@ public class SqlQueryProcessor implements QueryProcessor {
                 msgSrvc
         ));
 
-        SqlSchemaManagerImpl sqlSchemaManager = new SqlSchemaManagerImpl(tableManager, schemaManager, registry);
+        SqlSchemaManagerImpl sqlSchemaManager = new SqlSchemaManagerImpl(tableManager, schemaManager, registry, busyLock);
 
         sqlSchemaManager.registerListener(prepareSvc);
 
index 6e8d9395a14aa937a91d3b4c64ca5650fd9fcf27..e0698933e1046a71cc1d2c69b3f171da49a56b4f 100644 (file)
@@ -19,6 +19,8 @@ package org.apache.ignite.internal.sql.engine.schema;
 
 import static java.util.concurrent.CompletableFuture.completedFuture;
 import static java.util.concurrent.CompletableFuture.failedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
 
 import java.util.Comparator;
 import java.util.HashMap;
@@ -42,6 +44,7 @@ import org.apache.ignite.internal.schema.SchemaManager;
 import org.apache.ignite.internal.schema.SchemaRegistry;
 import org.apache.ignite.internal.table.TableImpl;
 import org.apache.ignite.internal.table.distributed.TableManager;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
 import org.apache.ignite.lang.IgniteInternalException;
 import org.apache.ignite.lang.IgniteStringFormatter;
 import org.apache.ignite.lang.NodeStoppingException;
@@ -66,6 +69,9 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
 
     private final Set<SchemaUpdateListener> listeners = new CopyOnWriteArraySet<>();
 
+    /** Busy lock for stop synchronisation. */
+    private final IgniteSpinBusyLock busyLock;
+
     /**
      * Constructor.
      * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
@@ -73,12 +79,14 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
     public SqlSchemaManagerImpl(
             TableManager tableManager,
             SchemaManager schemaManager,
-            Consumer<Function<Long, CompletableFuture<?>>> registry
+            Consumer<Function<Long, CompletableFuture<?>>> registry,
+            IgniteSpinBusyLock busyLock
     ) {
         this.tableManager = tableManager;
         this.schemaManager = schemaManager;
         schemasVv = new VersionedValue<>(registry, HashMap::new);
         tablesVv = new VersionedValue<>(registry, HashMap::new);
+        this.busyLock = busyLock;
 
         calciteSchemaVv = new VersionedValue<>(null, () -> {
             SchemaPlus newCalciteSchema = Frameworks.createRootSchema(false);
@@ -87,20 +95,29 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
         });
 
         schemasVv.whenComplete((token, stringIgniteSchemaMap, throwable) -> {
-            if (throwable != null) {
-                calciteSchemaVv.completeExceptionally(
-                        token,
-                        new IgniteInternalException("Couldn't evaluate sql schemas for causality token: " + token, throwable)
-                );
+            if (!busyLock.enterBusy()) {
+                calciteSchemaVv.completeExceptionally(token, new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()));
 
                 return;
             }
+            try {
+                if (throwable != null) {
+                    calciteSchemaVv.completeExceptionally(
+                            token,
+                            new IgniteInternalException("Couldn't evaluate sql schemas for causality token: " + token, throwable)
+                    );
+
+                    return;
+                }
 
-            SchemaPlus newCalciteSchema = rebuild(stringIgniteSchemaMap);
+                SchemaPlus newCalciteSchema = rebuild(stringIgniteSchemaMap);
 
-            listeners.forEach(SchemaUpdateListener::onSchemaUpdated);
+                listeners.forEach(SchemaUpdateListener::onSchemaUpdated);
 
-            calciteSchemaVv.complete(token, newCalciteSchema);
+                calciteSchemaVv.complete(token, newCalciteSchema);
+            } finally {
+                busyLock.leaveBusy();
+            }
         });
     }
 
@@ -116,27 +133,34 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
     @Override
     @NotNull
     public IgniteTable tableById(UUID id, int ver) {
-        IgniteTable table = tablesVv.latest().get(id);
-
-        // there is a chance that someone tries to resolve table before
-        // the distributed event of that table creation has been processed
-        // by TableManager, so we need to get in sync with the TableManager
-        if (table == null || ver > table.version()) {
-            table = awaitLatestTableSchema(id);
+        if (!busyLock.enterBusy()) {
+            throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
         }
+        try {
+            IgniteTable table = tablesVv.latest().get(id);
 
-        if (table == null) {
-            throw new IgniteInternalException(
-                IgniteStringFormatter.format("Table not found [tableId={}]", id));
-        }
+            // there is a chance that someone tries to resolve table before
+            // the distributed event of that table creation has been processed
+            // by TableManager, so we need to get in sync with the TableManager
+            if (table == null || ver > table.version()) {
+                table = awaitLatestTableSchema(id);
+            }
 
-        if (table.version() < ver) {
-            throw new IgniteInternalException(
-                    IgniteStringFormatter.format("Table version not found [tableId={}, requiredVer={}, latestKnownVer={}]",
-                            id, ver, table.version()));
-        }
+            if (table == null) {
+                throw new IgniteInternalException(
+                        IgniteStringFormatter.format("Table not found [tableId={}]", id));
+            }
 
-        return table;
+            if (table.version() < ver) {
+                throw new IgniteInternalException(
+                        IgniteStringFormatter.format("Table version not found [tableId={}, requiredVer={}, latestKnownVer={}]",
+                                id, ver, table.version()));
+            }
+
+            return table;
+        } finally {
+            busyLock.leaveBusy();
+        }
     }
 
     public void registerListener(SchemaUpdateListener listener) {
@@ -155,56 +179,10 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
 
             return convert(table);
         } catch (NodeStoppingException e) {
-            throw new IgniteInternalException(e);
+            throw new IgniteInternalException(NODE_STOPPING_ERR, e);
         }
     }
 
-    /**
-     * Schema creation handler.
-     *
-     * @param schemaName Schema name.
-     * @param causalityToken Causality token.
-     */
-    public synchronized void onSchemaCreated(String schemaName, long causalityToken) {
-        schemasVv.update(
-                causalityToken,
-                (schemas, e) -> {
-                    if (e != null) {
-                        return failedFuture(e);
-                    }
-
-                    Map<String, IgniteSchema> res = new HashMap<>(schemas);
-
-                    res.putIfAbsent(schemaName, new IgniteSchema(schemaName));
-
-                    return completedFuture(res);
-                }
-        );
-    }
-
-    /**
-     * Schema drop handler.
-     *
-     * @param schemaName Schema name.
-     * @param causalityToken Causality token.
-     */
-    public synchronized void onSchemaDropped(String schemaName, long causalityToken) {
-        schemasVv.update(
-                causalityToken,
-                (schemas, e) -> {
-                    if (e != null) {
-                        return failedFuture(e);
-                    }
-
-                    Map<String, IgniteSchema> res = new HashMap<>(schemas);
-
-                    res.remove(schemaName);
-
-                    return completedFuture(res);
-                }
-        );
-    }
-
     /**
      * OnSqlTypeCreated.
      * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
@@ -214,50 +192,45 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
             TableImpl table,
             long causalityToken
     ) {
-        schemasVv.update(
-                causalityToken,
-                (schemas, e) -> {
-                    if (e != null) {
-                        return failedFuture(e);
-                    }
+        if (!busyLock.enterBusy()) {
+            return failedFuture(new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()));
+        }
+        try {
+            schemasVv.update(causalityToken, (schemas, e) -> inBusyLock(busyLock, () -> {
+                if (e != null) {
+                    return failedFuture(e);
+                }
+
+                Map<String, IgniteSchema> res = new HashMap<>(schemas);
 
-                    Map<String, IgniteSchema> res = new HashMap<>(schemas);
+                IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
 
-                    IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
+                CompletableFuture<IgniteTableImpl> igniteTableFuture = convert(causalityToken, table);
 
-                    CompletableFuture<IgniteTableImpl> igniteTableFuture = convert(causalityToken, table);
+                return tablesVv.update(causalityToken, (tables, ex) -> inBusyLock(busyLock, () -> {
+                    if (ex != null) {
+                        return failedFuture(ex);
+                    }
 
-                    return tablesVv
-                            .update(
-                                    causalityToken,
-                                    (tables, ex) -> {
-                                        if (ex != null) {
-                                            return failedFuture(ex);
-                                        }
+                    Map<UUID, IgniteTable> resTbls = new HashMap<>(tables);
 
-                                        Map<UUID, IgniteTable> resTbls = new HashMap<>(tables);
+                    return igniteTableFuture.thenApply(igniteTable -> inBusyLock(busyLock, () -> {
+                        resTbls.put(igniteTable.id(), igniteTable);
 
-                                        return igniteTableFuture
-                                            .thenApply(igniteTable -> {
-                                                resTbls.put(igniteTable.id(), igniteTable);
+                        return resTbls;
+                    }));
+                })).thenCombine(igniteTableFuture, (v, igniteTable) -> inBusyLock(busyLock, () -> {
+                    schema.addTable(removeSchema(schemaName, table.name()), igniteTable);
 
-                                                return resTbls;
-                                            });
-                                    }
-                            )
-                            .thenCombine(
-                                igniteTableFuture,
-                                (v, igniteTable) -> {
-                                    schema.addTable(removeSchema(schemaName, table.name()), igniteTable);
+                    return null;
+                })).thenCompose(v -> inBusyLock(busyLock, () -> completedFuture(res)));
 
-                                    return null;
-                                }
-                            )
-                            .thenCompose(v -> completedFuture(res));
-                }
-        );
+            }));
 
-        return calciteSchemaVv.get(causalityToken);
+            return calciteSchemaVv.get(causalityToken);
+        } finally {
+            busyLock.leaveBusy();
+        }
     }
 
     /**
@@ -281,45 +254,46 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
             String tableName,
             long causalityToken
     ) {
-        schemasVv.update(causalityToken,
-                (schemas, e) -> {
-                    if (e != null) {
-                        return failedFuture(e);
-                    }
-
-                    Map<String, IgniteSchema> res = new HashMap<>(schemas);
+        if (!busyLock.enterBusy()) {
+            return failedFuture(new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()));
+        }
+        try {
+            schemasVv.update(causalityToken, (schemas, e) -> inBusyLock(busyLock, () -> {
+                if (e != null) {
+                    return failedFuture(e);
+                }
 
-                    IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
+                Map<String, IgniteSchema> res = new HashMap<>(schemas);
 
-                    String calciteTableName = removeSchema(schemaName, tableName);
+                IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
 
-                    InternalIgniteTable table = (InternalIgniteTable) schema.getTable(calciteTableName);
+                String calciteTableName = removeSchema(schemaName, tableName);
 
-                    if (table != null) {
-                        schema.removeTable(calciteTableName);
+                InternalIgniteTable table = (InternalIgniteTable) schema.getTable(calciteTableName);
 
-                        return tablesVv
-                                .update(causalityToken,
-                                        (tables, ex) -> {
-                                            if (ex != null) {
-                                                return failedFuture(ex);
-                                            }
+                if (table != null) {
+                    schema.removeTable(calciteTableName);
 
-                                            Map<UUID, IgniteTable> resTbls = new HashMap<>(tables);
+                    return tablesVv.update(causalityToken, (tables, ex) -> inBusyLock(busyLock, () -> {
+                        if (ex != null) {
+                            return failedFuture(ex);
+                        }
 
-                                            resTbls.remove(table.id());
+                        Map<UUID, IgniteTable> resTbls = new HashMap<>(tables);
 
-                                            return completedFuture(resTbls);
-                                        }
-                                )
-                                .thenCompose(tables -> completedFuture(res));
-                    }
+                        resTbls.remove(table.id());
 
-                    return completedFuture(res);
+                        return completedFuture(resTbls);
+                    })).thenCompose(tables -> inBusyLock(busyLock, () -> completedFuture(res)));
                 }
-        );
 
-        return calciteSchemaVv.get(causalityToken);
+                return completedFuture(res);
+            }));
+
+            return calciteSchemaVv.get(causalityToken);
+        } finally {
+            busyLock.leaveBusy();
+        }
     }
 
     /**
@@ -339,7 +313,7 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
 
     private CompletableFuture<IgniteTableImpl> convert(long causalityToken, TableImpl table) {
         return schemaManager.schemaRegistry(causalityToken, table.tableId())
-            .thenApply(schemaRegistry -> convert(table, schemaRegistry));
+            .thenApply(schemaRegistry -> inBusyLock(busyLock, () -> convert(table, schemaRegistry)));
     }
 
     private IgniteTableImpl convert(TableImpl table) {
index 3496b2dff42d881119b198bbefc5f4d4a09a16a3..37ab1cc8e0ab401c6e18e7d7af7531c32677f7ce 100644 (file)
@@ -50,6 +50,7 @@ import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManagerImpl;
 import org.apache.ignite.internal.table.InternalTable;
 import org.apache.ignite.internal.table.TableImpl;
 import org.apache.ignite.internal.table.distributed.TableManager;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
 import org.apache.ignite.lang.IgniteInternalException;
 import org.apache.ignite.lang.NodeStoppingException;
 import org.junit.jupiter.api.BeforeEach;
@@ -90,6 +91,9 @@ public class SqlSchemaManagerTest {
 
     private TestRevisionRegister testRevisionRegister;
 
+    /** Busy lock for stop synchronisation. */
+    private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
     @BeforeEach
     public void setup() throws NodeStoppingException {
         Mockito.reset(tableManager);
@@ -99,7 +103,8 @@ public class SqlSchemaManagerTest {
         sqlSchemaManager = new SqlSchemaManagerImpl(
                 tableManager,
                 schemaManager,
-                testRevisionRegister
+                testRevisionRegister,
+                busyLock
         );
 
         testRevisionRegister.moveForward();
index ba0e2926cf5c240369d3ed8fffc241c75e4301ba..2a3ccecec1b44e1163f86017d9808c94f4b68dc6 100644 (file)
@@ -23,6 +23,7 @@ import static java.util.concurrent.CompletableFuture.completedFuture;
 import static java.util.concurrent.CompletableFuture.failedFuture;
 import static org.apache.ignite.internal.configuration.util.ConfigurationUtil.getByInternalId;
 import static org.apache.ignite.internal.schema.SchemaManager.INITIAL_SCHEMA_VERSION;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
 import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
 import static org.apache.ignite.internal.utils.RebalanceUtil.PENDING_ASSIGNMENTS_PREFIX;
 import static org.apache.ignite.internal.utils.RebalanceUtil.STABLE_ASSIGNMENTS_PREFIX;
@@ -182,6 +183,12 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
     /** Set of futures that should complete before completion of {@link #tablesByIdVv}, after completion this set is cleared. */
     private final Set<CompletableFuture<?>> beforeTablesVvComplete = new ConcurrentHashSet<>();
 
+    /**
+     * {@link TableImpl} is created during update of tablesByIdVv, we store reference to it in case of updating of tablesByIdVv fails,
+     * so we can stop resources associated with the table.
+     */
+    private final Map<UUID, TableImpl> tablesToStopInCaseOfError = new ConcurrentHashMap<>();
+
     /** Resolver that resolves a network address to node id. */
     private final Function<NetworkAddress, String> netAddrResolver;
 
@@ -257,13 +264,42 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
 
             beforeTablesVvComplete.clear();
 
-            return CompletableFuture.allOf(futures.toArray(new CompletableFuture[] {}))
+            return CompletableFuture.allOf(futures.toArray(new CompletableFuture[]{}))
                     .orTimeout(TABLES_COMPLETE_TIMEOUT, TimeUnit.SECONDS)
                     .whenComplete((v, e) -> {
-                        if (e != null) {
-                            tablesByIdVv.completeExceptionally(token, e);
-                        } else {
+                        if (!busyLock.enterBusy()) {
+                            if (e != null) {
+                                LOG.warn("Error occurred while updating tables and stopping components.", e);
+                                // Stop of the components has been started, so we do nothing and resources of tablesByIdVv will be
+                                // freed in the logic of TableManager stop. We cannot complete tablesByIdVv exceptionally because
+                                // we will lose a context of tables.
+                            }
+                            return;
+                        }
+                        try {
+                            if (e != null) {
+                                LOG.warn("Error occurred while updating tables.", e);
+                                if (e instanceof CompletionException) {
+                                    Throwable th = e.getCause();
+                                    // Case when stopping of the previous component has been started and related futures completed
+                                    // exceptionally
+                                    if (th instanceof NodeStoppingException || (th.getCause() != null
+                                            && th.getCause() instanceof NodeStoppingException)) {
+                                        // Stop of the components has been started so we do nothing and resources will be freed in the
+                                        // logic of TableManager stop
+                                        return;
+                                    }
+                                }
+                                // TODO: https://issues.apache.org/jira/browse/IGNITE-17515
+                                tablesByIdVv.completeExceptionally(token, e);
+                            }
+
+                            //Normal scenario, when all related futures for tablesByIdVv are completed and we can complete tablesByIdVv
                             tablesByIdVv.complete(token);
+
+                            tablesToStopInCaseOfError.clear();
+                        } finally {
+                            busyLock.leaveBusy();
                         }
                     });
         });
@@ -599,6 +635,22 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
 
         Map<UUID, TableImpl> tables = tablesByIdVv.latest();
 
+        cleanUpTablesResources(tables);
+
+        cleanUpTablesResources(tablesToStopInCaseOfError);
+
+        tablesToStopInCaseOfError.clear();
+
+        shutdownAndAwaitTermination(rebalanceScheduler, 10, TimeUnit.SECONDS);
+        shutdownAndAwaitTermination(ioExecutor, 10, TimeUnit.SECONDS);
+    }
+
+    /**
+     * Stops resources that are related to provided tables.
+     *
+     * @param tables Tables to stop.
+     */
+    private void cleanUpTablesResources(Map<UUID, TableImpl> tables) {
         for (TableImpl table : tables.values()) {
             try {
                 for (int p = 0; p < table.internalTable().partitions(); p++) {
@@ -611,9 +663,6 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
                 LOG.info("Unable to stop table [name={}]", e, table.name());
             }
         }
-
-        shutdownAndAwaitTermination(rebalanceScheduler, 10, TimeUnit.SECONDS);
-        shutdownAndAwaitTermination(ioExecutor, 10, TimeUnit.SECONDS);
     }
 
     /**
@@ -657,7 +706,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
 
         var table = new TableImpl(internalTable);
 
-        tablesByIdVv.update(causalityToken, (previous, e) -> {
+        tablesByIdVv.update(causalityToken, (previous, e) -> inBusyLock(busyLock, () -> {
             if (e != null) {
                 return failedFuture(e);
             }
@@ -667,16 +716,20 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
             val.put(tblId, table);
 
             return completedFuture(val);
-        });
+        }));
 
         CompletableFuture<?> schemaFut = schemaManager.schemaRegistry(causalityToken, tblId)
-                .thenAccept(table::schemaView)
-                .thenCompose(v -> fireEvent(TableEvent.CREATE, new TableEventParameters(causalityToken, table)));
+                .thenAccept(schema -> inBusyLock(busyLock, () -> table.schemaView(schema)))
+                .thenCompose(
+                        v -> inBusyLock(busyLock, () -> fireEvent(TableEvent.CREATE, new TableEventParameters(causalityToken, table)))
+                );
 
         beforeTablesVvComplete.add(schemaFut);
 
+        tablesToStopInCaseOfError.put(tblId, table);
+
         // TODO should be reworked in IGNITE-16763
-        return tablesByIdVv.get(causalityToken).thenRun(() -> completeApiCreateFuture(table));
+        return tablesByIdVv.get(causalityToken).thenRun(() -> inBusyLock(busyLock, () -> completeApiCreateFuture(table)));
     }
 
     /**
@@ -710,7 +763,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
                 raftMgr.stopRaftGroup(partitionRaftGroupName(tblId, p));
             }
 
-            tablesByIdVv.update(causalityToken, (previousVal, e) -> {
+            tablesByIdVv.update(causalityToken, (previousVal, e) -> inBusyLock(busyLock, () -> {
                 if (e != null) {
                     return failedFuture(e);
                 }
@@ -720,7 +773,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
                 map.remove(tblId);
 
                 return completedFuture(map);
-            });
+            }));
 
             TableImpl table = tablesByIdVv.latest().get(tblId);
 
@@ -730,7 +783,9 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
             table.internalTable().storage().destroy();
 
             CompletableFuture<?> fut = schemaManager.dropRegistry(causalityToken, table.tableId())
-                    .thenCompose(v -> fireEvent(TableEvent.DROP, new TableEventParameters(causalityToken, table)));
+                    .thenCompose(
+                            v -> inBusyLock(busyLock, () -> fireEvent(TableEvent.DROP, new TableEventParameters(causalityToken, table)))
+                    );
 
             beforeTablesVvComplete.add(fut);
         } catch (Exception e) {
@@ -1082,8 +1137,8 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
      */
     private CompletableFuture<List<Table>> tablesAsyncInternal() {
         // TODO: IGNITE-16288 directTableIds should use async configuration API
-        return CompletableFuture.supplyAsync(this::directTableIds)
-                .thenCompose(tableIds -> {
+        return CompletableFuture.supplyAsync(() -> inBusyLock(busyLock, this::directTableIds))
+                .thenCompose(tableIds -> inBusyLock(busyLock, () -> {
                     var tableFuts = new CompletableFuture[tableIds.size()];
 
                     var i = 0;
@@ -1092,7 +1147,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
                         tableFuts[i++] = tableAsyncInternal(tblId, false);
                     }
 
-                    return allOf(tableFuts).thenApply(unused -> {
+                    return allOf(tableFuts).thenApply(unused -> inBusyLock(busyLock, () -> {
                         var tables = new ArrayList<Table>(tableIds.size());
 
                         try {
@@ -1108,8 +1163,8 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
                         }
 
                         return tables;
-                    });
-                });
+                    }));
+                }));
     }
 
     /**