From 362d92f464680ac44c74c7a33e3f773b9de35354 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 14 Oct 2020 16:52:36 +0800 Subject: [PATCH 1/2] Record unused schema ledgers to ZK --- .../schema/BookkeeperSchemaStorage.java | 48 +++++++++++++++---- .../schema/PartitionedTopicsSchemaTest.java | 45 +++++++++++++++++ 2 files changed, 84 insertions(+), 9 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java index 1021f5e04b267..892113313ce3f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java @@ -40,10 +40,10 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import javax.validation.constraints.NotNull; -import org.apache.bookkeeper.client.AsyncCallback; import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.LedgerEntry; @@ -298,7 +298,8 @@ private CompletableFuture putSchema(String schemaId, byte[] data, byte[] h } else { // No schema was defined yet CompletableFuture future = new CompletableFuture<>(); - createNewSchema(schemaId, data, hash) + AtomicLong ledgerId = new AtomicLong(-1); + createNewSchema(schemaId, data, hash, ledgerId) .thenAccept(future::complete) .exceptionally(ex -> { if (ex.getCause() instanceof NodeExistsException || @@ -306,12 +307,15 @@ private CompletableFuture putSchema(String schemaId, byte[] data, byte[] h // There was a race condition on the schema creation. Since it has now been created, // retry the whole operation so that we have a chance to recover without bubbling error // back to producer/consumer - putSchema(schemaId, data, hash) - .thenAccept(future::complete) - .exceptionally(ex2 -> { - future.completeExceptionally(ex2); - return null; - }); + recordLedger(ledgerId.get(), schemaId).thenApply(ignored -> { + putSchema(schemaId, data, hash) + .thenAccept(future::complete) + .exceptionally(ex2 -> { + future.completeExceptionally(ex2); + return null; + }); + return null; + }); } else { // For other errors, just fail the operation future.completeExceptionally(ex); @@ -324,7 +328,27 @@ private CompletableFuture putSchema(String schemaId, byte[] data, byte[] h }); } - private CompletableFuture createNewSchema(String schemaId, byte[] data, byte[] hash) { + @NotNull + private CompletableFuture recordLedger(long ledgerId, String schemaId) { + // When creating schema for a partitioned topic, each partition would try to create its schema but only one + // would success. Then N-1 ledgers would be created but they're not used anymore, if N is the partition count. + // So we record these ledger ids to ZK, then we can delete these ledgers later. + CompletableFuture future = new CompletableFuture<>(); + + final String path = getUnusedLedgerPath(schemaId) + "/" + ledgerId; + ZkUtils.asyncCreateFullPathOptimistic(zooKeeper, path, new byte[]{}, Acl, + CreateMode.PERSISTENT, (rc, path1, ctx, name) -> { + Code code = Code.get(rc); + if (code != Code.OK) { + log.warn("Failed to create {}: {}", path, code); + } + future.complete(null); + }, null); + + return future; + } + + private CompletableFuture createNewSchema(String schemaId, byte[] data, byte[] hash, AtomicLong ledgerId) { SchemaStorageFormat.IndexEntry emptyIndex = SchemaStorageFormat.IndexEntry.newBuilder() .setVersion(0) .setHash(copyFrom(hash)) @@ -335,6 +359,7 @@ private CompletableFuture createNewSchema(String schemaId, byte[] data, by return addNewSchemaEntryToStore(schemaId, Collections.singletonList(emptyIndex), data).thenCompose(position -> { // The schema was stored in the ledger, now update the z-node with the pointer to it + ledgerId.set(position.getLedgerId()); SchemaStorageFormat.IndexEntry info = SchemaStorageFormat.IndexEntry.newBuilder() .setVersion(0) .setPosition(position) @@ -393,6 +418,11 @@ private static String getSchemaPath(String schemaId) { return SchemaPath + "/" + schemaId; } + @NotNull + public static String getUnusedLedgerPath(String schemaId) { + return String.join("/", SchemaPath, schemaId, "unusedLedger"); + } + @NotNull private CompletableFuture addNewSchemaEntryToStore( String schemaId, diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/PartitionedTopicsSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/PartitionedTopicsSchemaTest.java index 93e41dd11e7af..81d26bfbfbc49 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/PartitionedTopicsSchemaTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/PartitionedTopicsSchemaTest.java @@ -22,10 +22,15 @@ import static org.testng.Assert.assertTrue; import java.util.Collections; +import java.util.List; import java.util.Set; import java.util.TreeSet; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import org.apache.bookkeeper.client.BKException; +import org.apache.bookkeeper.client.BookKeeper; +import org.apache.bookkeeper.conf.ClientConfiguration; import org.apache.pulsar.broker.service.BkEnsemblesTestBase; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; @@ -33,6 +38,7 @@ import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionInitialPosition; +import org.apache.zookeeper.ZooKeeper; import org.testng.annotations.Test; public class PartitionedTopicsSchemaTest extends BkEnsemblesTestBase { @@ -106,4 +112,43 @@ public void partitionedTopicWithSchema() throws Exception { client.close(); } + /** + * Test for a partitioned topic with N partitions, after schema created, N-1 unused ledgers would be recorded in ZK + */ + @Test + public void testUnusedLedgers() throws Exception { + final String namespace = "prop/partitioned-topics-schema-test"; + final String topic = namespace + "/unused-ledgers"; + final int numPartitions = 10; + + admin.namespaces().createNamespace(namespace); + admin.topics().createPartitionedTopic(topic, numPartitions); + + PulsarClient client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); + client.newProducer(Schema.STRING).topic(topic).create(); // schema will be created here + + // Check the z-node which would record N-1 unused schema ledgers + ZooKeeper zooKeeper = bkEnsemble.getZkClient(); + List unusedLedgers = zooKeeper.getChildren(BookkeeperSchemaStorage.getUnusedLedgerPath(topic), null); + assertEquals(unusedLedgers.size(), numPartitions - 1); + + client.close(); + + // Verify these ledgers exist + BookKeeper bookKeeper = new BookKeeper( + new ClientConfiguration().setMetadataServiceUri("zk+null://" + config.getZookeeperServers() + "/ledgers")); + + final CountDownLatch latch = new CountDownLatch(unusedLedgers.size()); + unusedLedgers.stream().map(Long::parseLong).forEach(ledgerId -> { + bookKeeper.asyncOpenLedger(ledgerId, BookKeeper.DigestType.CRC32, "".getBytes(), + (rc, lh, ctx) -> { + assertEquals(rc, BKException.Code.OK); + latch.countDown(); + }, null); + }); + latch.await(); + + bookKeeper.close(); + } + } From 8d4f55aa31eeed3fe619491eba2796bb0d605080 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 14 Oct 2020 23:03:38 +0800 Subject: [PATCH 2/2] Find schema ledgers from ZK instead of memory when delete a topic's schema --- .../schema/BookkeeperSchemaStorage.java | 133 +++++++++++++----- 1 file changed, 98 insertions(+), 35 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java index 892113313ce3f..d7f903d552ea1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java @@ -41,6 +41,7 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Stream; import javax.validation.constraints.NotNull; @@ -62,6 +63,7 @@ import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.Code; import org.apache.zookeeper.KeeperException.NodeExistsException; +import org.apache.zookeeper.ZKUtil; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.ACL; @@ -81,9 +83,6 @@ public class BookkeeperSchemaStorage implements SchemaStorage { private final ServiceConfiguration config; private BookKeeper bookKeeper; - // schemaId => ledgers of the schemaId - private final Map> schemaLedgers = new ConcurrentHashMap<>(); - private final ConcurrentMap> readSchemaOperations = new ConcurrentHashMap<>(); @VisibleForTesting @@ -375,40 +374,108 @@ private CompletableFuture createNewSchema(String schemaId, byte[] data, by }); } + /** + * Delete ledgers asynchronously and ignore the exception if failed to delete any ledger + */ + @NotNull + private CompletableFuture deleteLedgers(Stream ledgers, final int numOfLedgers) { + if (numOfLedgers == 0) { + return CompletableFuture.completedFuture(null); + } + CompletableFuture future = new CompletableFuture<>(); + AtomicInteger numOfDeletedLedgers = new AtomicInteger(0); + ledgers.forEach(ledgerId -> { + bookKeeper.asyncDeleteLedger(ledgerId, (rc, ctx) -> { + if (rc == BKException.Code.OK) { + log.debug("Schema ledger {} is deleted", ledgerId); + } else { + // It's not a serious error, we didn't need call future.completeExceptionally() + log.warn("Failed to delete ledger {}: {}", ledgerId, rc); + } + if (numOfDeletedLedgers.incrementAndGet() == numOfLedgers) { + future.complete(null); + } + }, null); + }); + return future; + } + + /** + * Delete the ledgers recorded in z-node `/schemas///`'s data + */ + @NotNull + private CompletableFuture deleteUsedLedgers(final String schemaId) { + CompletableFuture future = new CompletableFuture<>(); + getSchemaLocator(getSchemaPath(schemaId)).whenComplete((optional, e) -> { + if (e != null) { + future.completeExceptionally(e); + return; + } + if (optional.isPresent()) { + List entries = optional.get().locator.getIndexList(); + deleteLedgers(entries.stream().map(entry -> entry.getPosition().getLedgerId()), entries.size()) + .thenApply(future::complete); + } else { + future.completeExceptionally(null); + } + }); + return future; + } + + /** + * Delete the ledgers recorded in z-node `/schemas////unusedLedger`'s children + */ + @NotNull + private CompletableFuture deleteUnusedLedgers(final String schemaId) { + CompletableFuture future = new CompletableFuture<>(); + + zooKeeper.getChildren(getUnusedLedgerPath(schemaId), null, (rc, path, ctx, children) -> { + Code code = Code.get(rc); + if (code == Code.OK) { + deleteLedgers(children.stream().map(Long::parseLong), children.size()).whenComplete((ignored, e) -> { + future.complete(null); + }); + } else if (code == Code.NONODE) { // a non-partitioned topic has no unused ledgers + future.complete(null); + } else { + future.completeExceptionally(new RuntimeException("Failed to get children of " + path + ": " + code)); + log.error("Failed to get children of {}: {}", path, code); + } + }, null); + + return future; + } + + /** + * Delete the associated z-node of `schemaId` + */ + @NotNull + private CompletableFuture deleteMetadata(final String schemaId) { + CompletableFuture future = new CompletableFuture<>(); + + final String path = getSchemaPath(schemaId); + try { + ZKUtil.deleteRecursive(zooKeeper, path); + future.complete(null); + log.info("z-node {} is deleted recursively", path); + } catch (InterruptedException | KeeperException e) { + future.completeExceptionally(e); + log.error("Failed to delete z-node {} recursively", path); + } + + return future; + } + @NotNull private CompletableFuture deleteSchema(String schemaId) { return getSchema(schemaId).thenCompose(schemaAndVersion -> { if (isNull(schemaAndVersion)) { return completedFuture(null); } else { - // The version is only for the compatibility of the current interface - final long version = -1; - final List ledgerIds = schemaLedgers.get(schemaId); - if (ledgerIds != null) { - CompletableFuture future = new CompletableFuture<>(); - final AtomicInteger numOfLedgerIds = new AtomicInteger(ledgerIds.size()); - for (long ledgerId : ledgerIds) { - bookKeeper.asyncDeleteLedger(ledgerId, (int rc, Object cnx) -> { - if (rc != BKException.Code.OK) { - // It's not a serious error, we didn't need call future.completeExceptionally() - log.warn("Failed to delete ledger {} of {}: {}", ledgerId, schemaId, rc); - } - if (numOfLedgerIds.decrementAndGet() == 0) { - try { - ZkUtils.deleteFullPathOptimistic(zooKeeper, getSchemaPath(schemaId), -1); - } catch (InterruptedException | KeeperException e) { - future.completeExceptionally(e); - } - future.complete(version); - } - }, null); - } - return future; - } else { - // It should never reach here - log.warn("No ledgers for schema id: {}", schemaId); - return completedFuture(version); - } + return deleteUsedLedgers(schemaId) + .thenCompose(ignored -> deleteUnusedLedgers(schemaId)) + .thenCompose(ignored -> deleteMetadata(schemaId)) + .thenApply(ignored -> -1L); } }); } @@ -575,10 +642,6 @@ private CompletableFuture createLedger(String schemaId) { if (rc != BKException.Code.OK) { future.completeExceptionally(bkException("Failed to create ledger", rc, -1, -1)); } else { - schemaLedgers.computeIfAbsent( - schemaId, - key -> Collections.synchronizedList(new ArrayList<>()) - ).add(handle.getId()); future.complete(handle); } }, null, metadata);