From d6cae39dbcd03b30ee0b296169b05ec32f289963 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 7 Nov 2018 19:02:29 -0800 Subject: [PATCH 1/3] Fixed race condition in schema initialization in partitioned topics --- .../schema/BookkeeperSchemaStorage.java | 222 ++++++++++++------ .../schema/PartitionedTopicsSchemaTest.java | 4 +- .../client/impl/PartitionedProducerImpl.java | 4 +- 3 files changed, 153 insertions(+), 77 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 b8d9a27fe7a70..045a5ab4452df 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,8 @@ import javax.validation.constraints.NotNull; +import lombok.extern.slf4j.Slf4j; + import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.LedgerEntry; @@ -53,10 +55,12 @@ import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.Code; +import org.apache.zookeeper.KeeperException.NodeExistsException; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.ACL; +@Slf4j public class BookkeeperSchemaStorage implements SchemaStorage { private static final String SchemaPath = "/schemas"; private static final List Acl = ZooDefs.Ids.OPEN_ACL_UNSAFE; @@ -68,9 +72,6 @@ public class BookkeeperSchemaStorage implements SchemaStorage { private final ServiceConfiguration config; private BookKeeper bookKeeper; - - private final ConcurrentMap> locatorEntries = new ConcurrentHashMap<>(); - private final ConcurrentMap> readSchemaOperations = new ConcurrentHashMap<>(); @VisibleForTesting @@ -124,9 +125,15 @@ public CompletableFuture delete(String key) { private CompletableFuture getSchema(String schemaId) { // There's already a schema read operation in progress. Just piggyback on that return readSchemaOperations.computeIfAbsent(schemaId, key -> { + if (log.isDebugEnabled()) { + log.debug("[{}] Fetching schema from store", schemaId); + } CompletableFuture future = new CompletableFuture<>(); getSchemaLocator(getSchemaPath(schemaId)).thenCompose(locator -> { + if (log.isDebugEnabled()) { + log.debug("[{}] Got schema locator {}", schemaId, locator); + } if (!locator.isPresent()) { return completedFuture(null); } @@ -166,6 +173,9 @@ public void close() throws Exception { @NotNull private CompletableFuture getSchema(String schemaId, long version) { return getSchemaLocator(getSchemaPath(schemaId)).thenCompose(locator -> { + if (log.isDebugEnabled()) { + log.debug("[{}] Get schema - version: {} - locator: {}", schemaId, version, locator); + } if (!locator.isPresent()) { return completedFuture(null); @@ -188,29 +198,111 @@ private CompletableFuture getSchema(String schemaId, long version) @NotNull private CompletableFuture putSchema(String schemaId, byte[] data, byte[] hash) { - return getOrCreateSchemaLocator(getSchemaPath(schemaId)).thenCompose(locatorEntry -> - addNewSchemaEntryToStore(locatorEntry.locator.getIndexList(), data).thenCompose(position -> - updateSchemaLocator(schemaId, locatorEntry, position, hash) - ) - ); + return getSchemaLocator(getSchemaPath(schemaId)).thenCompose(optLocatorEntry -> { + if (optLocatorEntry.isPresent()) { + // Schema locator was already present + return addNewSchemaEntryToStore(optLocatorEntry.get().locator.getIndexList(), data) + .thenCompose(position -> updateSchemaLocator(schemaId, optLocatorEntry.get(), position, hash)); + } else { + // No schema was defined yet + CompletableFuture future = new CompletableFuture<>(); + createNewSchema(schemaId, data, hash) + .thenAccept(version -> future.complete(version)) + .exceptionally(ex -> { + if (ex.getCause() instanceof NodeExistsException) { + // 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(version -> future.complete(version)) + .exceptionally(ex2 -> { + future.completeExceptionally(ex2); + return null; + }); + } else { + // For other errors, just fail the operation + future.completeExceptionally(ex); + } + + return null; + }); + + return future; + } + }); } @NotNull private CompletableFuture putSchemaIfAbsent(String schemaId, byte[] data, byte[] hash) { - return getOrCreateSchemaLocator(getSchemaPath(schemaId)).thenCompose(locatorEntry -> { - byte[] storedHash = locatorEntry.locator.getInfo().getHash().toByteArray(); - if (storedHash.length > 0 && Arrays.equals(storedHash, hash)) { - return completedFuture(locatorEntry.locator.getInfo().getVersion()); - } - return findSchemaEntryByHash(locatorEntry.locator.getIndexList(), hash).thenCompose(version -> { - if (isNull(version)) { - return addNewSchemaEntryToStore(locatorEntry.locator.getIndexList(), data).thenCompose(position -> - updateSchemaLocator(schemaId, locatorEntry, position, hash) - ); - } else { - return completedFuture(version); + return getSchemaLocator(getSchemaPath(schemaId)).thenCompose(optLocatorEntry -> { + + if (optLocatorEntry.isPresent()) { + // Schema locator was already present + SchemaStorageFormat.SchemaLocator locator = optLocatorEntry.get().locator; + byte[] storedHash = locator.getInfo().getHash().toByteArray(); + if (storedHash.length > 0 && Arrays.equals(storedHash, hash)) { + return completedFuture(locator.getInfo().getVersion()); } - }); + return findSchemaEntryByHash(locator.getIndexList(), hash).thenCompose(version -> { + if (isNull(version)) { + return addNewSchemaEntryToStore(locator.getIndexList(), data).thenCompose( + position -> updateSchemaLocator(schemaId, optLocatorEntry.get(), position, hash)); + } else { + return completedFuture(version); + } + }); + } else { + // No schema was defined yet + CompletableFuture future = new CompletableFuture<>(); + createNewSchema(schemaId, data, hash) + .thenAccept(version -> future.complete(version)) + .exceptionally(ex -> { + if (ex.getCause() instanceof NodeExistsException) { + // 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 + putSchemaIfAbsent(schemaId, data, hash) + .thenAccept(version -> future.complete(version)) + .exceptionally(ex2 -> { + future.completeExceptionally(ex2); + return null; + }); + } else { + // For other errors, just fail the operation + future.completeExceptionally(ex); + } + + return null; + }); + + return future; + } + }); + } + + private CompletableFuture createNewSchema(String schemaId, byte[] data, byte[] hash) { + SchemaStorageFormat.IndexEntry emptyIndex = SchemaStorageFormat.IndexEntry.newBuilder() + .setVersion(-1L) + .setHash(ByteString.EMPTY) + .setPosition(SchemaStorageFormat.PositionInfo.newBuilder() + .setEntryId(-1L) + .setLedgerId(-1L) + ).build(); + + return addNewSchemaEntryToStore(Collections.singletonList(emptyIndex), data).thenCompose(position -> { + // The schema was stored in the ledger, now update the z-node with the pointer to it + SchemaStorageFormat.IndexEntry info = SchemaStorageFormat.IndexEntry.newBuilder() + .setVersion(0) + .setPosition(position) + .setHash(copyFrom(hash)) + .build(); + + return createSchemaLocator(getSchemaPath(schemaId), SchemaStorageFormat.SchemaLocator.newBuilder() + .setInfo(info) + .addAllIndex( + newArrayList(info)) + .build()) + .thenApply(ignore -> 0L); }); } @@ -226,7 +318,7 @@ private CompletableFuture deleteSchema(String schemaId) { } @NotNull - private String getSchemaPath(String schemaId) { + private static String getSchemaPath(String schemaId) { return SchemaPath + "/" + schemaId; } @@ -319,6 +411,10 @@ private CompletableFuture findSchemaEntryByHash( private CompletableFuture readSchemaEntry( SchemaStorageFormat.PositionInfo position ) { + if (log.isDebugEnabled()) { + log.debug("Reading schema entry from {}", position); + } + return openLedger(position.getLedgerId()) .thenCompose((ledger) -> Functions.getLedgerEntry(ledger, position.getEntryId()) @@ -342,6 +438,24 @@ private CompletableFuture updateSchemaLocator(String id, SchemaStorageForm return future; } + @NotNull + private CompletableFuture createSchemaLocator(String id, SchemaStorageFormat.SchemaLocator locator) { + CompletableFuture future = new CompletableFuture<>(); + + ZkUtils.asyncCreateFullPathOptimistic(zooKeeper, id, locator.toByteArray(), Acl, + CreateMode.PERSISTENT, (rc, path, ctx, name) -> { + Code code = Code.get(rc); + if (code != Code.OK) { + future.completeExceptionally(KeeperException.create(code)); + } else { + // Newly created z-node will have version 0 + future.complete(new LocatorEntry(locator, 0)); + } + }, null); + + return future; + } + @NotNull private CompletableFuture> getSchemaLocator(String schema) { return localZkCache.getEntryAsync(schema, new SchemaLocatorDeserializer()).thenApply(optional -> @@ -349,58 +463,13 @@ private CompletableFuture> getSchemaLocator(String schema ); } - @NotNull - private CompletableFuture getOrCreateSchemaLocator(String schema) { - // Protect from concurrent schema locator creation - return locatorEntries.computeIfAbsent(schema, key -> { - CompletableFuture future = new CompletableFuture<>(); - - getSchemaLocator(schema).thenCompose(schemaLocatorStatEntry -> { - if (schemaLocatorStatEntry.isPresent()) { - return completedFuture(schemaLocatorStatEntry.get()); - } else { - SchemaStorageFormat.SchemaLocator locator = SchemaStorageFormat.SchemaLocator.newBuilder() - .setInfo(SchemaStorageFormat.IndexEntry.newBuilder().setVersion(-1L) - .setHash(ByteString.EMPTY).setPosition(SchemaStorageFormat.PositionInfo.newBuilder() - .setEntryId(-1L).setLedgerId(-1L))) - .build(); - - CompletableFuture zkFuture = new CompletableFuture<>(); - - ZkUtils.asyncCreateFullPathOptimistic(zooKeeper, schema, locator.toByteArray(), Acl, - CreateMode.PERSISTENT, (rc, path, ctx, name) -> { - Code code = Code.get(rc); - if (code != Code.OK) { - zkFuture.completeExceptionally(KeeperException.create(code)); - } else { - zkFuture.complete(new LocatorEntry(locator, -1)); - } - }, null); - - return zkFuture; - } - }).handleAsync((res, ex) -> { - // Cleanup the pending ops from the map - locatorEntries.remove(schema, future); - if (ex != null) { - future.completeExceptionally(ex); - } else { - future.complete(res); - } - return null; - }); - - return future; - }); - } - @NotNull private CompletableFuture addEntry(LedgerHandle ledgerHandle, SchemaStorageFormat.SchemaEntry entry) { final CompletableFuture future = new CompletableFuture<>(); ledgerHandle.asyncAddEntry(entry.toByteArray(), (rc, handle, entryId, ctx) -> { if (rc != BKException.Code.OK) { - future.completeExceptionally(BKException.create(rc)); + future.completeExceptionally(bkException("Failed to add entry", rc, ledgerHandle.getId(), -1)); } else { future.complete(entryId); } @@ -420,7 +489,7 @@ private CompletableFuture createLedger() { LedgerPassword, (rc, handle, ctx) -> { if (rc != BKException.Code.OK) { - future.completeExceptionally(BKException.create(rc)); + future.completeExceptionally(bkException("Failed to create ledger", rc, -1, -1)); } else { future.complete(handle); } @@ -438,7 +507,7 @@ private CompletableFuture openLedger(Long ledgerId) { LedgerPassword, (rc, handle, ctx) -> { if (rc != BKException.Code.OK) { - future.completeExceptionally(BKException.create(rc)); + future.completeExceptionally(bkException("Failed to open ledger", rc, ledgerId, -1)); } else { future.complete(handle); } @@ -452,7 +521,7 @@ private CompletableFuture closeLedger(LedgerHandle ledgerHandle) { CompletableFuture future = new CompletableFuture<>(); ledgerHandle.asyncClose((rc, handle, ctx) -> { if (rc != BKException.Code.OK) { - future.completeExceptionally(BKException.create(rc)); + future.completeExceptionally(bkException("Failed to close ledger", rc, ledgerHandle.getId(), -1)); } else { future.complete(null); } @@ -466,7 +535,7 @@ static CompletableFuture getLedgerEntry(LedgerHandle ledger, long e ledger.asyncReadEntries(entry, entry, (rc, handle, entries, ctx) -> { if (rc != BKException.Code.OK) { - future.completeExceptionally(BKException.create(rc)); + future.completeExceptionally(bkException("Failed to read entry", rc, ledger.getId(), entry)); } else { future.complete(entries.nextElement()); } @@ -519,4 +588,13 @@ static class LocatorEntry { this.zkZnodeVersion = zkZnodeVersion; } } + + public static Exception bkException(String operation, int rc, long ledgerId, long entryId) { + String message = org.apache.bookkeeper.client.api.BKException.getMessage(rc) + " - ledger=" + ledgerId; + + if (entryId != -1) { + message += " - entry=" + entryId; + } + return new IOException(message); + } } 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 e7723d7754b17..6e9b1220a5fd9 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 @@ -38,10 +38,8 @@ public class PartitionedTopicsSchemaTest extends BrokerBkEnsemblesTests { /** * Test that sequence id from a producer is correct when there are send errors - * - * the test is disabled {@link https://github.com/apache/pulsar/issues/2651} */ - @Test(enabled = false) + @Test public void partitionedTopicWithSchema() throws Exception { admin.namespaces().createNamespace("prop/my-test", Collections.singleton("usc")); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java index a6f40fe1c3bbd..3b87e4e086004 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java @@ -126,15 +126,15 @@ private void start() { if (completed.incrementAndGet() == topicMetadata.numPartitions()) { if (createFail.get() == null) { setState(State.Ready); - producerCreatedFuture().complete(PartitionedProducerImpl.this); log.info("[{}] Created partitioned producer", topic); + producerCreatedFuture().complete(PartitionedProducerImpl.this); } else { + log.error("[{}] Could not create partitioned producer.", topic, createFail.get().getCause()); closeAsync().handle((ok, closeException) -> { producerCreatedFuture().completeExceptionally(createFail.get()); client.cleanupProducer(this); return null; }); - log.error("[{}] Could not create partitioned producer.", topic, createFail.get().getCause()); } } From 9159d86ab53fa7663e5d747a3f7c52c60cc910ed Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 7 Nov 2018 19:18:59 -0800 Subject: [PATCH 2/3] Removed lombok log --- .../broker/service/schema/BookkeeperSchemaStorage.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 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 045a5ab4452df..97f01180fb0a7 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,8 +41,6 @@ import javax.validation.constraints.NotNull; -import lombok.extern.slf4j.Slf4j; - import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.LedgerEntry; @@ -59,9 +57,12 @@ import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.ACL; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; -@Slf4j public class BookkeeperSchemaStorage implements SchemaStorage { + private static final Logger log = LoggerFactory.getLogger(BookkeeperSchemaStorage.class); + private static final String SchemaPath = "/schemas"; private static final List Acl = ZooDefs.Ids.OPEN_ACL_UNSAFE; private static final byte[] LedgerPassword = "".getBytes(); From b75284b9e86588703b53f2bf22ed4fc453ca4771 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 7 Nov 2018 22:49:45 -0800 Subject: [PATCH 3/3] Fixed tests --- .../schema/BookkeeperSchemaStorage.java | 25 ++++++++++++++++--- 1 file changed, 21 insertions(+), 4 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 97f01180fb0a7..f0e9699dd406c 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 @@ -144,6 +144,10 @@ private CompletableFuture getSchema(String schemaId) { .thenApply(entry -> new StoredSchema(entry.getSchemaData().toByteArray(), new LongSchemaVersion(schemaLocator.getInfo().getVersion()))); }).handleAsync((res, ex) -> { + if (log.isDebugEnabled()) { + log.debug("[{}] Get operation completed. res={} -- ex={}", schemaId, res, ex); + } + // Cleanup the pending ops from the map readSchemaOperations.remove(schemaId, future); if (ex != null) { @@ -173,6 +177,10 @@ public void close() throws Exception { @NotNull private CompletableFuture getSchema(String schemaId, long version) { + if (log.isDebugEnabled()) { + log.debug("[{}] Get schema - version: {}", schemaId, version); + } + return getSchemaLocator(getSchemaPath(schemaId)).thenCompose(locator -> { if (log.isDebugEnabled()) { log.debug("[{}] Get schema - version: {} - locator: {}", schemaId, version, locator); @@ -244,6 +252,11 @@ private CompletableFuture putSchemaIfAbsent(String schemaId, byte[] data, if (storedHash.length > 0 && Arrays.equals(storedHash, hash)) { return completedFuture(locator.getInfo().getVersion()); } + + if (log.isDebugEnabled()) { + log.debug("[{}] findSchemaEntryByHash - hash={}", schemaId, hash); + } + return findSchemaEntryByHash(locator.getIndexList(), hash).thenCompose(version -> { if (isNull(version)) { return addNewSchemaEntryToStore(locator.getIndexList(), data).thenCompose( @@ -283,8 +296,8 @@ private CompletableFuture putSchemaIfAbsent(String schemaId, byte[] data, private CompletableFuture createNewSchema(String schemaId, byte[] data, byte[] hash) { SchemaStorageFormat.IndexEntry emptyIndex = SchemaStorageFormat.IndexEntry.newBuilder() - .setVersion(-1L) - .setHash(ByteString.EMPTY) + .setVersion(0) + .setHash(copyFrom(hash)) .setPosition(SchemaStorageFormat.PositionInfo.newBuilder() .setEntryId(-1L) .setLedgerId(-1L) @@ -403,8 +416,12 @@ private CompletableFuture findSchemaEntryByHash( } } - return readSchemaEntry(index.get(0).getPosition()) - .thenCompose(entry -> findSchemaEntryByHash(entry.getIndexList(), hash)); + if (index.get(0).getPosition().getLedgerId() == -1) { + return completedFuture(null); + } else { + return readSchemaEntry(index.get(0).getPosition()) + .thenCompose(entry -> findSchemaEntryByHash(entry.getIndexList(), hash)); + } }