Handle KeeperException.BadVersionException thrown by updateSchemaLocator()
@@ -264,7 +264,29 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
//don't check the schema whether already exist
return readSchemaEntry(locator.getIndexList().get(0).getPosition())
.thenCompose(schemaEntry -> addNewSchemaEntryToStore(schemaId, locator.getIndexList(), data).thenCompose(
- position -> updateSchemaLocator(schemaId, optLocatorEntry.get(), position, hash)));
+ position -> {
+ CompletableFuture<Long> future = new CompletableFuture<>();
+ updateSchemaLocator(schemaId, optLocatorEntry.get(), position, hash)
+ .thenAccept(future::complete)
+ .exceptionally(ex -> {
+ if (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
+ putSchema(schemaId, data, hash)
+ .thenAccept(future::complete)
+ .exceptionally(ex2 -> {
+ future.completeExceptionally(ex2);
+ return null;
+ });
+ } else {
+ // For other errors, just fail the operation
+ future.completeExceptionally(ex);
+ }
+ return null;
+ });
+ return future;
+ })
+ );
} else {
// No schema was defined yet
CompletableFuture<Long> future = new CompletableFuture<>();
