This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 0257d0d886 [#13175] fix(fileset): Reject concurrent duplicate creates
(#13184)
0257d0d886 is described below
commit 0257d0d8869c9630fb3f884258c3c9ddb570b6a6
Author: Qi Yu <[email protected]>
AuthorDate: Wed Sep 16 08:45:19 2026 +0800
[#13175] fix(fileset): Reject concurrent duplicate creates (#13184)
### What changes were proposed in this pull request?
Use the existing `store.put(filesetEntity, false)` API for strict
fileset creation and map storage duplicates to
`FilesetAlreadyExistsException`.
Add a regression test with two independent catalog/store instances
sharing H2. Both requests pass the existence check before writing;
exactly one succeeds, and its ID, identifier property, comment, and
storage location remain intact.
### Why are the changes needed?
The existence check does not serialize requests across servers. Using
overwrite afterward lets both requests report success and allows the
second request to replace the winner's metadata.
Part of #13175. The EntityStore write-intent redesign is deferred to a
separate PR.
Fixed: #13175
### Does this PR introduce _any_ user-facing change?
Concurrent duplicate fileset creation now fails with an already-exists
error. Directory preparation retains its existing order before
insertion, so a losing request may still leave a directory. Filesystem
side-effect handling remains follow-up work.
### How was this patch tested?
All 163 fileset catalog operation tests passed, including the new
concurrent-create regression and the existing schema-deletion error
mapping test. Spotless and compilation with Error Prone/Werror passed.
Rerun with:
```bash
./gradlew :catalogs:catalog-fileset:test \
--tests
'org.apache.gravitino.catalog.fileset.TestFilesetCatalogOperations' -PskipITs
```
The test uses independent instances and database connections, without
launching two complete HTTP server processes.
---
.../catalog/fileset/FilesetCatalogOperations.java | 6 +-
.../fileset/TestFilesetCatalogOperations.java | 95 +++++++++++++++++++++-
2 files changed, 99 insertions(+), 2 deletions(-)
diff --git
a/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java
b/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java
index 1903f95924..5f06f75f25 100644
---
a/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java
+++
b/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java
@@ -69,6 +69,7 @@ import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.EntityStore;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.NameIdentifier;
@@ -612,7 +613,10 @@ public class FilesetCatalogOperations extends
ManagedSchemaOperations
.build();
try {
- store.put(filesetEntity, true /* overwrite */);
+ // The existence check is advisory; the strict insert decides concurrent
creates.
+ store.put(filesetEntity, false /* overwrite */);
+ } catch (EntityAlreadyExistsException exception) {
+ throw new FilesetAlreadyExistsException(exception, "Fileset %s already
exists", ident);
} catch (NoSuchEntityException exception) {
// The schema can disappear after the check near the start of this
method. The relational
// store detects that race while taking the parent-schema lock;
translate its storage-level
diff --git
a/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java
b/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java
index 1db30f01e2..6ca9cc9163 100644
---
a/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java
+++
b/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java
@@ -66,6 +66,10 @@ import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -89,6 +93,7 @@ import org.apache.gravitino.UserPrincipal;
import org.apache.gravitino.audit.CallerContext;
import org.apache.gravitino.audit.FilesetAuditConstants;
import org.apache.gravitino.audit.FilesetDataOperation;
+import org.apache.gravitino.cache.NoOpsCache;
import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
import org.apache.gravitino.catalog.hadoop.fs.FileSystemUtils;
import org.apache.gravitino.catalog.hadoop.fs.LocalFileSystemProvider;
@@ -101,6 +106,7 @@ import
org.apache.gravitino.credential.CatalogCredentialManager;
import org.apache.gravitino.credential.Credential;
import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.exceptions.ConnectionFailedException;
+import org.apache.gravitino.exceptions.FilesetAlreadyExistsException;
import org.apache.gravitino.exceptions.GravitinoRuntimeException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchFilesetException;
@@ -996,6 +1002,55 @@ public class TestFilesetCatalogOperations {
}
}
+ /** Checks independent catalog instances cannot both create the same
fileset. */
+ @Test
+ public void testConcurrentCreateAcrossIndependentNodesPreservesWinner()
throws Exception {
+ long testId = generateTestId();
+ String schemaName = "schema" + testId;
+ String catalogPath = TEST_ROOT_PATH + "/catalog" + testId;
+ createSchema(schemaName, "comment", catalogPath, null, true);
+ NameIdentifier ident = NameIdentifier.of("m1", "c1", schemaName,
"concurrent");
+ CountDownLatch checkedAbsent = new CountDownLatch(2);
+ EntityStore firstStore = independentNodeStore(ident, checkedAbsent);
+ EntityStore secondStore = independentNodeStore(ident, checkedAbsent);
+ String firstPath = catalogPath + "/" + schemaName + "/first";
+ String secondPath = catalogPath + "/" + schemaName + "/second";
+ long firstId = idGenerator.nextId();
+ long secondId = idGenerator.nextId();
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try (FilesetCatalogOperations firstOps =
+ new FilesetCatalogOperations(firstStore, secretManager);
+ FilesetCatalogOperations secondOps =
+ new FilesetCatalogOperations(secondStore, secretManager)) {
+ firstOps.initialize(Maps.newHashMap(), randomCatalogInfo(),
FILESET_PROPERTIES_METADATA);
+ secondOps.initialize(Maps.newHashMap(), randomCatalogInfo(),
FILESET_PROPERTIES_METADATA);
+ Future<Boolean> first =
+ executor.submit(
+ () -> createConcurrentFileset(firstOps, ident, firstPath,
"first", firstId));
+ Future<Boolean> second =
+ executor.submit(
+ () -> createConcurrentFileset(secondOps, ident, secondPath,
"second", secondId));
+ boolean firstWon = first.get(30, TimeUnit.SECONDS);
+ boolean secondWon = second.get(30, TimeUnit.SECONDS);
+ Assertions.assertNotEquals(firstWon, secondWon);
+ FilesetEntity actual = store.get(ident, Entity.EntityType.FILESET,
FilesetEntity.class);
+ Assertions.assertEquals(firstWon ? firstId : secondId, actual.id());
+ Assertions.assertEquals(firstWon ? "first" : "second", actual.comment());
+ Assertions.assertEquals(
+ actual.id(),
StringIdentifier.fromProperties(actual.properties()).id());
+ String winnerPath = firstWon ? firstPath : secondPath;
+ try (FileSystem fs = FileSystem.newInstance(new Configuration())) {
+ Assertions.assertTrue(fs.exists(new Path(winnerPath)));
+ Assertions.assertEquals(
+ new Path(winnerPath).makeQualified(fs.getUri(),
fs.getWorkingDirectory()).toString(),
+ new
Path(actual.storageLocations().values().iterator().next()).toString());
+ }
+ } finally {
+ executor.shutdownNow();
+ Assertions.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
+ }
+ }
+
@Test
public void testCreateFilesetMapsSchemaDeletionDuringStoreWrite() throws
IOException {
long testId = generateTestId();
@@ -1010,7 +1065,7 @@ public class TestFilesetCatalogOperations {
NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, "schema",
schemaName);
Mockito.doThrow(deletedSchema)
.when(racingStore)
- .put(Mockito.any(FilesetEntity.class), Mockito.eq(true));
+ .put(Mockito.any(FilesetEntity.class), Mockito.eq(false));
try (FilesetCatalogOperations ops = new
FilesetCatalogOperations(racingStore, secretManager)) {
ops.initialize(
@@ -3879,4 +3934,42 @@ public class TestFilesetCatalogOperations {
SecretConstants.ATTR_ENTITY_ID, String.valueOf(entityId),
SecretConstants.ATTR_PROPERTY_KEY, key));
}
+
+ private EntityStore independentNodeStore(NameIdentifier ident,
CountDownLatch checkedAbsent)
+ throws Exception {
+ RelationalEntityStore node = new RelationalEntityStore();
+ FieldUtils.writeField(node, "backend", FieldUtils.readField(store,
"backend", true), true);
+ FieldUtils.writeField(node, "cache", new
NoOpsCache(Mockito.mock(Config.class)), true);
+ EntityStore racingStore = Mockito.spy(node);
+ Mockito.doAnswer(
+ invocation -> {
+ boolean exists = (boolean) invocation.callRealMethod();
+ Assertions.assertFalse(exists);
+ checkedAbsent.countDown();
+ Assertions.assertTrue(checkedAbsent.await(30, TimeUnit.SECONDS));
+ return exists;
+ })
+ .when(racingStore)
+ .exists(ident, Entity.EntityType.FILESET);
+ return racingStore;
+ }
+
+ private boolean createConcurrentFileset(
+ FilesetCatalogOperations ops,
+ NameIdentifier ident,
+ String location,
+ String comment,
+ long id) {
+ try {
+ ops.createMultipleLocationFileset(
+ ident,
+ comment,
+ Fileset.Type.MANAGED,
+ ImmutableMap.of("default", location),
+ ImmutableMap.of(StringIdentifier.ID_KEY,
StringIdentifier.fromId(id).toString()));
+ return true;
+ } catch (FilesetAlreadyExistsException expected) {
+ return false;
+ }
+ }
}
