Skip to content

[WIP]Record unused schema ledgers to ZK for partitioned topics #8258

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.

Sign up for GitHub

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

Closed
BewareMyPower wants to merge 2 commits into from
Closed

[WIP]Record unused schema ledgers to ZK for partitioned topics #8258

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
362d92f
Record unused schema ledgers to ZK
BewareMyPower Oct 14, 2020
8d4f55a
Find schema ledgers from ZK instead of memory when delete a topic's s…
BewareMyPower Oct 14, 2020
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
181 changes: 137 additions & 44 deletions ...broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,10 +40,11 @@
import java.util.concurrent.ConcurrentHashMap;
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;

import org.apache.bookkeeper.client.AsyncCallback;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerEntry;
Expand All @@ -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;
Expand All @@ -81,9 +83,6 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
private final ServiceConfiguration config;
private BookKeeper bookKeeper;

// schemaId => ledgers of the schemaId
private final Map<String, List<Long>> schemaLedgers = new ConcurrentHashMap<>();

private final ConcurrentMap<String, CompletableFuture<StoredSchema>> readSchemaOperations = new ConcurrentHashMap<>();

@VisibleForTesting
Expand Down Expand Up @@ -298,20 +297,24 @@ private CompletableFuture<Long> putSchema(String schemaId, byte[] data, byte[] h
} else {
// No schema was defined yet
CompletableFuture<Long> 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 ||
ex.getCause() instanceof KeeperException.BadVersionException) {
// 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);
Expand All @@ -324,7 +327,27 @@ private CompletableFuture<Long> putSchema(String schemaId, byte[] data, byte[] h
});
}

private CompletableFuture<Long> createNewSchema(String schemaId, byte[] data, byte[] hash) {
@NotNull
private CompletableFuture<Void> 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<Void> 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<Long> createNewSchema(String schemaId, byte[] data, byte[] hash, AtomicLong ledgerId) {
SchemaStorageFormat.IndexEntry emptyIndex = SchemaStorageFormat.IndexEntry.newBuilder()
.setVersion(0)
.setHash(copyFrom(hash))
Expand All @@ -335,6 +358,7 @@ private CompletableFuture<Long> 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)
Expand All @@ -350,40 +374,108 @@ private CompletableFuture<Long> createNewSchema(String schemaId, byte[] data, by
});
}

/**
* Delete ledgers asynchronously and ignore the exception if failed to delete any ledger
*/
@NotNull
private CompletableFuture<Void> deleteLedgers(Stream<Long> ledgers, final int numOfLedgers) {
if (numOfLedgers == 0) {
return CompletableFuture.completedFuture(null);
}
CompletableFuture<Void> 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/<tenant>/<namespace>/<topic>`'s data
*/
@NotNull
private CompletableFuture<Void> deleteUsedLedgers(final String schemaId) {
CompletableFuture<Void> future = new CompletableFuture<>();
getSchemaLocator(getSchemaPath(schemaId)).whenComplete((optional, e) -> {
if (e != null) {
future.completeExceptionally(e);
return;
}
if (optional.isPresent()) {
List<SchemaStorageFormat.IndexEntry> 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/<tenant>/<namespace>/<topic>/unusedLedger`'s children
*/
@NotNull
private CompletableFuture<Void> deleteUnusedLedgers(final String schemaId) {
CompletableFuture<Void> 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<Void> deleteMetadata(final String schemaId) {
CompletableFuture<Void> 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<Long> 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<Long> ledgerIds = schemaLedgers.get(schemaId);
if (ledgerIds != null) {
CompletableFuture<Long> 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);
}
});
}
Expand All @@ -393,6 +485,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<SchemaStorageFormat.PositionInfo> addNewSchemaEntryToStore(
String schemaId,
Expand Down Expand Up @@ -545,10 +642,6 @@ private CompletableFuture<LedgerHandle> 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);
Expand Down
45 changes: 45 additions & 0 deletions ...er/src/test/java/org/apache/pulsar/broker/service/schema/PartitionedTopicsSchemaTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,17 +22,23 @@
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;
import org.apache.pulsar.client.api.Producer;
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 {
Expand Down Expand Up @@ -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<String> 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();
}

}