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

…me node

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<Integer> 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.exceptionally(ex -> null);
+
+            return chainedFuture;
+        }
+    }
+
     private CompletableFuture<Integer> awaitVersionActivation(int version) {
         Catalog catalog = catalogByVer.get(version);