congbobo184 commented on code in PR #18293:
URL: https://github.com/apache/pulsar/pull/18293#discussion_r1015303193
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistry.java:
##########
@@ -37,6 +37,9 @@ public interface SchemaRegistry extends AutoCloseable {
CompletableFuture<SchemaVersion> putSchemaIfAbsent(String schemaId,
SchemaData schema,
SchemaCompatibilityStrategy strategy);
+ CompletableFuture<SchemaVersion> putSchemaIfAbsent(String schemaId,
SchemaData schema,
Review Comment:
don't need add a new interface
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java:
##########
@@ -284,56 +282,12 @@ private CompletableFuture<Long> putSchema(String
schemaId, byte[] data, byte[] h
return
readSchemaEntry(locator.getIndexList().get(0).getPosition())
.thenCompose(schemaEntry ->
addNewSchemaEntryToStore(schemaId,
locator.getIndexList(), data).thenCompose(
- position -> {
- CompletableFuture<Long> future = new
CompletableFuture<>();
- updateSchemaLocator(schemaId,
optLocatorEntry.get(), position, hash)
- .thenAccept(future::complete)
- .exceptionally(ex -> {
- if (ex.getCause() instanceof
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;
- })
+ position -> updateSchemaLocator(schemaId,
optLocatorEntry.get(), position, hash))
);
} else {
// No schema was defined yet
CompletableFuture<Long> future = new CompletableFuture<>();
Review Comment:
delete
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryStats.java:
##########
@@ -36,6 +36,7 @@ class SchemaRegistryStats implements AutoCloseable, Runnable {
private final Counter getOpsFailedCounter;
private final Counter putOpsFailedCounter;
+ private final Counter putOpsRetryCounter;
Review Comment:
if add this metrics, please add the test for it
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
