Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

IGNITE-23220 Avoid concurrent Catalog writes to Metastorage on the same node #4406

Merged
merged 1 commit into from
Sep 17, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions modules/catalog/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -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')
Expand Down Expand Up @@ -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"
Original file line number Diff line number Diff line change
@@ -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 + "'"
);
});
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*/
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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);

Expand Down