You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "geniusjoe (via GitHub)" <gi...@apache.org> on 2026/04/13 14:29:51 UTC
[PR] [fix][broker] Clean up orphan ledger on concurrent initial schema creation in BookkeeperSchemaStorage [pulsar]
geniusjoe opened a new pull request, #25514:
URL: https://github.com/apache/pulsar/pull/25514
Fixes #18292
Related #18701
### Motivation
When multiple requests concurrently create a schema for a brand-new topic (i.e., the schema locator z-node does not yet exist), each request first creates a BookKeeper ledger via `addNewSchemaEntryToStore`, then attempts to create the schema locator z-node via CAS (`createSchemaLocator` with `expectedVersion = -1L`).
Only one request succeeds; the others fail with BadVersionException (because ZK MetadataStore translates NODEEXISTS to BadVersionException when expectedVersion == -1). However, the ledgers created by the failed requests were never cleaned up, resulting in orphan ledgers (dirty data) in BookKeeper.
Note that the existing `updateSchemaLocator` method already has cleanup logic for this scenario (deleting the orphan ledger when CAS fails with `BadVersionException`) in #18701, but the `createNewSchema` method — which handles the **initial** schema creation path — was missing this cleanup.
### Modifications
- **`BookkeeperSchemaStorage.createNewSchema`**: Added a `whenComplete` callback after `createSchemaLocator`. When the CAS operation fails due to `AlreadyExistsException` or `BadVersionException`, the orphan BookKeeper ledger is asynchronously deleted, consistent with the existing cleanup logic in `updateSchemaLocator`. The method is also refactored into three clearly commented steps for better readability.
- **`PulsarMockLedgerHandle`**: Added a new constructor that accepts `Map<String, byte[]> customMetadata` and passes it to `LedgerMetadataBuilder.withCustomMetadata()`, so that mock ledgers can retain custom metadata set during creation.
- **`PulsarMockBookKeeper`**: Updated `asyncCreateLedger` to forward the `properties` parameter to the new `PulsarMockLedgerHandle` constructor, enabling tests to inspect ledger custom metadata.
- **`SchemaTest`**: Added `testConcurrentCreateSchemaNoOrphanLedger` test that verifies orphan ledgers are cleaned up when 16 producers concurrently create schema on a brand-new topic. The test inspects surviving BK ledgers via `PulsarMockBookKeeper.getLedgerMap()` and asserts that only 1 ledger with matching `pulsar/schemaId` custom metadata exists.
### Verifying this change
This change added tests and can be verified as follows:
- Added `testConcurrentCreateSchemaNoOrphanLedger` in `SchemaTest` that concurrently creates 16 producers with the same AVRO schema on a brand-new topic, then verifies:
1. Only 1 schema version exists via `admin.schemas().getAllSchemas()`
2. Only 1 surviving BK ledger has `customMetadata["pulsar/schemaId"]` matching the topic's schema name (orphan ledgers from failed concurrent creations were deleted)
### Does this pull request potentially affect one of the following parts:
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [x] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Clean up orphan ledger on concurrent initial schema creation in BookkeeperSchemaStorage [pulsar]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on PR #25514:
URL: https://github.com/apache/pulsar/pull/25514#issuecomment-4305318748
@Denovo1998
Thank you very much for your review. I initially considered removing the redundant schema ledger as an asynchronous task to avoid blocking the main schema creation logic.
However, the current approach could indeed introduce race condition issues in unit tests. I think your proposed solution makes sense. If you have time, you could go ahead and implement it as you suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Clean up orphan ledger on concurrent initial schema creation in BookkeeperSchemaStorage [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on PR #25514:
URL: https://github.com/apache/pulsar/pull/25514#issuecomment-4304763201
@geniusjoe @dao-jun @codelipenghui @hanmz
Seems like there are some issues?
```
Request A and Request B create a schema for a new topic simultaneously.
T1: A finds the locator does not exist.
T2: B also finds the locator does not exist.
T3: A creates BK ledger L1 and writes the schema entry.
T4: B creates BK ledger L2 and writes the schema entry.
T5: A's createSchemaLocator is successful, and the locator points to L1.
T6: B's createSchemaLocator fails, receiving a BadVersionException / AlreadyExistsException.
T7: B's whenComplete calls asyncDeleteLedger(L2).
T8: asyncDeleteLedger only asynchronously commits deletion and returns immediately.
T9: B's createNewSchema future still completes with a BadVersionException.
T10: The upper BookkeeperSchemaStorage#put catches the BadVersionException and immediately retries.
T11: The retry re-reads the locator, finds that the schema already exists, and the request succeeds and returns.
T12: But the deletion callback of L2 may not have finished executing.
```
`whenComplete` only waits for this callback to return, but the callback starts a callback-based `bookKeeper.asyncDeleteLedger(...)` and returns before the ledger deletion is confirmed. As a result, the outer `put(...)` can observe the original `BadVersionException` / `AlreadyExistsException`, retry immediately, and complete the schema creation before the orphan ledger is actually removed. This makes the new `getLedgerMap()` assertion timing-sensitive, and also weakens the cleanup guarantee exposed to callers. Could we wrap `asyncDeleteLedger` in a `CompletableFuture` and compose it before rethrowing the original CAS exception so the existing retry happens after the delete callback completes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Clean up orphan ledger on concurrent initial schema creation in BookkeeperSchemaStorage [pulsar]
Posted by "hanmz (via GitHub)" <gi...@apache.org>.
hanmz commented on PR #25514:
URL: https://github.com/apache/pulsar/pull/25514#issuecomment-4302328921
/pulsarbot rerun-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Clean up orphan ledger on concurrent initial schema creation in BookkeeperSchemaStorage [pulsar]
Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun merged PR #25514:
URL: https://github.com/apache/pulsar/pull/25514
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Clean up orphan ledger on concurrent initial schema creation in BookkeeperSchemaStorage [pulsar]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #25514:
URL: https://github.com/apache/pulsar/pull/25514#discussion_r3124840529
##########
pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java:
##########
@@ -1376,6 +1376,68 @@ public void testCreateSchemaInParallel() throws Exception {
producers2.clear();
}
+ /**
+ * Test that when multiple producers concurrently create schema for a brand-new topic,
+ * the orphan BookKeeper ledgers created by the losing requests are properly cleaned up.
+ */
+ @Test
+ public void testConcurrentCreateSchemaNoOrphanLedger() throws Exception {
+ final String namespace = "test-namespace-" + randomName(16);
+ String ns = PUBLIC_TENANT + "/" + namespace;
+ admin.namespaces().createNamespace(ns, Sets.newHashSet(CLUSTER_NAME));
+
+ final String topic = getTopicName(ns, "testConcurrentCreateSchemaNoOrphanLedger");
+ final String schemaName = TopicName.get(topic).getSchemaName();
+
+ org.apache.bookkeeper.client.PulsarMockBookKeeper mockBk =
+ (org.apache.bookkeeper.client.PulsarMockBookKeeper) pulsar.getBookKeeperClient();
+
+ // Concurrently create producers with the same schema on a brand-new topic
+ int concurrency = 16;
+ @Cleanup("shutdownNow")
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ List<CompletableFuture<Producer<Schemas.PersonOne>>> producers = new ArrayList<>(concurrency);
+ CountDownLatch latch = new CountDownLatch(concurrency);
+ for (int i = 0; i < concurrency; i++) {
+ executor.execute(() -> {
+ producers.add(pulsarClient.newProducer(Schema.AVRO(Schemas.PersonOne.class))
+ .topic(topic).createAsync());
+ latch.countDown();
+ });
+ }
+ latch.await();
+ FutureUtil.waitForAll(producers).join();
+
+ // Verify only 1 schema version exists
+ assertEquals(admin.schemas().getAllSchemas(topic).size(), 1);
+
+ // Count surviving BK ledgers whose customMetadata "pulsar/schemaId" matches this topic's schemaName.
+ // If orphan ledgers were not cleaned up, there would be more than 1.
+ int schemaLedgerCount = 0;
+ for (org.apache.bookkeeper.client.PulsarMockLedgerHandle lh
+ : mockBk.getLedgerMap().values()) {
+ Map<String, byte[]> metadata = lh.getLedgerMetadata().getCustomMetadata();
+ byte[] schemaIdBytes = metadata.get("pulsar/schemaId");
Review Comment:
Fixed
##########
pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java:
##########
@@ -1376,6 +1376,68 @@ public void testCreateSchemaInParallel() throws Exception {
producers2.clear();
}
+ /**
+ * Test that when multiple producers concurrently create schema for a brand-new topic,
+ * the orphan BookKeeper ledgers created by the losing requests are properly cleaned up.
+ */
+ @Test
+ public void testConcurrentCreateSchemaNoOrphanLedger() throws Exception {
+ final String namespace = "test-namespace-" + randomName(16);
+ String ns = PUBLIC_TENANT + "/" + namespace;
+ admin.namespaces().createNamespace(ns, Sets.newHashSet(CLUSTER_NAME));
+
+ final String topic = getTopicName(ns, "testConcurrentCreateSchemaNoOrphanLedger");
+ final String schemaName = TopicName.get(topic).getSchemaName();
+
+ org.apache.bookkeeper.client.PulsarMockBookKeeper mockBk =
+ (org.apache.bookkeeper.client.PulsarMockBookKeeper) pulsar.getBookKeeperClient();
+
+ // Concurrently create producers with the same schema on a brand-new topic
+ int concurrency = 16;
+ @Cleanup("shutdownNow")
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ List<CompletableFuture<Producer<Schemas.PersonOne>>> producers = new ArrayList<>(concurrency);
Review Comment:
Fixed
##########
pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java:
##########
@@ -1376,6 +1376,68 @@ public void testCreateSchemaInParallel() throws Exception {
producers2.clear();
}
+ /**
+ * Test that when multiple producers concurrently create schema for a brand-new topic,
+ * the orphan BookKeeper ledgers created by the losing requests are properly cleaned up.
+ */
+ @Test
+ public void testConcurrentCreateSchemaNoOrphanLedger() throws Exception {
+ final String namespace = "test-namespace-" + randomName(16);
+ String ns = PUBLIC_TENANT + "/" + namespace;
+ admin.namespaces().createNamespace(ns, Sets.newHashSet(CLUSTER_NAME));
+
+ final String topic = getTopicName(ns, "testConcurrentCreateSchemaNoOrphanLedger");
+ final String schemaName = TopicName.get(topic).getSchemaName();
+
+ org.apache.bookkeeper.client.PulsarMockBookKeeper mockBk =
+ (org.apache.bookkeeper.client.PulsarMockBookKeeper) pulsar.getBookKeeperClient();
+
+ // Concurrently create producers with the same schema on a brand-new topic
+ int concurrency = 16;
+ @Cleanup("shutdownNow")
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ List<CompletableFuture<Producer<Schemas.PersonOne>>> producers = new ArrayList<>(concurrency);
+ CountDownLatch latch = new CountDownLatch(concurrency);
+ for (int i = 0; i < concurrency; i++) {
+ executor.execute(() -> {
+ producers.add(pulsarClient.newProducer(Schema.AVRO(Schemas.PersonOne.class))
+ .topic(topic).createAsync());
+ latch.countDown();
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Clean up orphan ledger on concurrent initial schema creation in BookkeeperSchemaStorage [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #25514:
URL: https://github.com/apache/pulsar/pull/25514#discussion_r3121583301
##########
pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java:
##########
@@ -1376,6 +1376,68 @@ public void testCreateSchemaInParallel() throws Exception {
producers2.clear();
}
+ /**
+ * Test that when multiple producers concurrently create schema for a brand-new topic,
+ * the orphan BookKeeper ledgers created by the losing requests are properly cleaned up.
+ */
+ @Test
+ public void testConcurrentCreateSchemaNoOrphanLedger() throws Exception {
+ final String namespace = "test-namespace-" + randomName(16);
+ String ns = PUBLIC_TENANT + "/" + namespace;
+ admin.namespaces().createNamespace(ns, Sets.newHashSet(CLUSTER_NAME));
+
+ final String topic = getTopicName(ns, "testConcurrentCreateSchemaNoOrphanLedger");
+ final String schemaName = TopicName.get(topic).getSchemaName();
+
+ org.apache.bookkeeper.client.PulsarMockBookKeeper mockBk =
+ (org.apache.bookkeeper.client.PulsarMockBookKeeper) pulsar.getBookKeeperClient();
+
+ // Concurrently create producers with the same schema on a brand-new topic
+ int concurrency = 16;
+ @Cleanup("shutdownNow")
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ List<CompletableFuture<Producer<Schemas.PersonOne>>> producers = new ArrayList<>(concurrency);
Review Comment:
Use Collections.synchronizedList(...)
##########
pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java:
##########
@@ -1376,6 +1376,68 @@ public void testCreateSchemaInParallel() throws Exception {
producers2.clear();
}
+ /**
+ * Test that when multiple producers concurrently create schema for a brand-new topic,
+ * the orphan BookKeeper ledgers created by the losing requests are properly cleaned up.
+ */
+ @Test
+ public void testConcurrentCreateSchemaNoOrphanLedger() throws Exception {
+ final String namespace = "test-namespace-" + randomName(16);
+ String ns = PUBLIC_TENANT + "/" + namespace;
+ admin.namespaces().createNamespace(ns, Sets.newHashSet(CLUSTER_NAME));
+
+ final String topic = getTopicName(ns, "testConcurrentCreateSchemaNoOrphanLedger");
+ final String schemaName = TopicName.get(topic).getSchemaName();
+
+ org.apache.bookkeeper.client.PulsarMockBookKeeper mockBk =
+ (org.apache.bookkeeper.client.PulsarMockBookKeeper) pulsar.getBookKeeperClient();
+
+ // Concurrently create producers with the same schema on a brand-new topic
+ int concurrency = 16;
+ @Cleanup("shutdownNow")
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ List<CompletableFuture<Producer<Schemas.PersonOne>>> producers = new ArrayList<>(concurrency);
+ CountDownLatch latch = new CountDownLatch(concurrency);
+ for (int i = 0; i < concurrency; i++) {
+ executor.execute(() -> {
+ producers.add(pulsarClient.newProducer(Schema.AVRO(Schemas.PersonOne.class))
+ .topic(topic).createAsync());
+ latch.countDown();
+ });
+ }
+ latch.await();
+ FutureUtil.waitForAll(producers).join();
+
+ // Verify only 1 schema version exists
+ assertEquals(admin.schemas().getAllSchemas(topic).size(), 1);
+
+ // Count surviving BK ledgers whose customMetadata "pulsar/schemaId" matches this topic's schemaName.
+ // If orphan ledgers were not cleaned up, there would be more than 1.
+ int schemaLedgerCount = 0;
+ for (org.apache.bookkeeper.client.PulsarMockLedgerHandle lh
+ : mockBk.getLedgerMap().values()) {
+ Map<String, byte[]> metadata = lh.getLedgerMetadata().getCustomMetadata();
+ byte[] schemaIdBytes = metadata.get("pulsar/schemaId");
Review Comment:
Replace with LedgerMetadataUtils.METADATA_PROPERTY_SCHEMAID
##########
pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java:
##########
@@ -1376,6 +1376,68 @@ public void testCreateSchemaInParallel() throws Exception {
producers2.clear();
}
+ /**
+ * Test that when multiple producers concurrently create schema for a brand-new topic,
+ * the orphan BookKeeper ledgers created by the losing requests are properly cleaned up.
+ */
+ @Test
+ public void testConcurrentCreateSchemaNoOrphanLedger() throws Exception {
+ final String namespace = "test-namespace-" + randomName(16);
+ String ns = PUBLIC_TENANT + "/" + namespace;
+ admin.namespaces().createNamespace(ns, Sets.newHashSet(CLUSTER_NAME));
+
+ final String topic = getTopicName(ns, "testConcurrentCreateSchemaNoOrphanLedger");
+ final String schemaName = TopicName.get(topic).getSchemaName();
+
+ org.apache.bookkeeper.client.PulsarMockBookKeeper mockBk =
+ (org.apache.bookkeeper.client.PulsarMockBookKeeper) pulsar.getBookKeeperClient();
+
+ // Concurrently create producers with the same schema on a brand-new topic
+ int concurrency = 16;
+ @Cleanup("shutdownNow")
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ List<CompletableFuture<Producer<Schemas.PersonOne>>> producers = new ArrayList<>(concurrency);
+ CountDownLatch latch = new CountDownLatch(concurrency);
+ for (int i = 0; i < concurrency; i++) {
+ executor.execute(() -> {
+ producers.add(pulsarClient.newProducer(Schema.AVRO(Schemas.PersonOne.class))
+ .topic(topic).createAsync());
+ latch.countDown();
Review Comment:
Need to execute `latch.countDown();` in finally block to avoid any exceptions cause the test stuck.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org