[fix][broker] Allow broker to handle non-recoverable schema error onl…

…y if SchemaLedgerForceRecovery flag is enabled (#23428)
apache · Oct 10, 2024 · bc3e7f6 · bc3e7f6
1 parent 5506f50
commit bc3e7f6
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 2 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -671,7 +671,8 @@ public CompletableFuture<Boolean> hasSchema() {
         return brokerService.pulsar().getSchemaRegistryService().getSchema(getSchemaId()).thenApply(Objects::nonNull)
                 .exceptionally(e -> {
                     Throwable ex = e.getCause();
-                    if (ex instanceof SchemaException || !((SchemaException) ex).isRecoverable()) {
+                    if (brokerService.pulsar().getConfig().isSchemaLedgerForceRecovery()
+                            && (ex instanceof SchemaException && !((SchemaException) ex).isRecoverable())) {
                         return false;
                     }
                     throw ex instanceof CompletionException ? (CompletionException) ex : new CompletionException(ex);

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ClientGetSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ClientGetSchemaTest.java
@@ -186,7 +186,7 @@ public void testAddProducerOnDeletedSchemaLedgerTopic() throws Exception {
         final String topicOne = "test-deleted-schema-ledger";
         final String fqtnOne = TopicName.get(TopicDomain.persistent.value(), tenant, namespace, topicOne).toString();
 
-        //pulsar.getConfig().setManagedLedgerForceRecovery(true);
+        pulsar.getConfig().setSchemaLedgerForceRecovery(true);
         admin.namespaces().createNamespace(tenant + "/" + namespace, Sets.newHashSet("test"));
 
         // (1) create topic with schema