[improve][cli] topic offline internal-info provide schema ledger info

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -27,6 +27,7 @@
 import com.fasterxml.jackson.databind.ObjectReader;
 import com.github.zafarkhaja.semver.Version;
 import com.google.common.base.Throwables;
+import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import com.google.gson.Gson;
 import io.netty.buffer.ByteBuf;
@@ -88,6 +89,8 @@
 import org.apache.pulsar.broker.service.persistent.PersistentReplicator;
 import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage;
+import org.apache.pulsar.broker.stats.TopicInternalInfo;
 import org.apache.pulsar.broker.web.RestException;
 import org.apache.pulsar.client.admin.LongRunningProcessStatus;
 import org.apache.pulsar.client.admin.OffloadProcessStatus;
@@ -1402,20 +1405,19 @@ protected void internalGetManagedLedgerInfo(AsyncResponse asyncResponse, boolean
     }
 
     protected void internalGetManagedLedgerInfoForNonPartitionedTopic(AsyncResponse asyncResponse) {
+        CompletableFuture<ManagedLedgerInfo> mlFuture = new CompletableFuture<>();
         validateTopicOperationAsync(topicName, TopicOperation.GET_STATS)
                 .thenAccept(__ -> {
                     String managedLedger = topicName.getPersistenceNamingEncoding();
                     pulsar().getManagedLedgerFactory()
                             .asyncGetManagedLedgerInfo(managedLedger, new ManagedLedgerInfoCallback() {
                         @Override
                         public void getInfoComplete(ManagedLedgerInfo info, Object ctx) {
-                            asyncResponse.resume((StreamingOutput) output -> {
-                                objectWriter().writeValue(output, info);
-                            });
+                            mlFuture.complete(info);
                         }
                         @Override
                         public void getInfoFailed(ManagedLedgerException exception, Object ctx) {
-                            asyncResponse.resume(exception);
+                            mlFuture.completeExceptionally(exception);
                         }
                     }, null);
                 }).exceptionally(ex -> {
@@ -1424,6 +1426,22 @@ public void getInfoFailed(ManagedLedgerException exception, Object ctx) {
                     return null;
                 });
 
+        // Schema store ledgers
+        CompletableFuture<List<Long>> schemaFuture = ((BookkeeperSchemaStorage) pulsar().getSchemaStorage())
+                .getStoreLedgerIdsBySchemaId(topicName.getSchemaName());
+
+        FutureUtil.waitForAll(Lists.newArrayList(mlFuture, schemaFuture)).handle((res, ex) -> {
+            if (ex != null) {
+                asyncResponse.resume(ex);
+                return null;
+            }
+            TopicInternalInfo info = new TopicInternalInfo(mlFuture.getNow(null), schemaFuture.getNow(null));
+            asyncResponse.resume((StreamingOutput) output -> {
+                objectWriter().writeValue(output, info);
+            });
+            return null;
+        });
+
     }
 
     protected void internalGetPartitionedStats(AsyncResponse asyncResponse, boolean authoritative, boolean perPartition,

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/TopicInternalInfo.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.stats;
+
+import java.util.List;
+import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
+
+/**
+ * Persistent topic internal statistics.
+ */
+public class TopicInternalInfo extends ManagedLedgerInfo {
+
+    public List<Long> schemaLedgers;
+
+    public TopicInternalInfo() {
+        super();
+    }
+
+    public TopicInternalInfo(ManagedLedgerInfo info, List<Long> schemaLedgers) {
+        this.schemaLedgers = schemaLedgers;
+        if (info != null) {
+            version = info.version;
+            creationDate = info.creationDate;
+            modificationDate = info.modificationDate;
+            ledgers = info.ledgers;
+            terminatedPosition = info.terminatedPosition;
+            cursors = info.cursors;
+            properties = info.properties;
+        }
+    }
+
+}

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ClientGetSchemaTest.java

@@ -23,14 +23,15 @@
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotEquals;
 import static org.testng.Assert.assertNotNull;
-
+import static org.testng.Assert.assertFalse;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.Sets;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
-
 import java.util.function.Supplier;
 import lombok.Cleanup;
-
+import org.apache.pulsar.broker.stats.TopicInternalInfo;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.ProducerConsumerBase;
@@ -41,14 +42,13 @@
 import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.common.naming.TopicDomain;
 import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.apache.pulsar.schema.Schemas;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
-import com.google.common.collect.Sets;
-
 @Test(groups = "broker")
 public class ClientGetSchemaTest extends ProducerConsumerBase {
 
@@ -210,4 +210,51 @@ public void testAddProducerOnDeletedSchemaLedgerTopic() throws Exception {
 
         assertNotNull(producerWihtoutSchema);
     }
-}
+
+    /**
+     * Test verifies that broker should be able to handle non-recoverable error while deleting topic if topic's schema
+     * ledger doesn't exist.
+     *
+     * @throws Exception
+     */
+    @Test
+    public void testDeletedSchemaLedgerTopic() throws Exception {
+        final String tenant = PUBLIC_TENANT;
+        final String namespace = "test-namespace-" + randomName(16);
+        final String topicOne = "test-schema-info";
+        String fqtnOne = TopicName.get(TopicDomain.persistent.value(), tenant, namespace, topicOne).toString();
+
+        admin.namespaces().createNamespace(tenant + "/" + namespace, Sets.newHashSet("test"));
+
+        // (1) create topic with schema
+        Producer<Schemas.PersonTwo> producer = pulsarClient
+                .newProducer(Schema.AVRO(SchemaDefinition.<Schemas.PersonTwo> builder().withAlwaysAllowNull(false)
+                        .withSupportSchemaVersioning(true).withPojo(Schemas.PersonTwo.class).build()))
+                .topic(fqtnOne).create();
+
+        Consumer<Schemas.PersonTwo> consumer = pulsarClient
+                .newConsumer(Schema.AVRO(SchemaDefinition.<Schemas.PersonTwo> builder().withAlwaysAllowNull(false)
+                        .withSupportSchemaVersioning(true).withPojo(Schemas.PersonTwo.class).build()))
+                .subscriptionName("test").topic(fqtnOne).subscribe();
+
+        consumer.close();
+
+        String key = TopicName.get(fqtnOne).getSchemaName();
+        BookkeeperSchemaStorage schemaStrogate = (BookkeeperSchemaStorage) pulsar.getSchemaStorage();
+        long schemaLedgerId = schemaStrogate.getSchemaLedgerList(key).get(0);
+
+        // (2) break schema locator by deleting schema-ledger
+        schemaStrogate.getBookKeeper().deleteLedger(schemaLedgerId);
+
+        pulsar.getBrokerService().unloadNamespaceBundlesGracefully();
+
+        // fetch internal info with schema
+        String info = admin.topics().getInternalInfo(fqtnOne);
+
+        ObjectMapper objectMapper = ObjectMapperFactory.getMapper().getObjectMapper();
+        TopicInternalInfo infoObj = objectMapper.readValue(info, TopicInternalInfo.class);
+        assertFalse(infoObj.schemaLedgers.isEmpty());
+
+        producer.close();
+    }
+}