IGNITE-23220 Avoid concurrent Catalog writes to Metastorage on the sa…

…me node (#4406)

Each DDL tries to make a write to the Metastorage. Each write is assigned a sequential number (version); if a few DDLs try to write to the Metastorage, they will all compete for the same version, and only one of them can succeed. This competition is wasteful. This commit avoids such concurrency on the same node by explicitly linearizing writes to the Metastorage from the Catalog before doing them.

modules/catalog/build.gradle

@@ -19,6 +19,7 @@ apply from: "$rootDir/buildscripts/java-core.gradle"
 apply from: "$rootDir/buildscripts/publishing.gradle"
 apply from: "$rootDir/buildscripts/java-junit5.gradle"
 apply from: "$rootDir/buildscripts/java-test-fixtures.gradle"
+apply from: "$rootDir/buildscripts/java-integration-test.gradle"
 
 dependencies {
     annotationProcessor project(':ignite-configuration-annotation-processor')
@@ -56,6 +57,10 @@ dependencies {
     testFixturesImplementation(testFixtures(project(':ignite-core')))
     testFixturesImplementation(testFixtures(project(':ignite-vault')))
     testFixturesImplementation(testFixtures(project(':ignite-metastorage')))
+
+    integrationTestImplementation project(':ignite-api')
+    integrationTestImplementation testFixtures(project(':ignite-runner'))
+    integrationTestImplementation testFixtures(project(':ignite-core'))
 }
 
 description = "ignite-catalog"

modules/catalog/src/integrationTest/java/org/apache/ignite/internal/catalog/it/ItConcurrentDdlsTest.java

@@ -0,0 +1,69 @@
+/*
+ * 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.catalog.it;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+import java.util.concurrent.ForkJoinPool;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.util.IgniteUtils;
+import org.junit.jupiter.api.Test;
+
+class ItConcurrentDdlsTest extends ClusterPerTestIntegrationTest {
+    private static final String ZONE_NAME = "TEST_ZONE";
+
+    @Override
+    protected int initialNodes() {
+        return 1;
+    }
+
+    /**
+     * Makes sure we can request DDLs with high concurrency and not fall into 'no more retry attempts' trap.
+     */
+    @Test
+    void createTablesConcurrently() {
+        // Just 1 partition to make the test lighter and faster (number of partitions is not the focus of this test).
+        createZoneWith1Partition();
+
+        ForkJoinPool pool = new ForkJoinPool(10);
+
+        assertDoesNotThrow(() -> pool.submit(this::createTablesInParallel).get());
+
+        IgniteUtils.shutdownAndAwaitTermination(pool, 10, SECONDS);
+    }
+
+    private void createZoneWith1Partition() {
+        node(0).sql().executeScript(
+                "CREATE ZONE " + ZONE_NAME + " with partitions=1, replicas=1, storage_profiles='"
+                        + CatalogService.DEFAULT_STORAGE_PROFILE + "'"
+        );
+    }
+
+    private void createTablesInParallel() {
+        IntStream.range(0, 30).parallel().forEach(n -> {
+            String tableName = "TEST" + n;
+
+            node(0).sql().executeScript(
+                    "CREATE TABLE " + tableName + " (id INT PRIMARY KEY, val VARCHAR) WITH primary_zone='" + ZONE_NAME + "'"
+            );
+        });
+    }
+}

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java

@@ -124,6 +124,19 @@ public class CatalogManagerImpl extends AbstractEventProducer<CatalogEvent, Cata
     /** Busy lock to stop synchronously. */
     private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
 
+    /**
+     * Future used to chain local appends to UpdateLog to avoid useless concurrency (as all concurrent attempts to append compete for
+     * the same catalog version, hence only one will win and the rest will have to retry).
+     *
+     * <p>Guarded by {@link #lastSaveUpdateFutureMutex}.
+     */
+    private CompletableFuture<Void> lastSaveUpdateFuture = nullCompletedFuture();
+
+    /**
+     * Guards access to {@link #lastSaveUpdateFuture}.
+     */
+    private final Object lastSaveUpdateFutureMutex = new Object();
+
     /**
      * Constructor.
      */
@@ -408,7 +421,7 @@ private void truncateUpTo(Catalog catalog) {
     private CompletableFuture<Integer> saveUpdateAndWaitForActivation(UpdateProducer updateProducer) {
         CompletableFuture<Integer> resultFuture = new CompletableFuture<>();
 
-        saveUpdate(updateProducer, 0)
+        saveUpdateEliminatingLocalConcurrency(updateProducer)
                 .thenCompose(this::awaitVersionActivation)
                 .whenComplete((newVersion, err) -> {
                     if (err != null) {
@@ -444,6 +457,20 @@ private CompletableFuture<Integer> saveUpdateAndWaitForActivation(UpdateProducer
         return resultFuture;
     }
 
+    private CompletableFuture<Integer> saveUpdateEliminatingLocalConcurrency(UpdateProducer updateProducer) {
+        // Avoid useless and wasteful competition for the save catalog version by enforcing an order.
+        synchronized (lastSaveUpdateFutureMutex) {
+            CompletableFuture<Integer> chainedFuture = lastSaveUpdateFuture
+                    .thenCompose(unused -> saveUpdate(updateProducer, 0));
+
+            // Suppressing any exception to make sure it doesn't ruin subsequent appends. The suppression is not a problem
+            // as the callers will handle exceptions anyway.
+            lastSaveUpdateFuture = chainedFuture.handle((res, ex) -> null);
+
+            return chainedFuture;
+        }
+    }
+
     private CompletableFuture<Integer> awaitVersionActivation(int version) {
         Catalog catalog = catalogByVer.get(version);