[feat][broker] Segmented transaction buffer snapshot segment and index system topic

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactory.java

@@ -116,6 +116,16 @@ ReadOnlyCursor openReadOnlyCursor(String managedLedgerName, Position startPositi
     void asyncOpenReadOnlyCursor(String managedLedgerName, Position startPosition, ManagedLedgerConfig config,
             OpenReadOnlyCursorCallback callback, Object ctx);
 
+    /**
+     * Asynchronous open a Read-only managedLedger.
+     * @param managedLedgerName the unique name that identifies the managed ledger
+     * @param config the managed ledegr configuratiion.
+     * @param callback callback object
+     * @param ctx opaque context
+     */
+    void asyncOpenReadOnlyManagedLedger(String managedLedgerName, ManagedLedgerConfig config,
+                                        OpenLedgerCallback callback, Object ctx);
+
     /**
      * Get the current metadata info for a managed ledger.
      *

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java

@@ -418,7 +418,33 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) {
         });
     }
 
+    @Override
+    public void asyncOpenReadOnlyManagedLedger(String managedLedgerName, ManagedLedgerConfig config,
+                                               OpenLedgerCallback callback, Object ctx) {
+        if (closed) {
+            callback.openLedgerFailed(new ManagedLedgerException.ManagedLedgerFactoryClosedException(), ctx);
+            return;
+        }
+        ReadOnlyManagedLedgerImpl roManagedLedger = new ReadOnlyManagedLedgerImpl(this,
+                bookkeeperFactory
+                        .get(new EnsemblePlacementPolicyConfig(config.getBookKeeperEnsemblePlacementPolicyClassName(),
+                                config.getBookKeeperEnsemblePlacementPolicyProperties())),
+                store, config, scheduledExecutor, managedLedgerName);
+
+        roManagedLedger.initialize(new ManagedLedgerInitializeLedgerCallback() {
+            @Override
+            public void initializeComplete() {
+                log.info("[{}] Successfully initialize Read-only managed ledger", managedLedgerName);
+                callback.openLedgerComplete(roManagedLedger, ctx);
+            }
 
+            @Override
+            public void initializeFailed(ManagedLedgerException e) {
+                log.error("[{}] Failed to initialize Read-only managed ledger", managedLedgerName, e);
+                callback.openLedgerFailed(e, ctx);
+            }
+        }, ctx);
+    }
 
     @Override
     public ReadOnlyCursor openReadOnlyCursor(String managedLedgerName, Position startPosition,

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -101,11 +101,12 @@
 import org.apache.pulsar.broker.rest.Topics;
 import org.apache.pulsar.broker.service.BrokerService;
 import org.apache.pulsar.broker.service.PulsarMetadataEventSynchronizer;
-import org.apache.pulsar.broker.service.SystemTopicBaseTxnBufferSnapshotService;
 import org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService;
 import org.apache.pulsar.broker.service.Topic;
 import org.apache.pulsar.broker.service.TopicPoliciesService;
+import org.apache.pulsar.broker.service.TransactionBufferSnapshotSegmentServiceImpl;
 import org.apache.pulsar.broker.service.TransactionBufferSnapshotService;
+import org.apache.pulsar.broker.service.TransactionBufferSnapshotServiceImpl;
 import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
 import org.apache.pulsar.broker.service.schema.SchemaStorageFactory;
 import org.apache.pulsar.broker.stats.MetricsGenerator;
@@ -261,7 +262,6 @@ public class PulsarService implements AutoCloseable, ShutdownService {
     private PulsarMetadataEventSynchronizer localMetadataSynchronizer;
     private CoordinationService coordinationService;
     private TransactionBufferSnapshotService transactionBufferSnapshotService;
-
     private MetadataStore configurationMetadataStore;
     private PulsarMetadataEventSynchronizer configMetadataSynchronizer;
     private boolean shouldShutdownConfigurationMetadataStore;
@@ -819,7 +819,11 @@ public void start() throws PulsarServerException {
 
             // Register pulsar system namespaces and start transaction meta store service
             if (config.isTransactionCoordinatorEnabled()) {
-                this.transactionBufferSnapshotService = new SystemTopicBaseTxnBufferSnapshotService(getClient());
+//                if (transactionBufferSegmentedSnapshotEnabled)
+                this.transactionBufferSnapshotService = new TransactionBufferSnapshotSegmentServiceImpl(getClient());
+//                  else
+                this.transactionBufferSnapshotService = new TransactionBufferSnapshotServiceImpl(getClient());
+
                 this.transactionTimer =
                         new HashedWheelTimer(new DefaultThreadFactory("pulsar-transaction-timer"));
                 transactionBufferClient = TransactionBufferClientImpl.create(this, transactionTimer,

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBaseTxnBufferSnapshotBaseService.java

@@ -0,0 +1,66 @@
+/**
+ * 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.service;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.broker.systopic.SystemTopicClient;
+import org.apache.pulsar.broker.systopic.SystemTopicClientBase;
+import org.apache.pulsar.common.naming.TopicName;
+
+public abstract class  SystemTopicBaseTxnBufferSnapshotBaseService<T> implements
+        SystemTopicTxnBufferSnapshotService<T> {
+
+    protected final Map<TopicName, SystemTopicClient<T>> clients;
+
+    public SystemTopicBaseTxnBufferSnapshotBaseService(
+            Map<TopicName, SystemTopicClient<T>> clients) {
+        this.clients = clients;
+    }
+
+    @Override
+    public CompletableFuture<SystemTopicClient.Writer<T>> createWriter(TopicName topicName) {
+        return getTransactionBufferSystemTopicClient(topicName).thenCompose(SystemTopicClient::newWriterAsync);
+
+    }
+
+    @Override
+    public CompletableFuture<SystemTopicClient.Reader<T>> createReader(TopicName topicName) {
+        return getTransactionBufferSystemTopicClient(topicName).thenCompose(SystemTopicClient::newReaderAsync);
+    }
+
+    @Override
+    public void removeClient(TopicName topicName, SystemTopicClientBase<T> transactionBufferSystemTopicClient) {
+        if (transactionBufferSystemTopicClient.getReaders().size() == 0
+                && transactionBufferSystemTopicClient.getWriters().size() == 0) {
+            clients.remove(topicName);
+        }
+    }
+
+    @Override
+    public void close() throws Exception {
+        for (Map.Entry<TopicName, SystemTopicClient<T>> entry : clients.entrySet()) {
+            entry.getValue().close();
+        }
+    }
+
+    protected abstract CompletableFuture<SystemTopicClient<T>> getTransactionBufferSystemTopicClient(
+            TopicName topicName);
+
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBaseTxnBufferSnapshotIndexService.java

@@ -0,0 +1,59 @@
+/**
+ * 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.service;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory;
+import org.apache.pulsar.broker.systopic.SystemTopicClient;
+import org.apache.pulsar.broker.transaction.buffer.matadata.v2.TransactionBufferSnapshotIndexes;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException.InvalidTopicNameException;
+import org.apache.pulsar.common.events.EventType;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.util.FutureUtil;
+
+public class SystemTopicBaseTxnBufferSnapshotIndexService extends
+        SystemTopicBaseTxnBufferSnapshotBaseService<TransactionBufferSnapshotIndexes> {
+
+    private final NamespaceEventsSystemTopicFactory namespaceEventsSystemTopicFactory;
+
+    public SystemTopicBaseTxnBufferSnapshotIndexService(PulsarClient client) {
+        super(new ConcurrentHashMap<>());
+        this.namespaceEventsSystemTopicFactory = new NamespaceEventsSystemTopicFactory(client);
+    }
+
+    @Override
+    protected CompletableFuture<SystemTopicClient<TransactionBufferSnapshotIndexes>>
+    getTransactionBufferSystemTopicClient(
+            TopicName topicName) {
+        TopicName systemTopicName = NamespaceEventsSystemTopicFactory
+                .getSystemTopicName(topicName.getNamespaceObject(), EventType.TRANSACTION_BUFFER_SNAPSHOT);
+        if (systemTopicName == null) {
+            return FutureUtil.failedFuture(
+                    new InvalidTopicNameException("Can't create SystemTopicBaseTxnBufferSnapshotIndexService, "
+                            + "because the topicName is null!"));
+        }
+        return CompletableFuture.completedFuture(clients.computeIfAbsent(systemTopicName,
+                (v) -> namespaceEventsSystemTopicFactory
+                        .createTransactionBufferSnapshotIndexSystemTopicClient(topicName.getNamespaceObject(),
+                                this)));
+    }
+
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBaseTxnBufferSnapshotSegmentService.java

@@ -0,0 +1,60 @@
+/**
+ * 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.service;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory;
+import org.apache.pulsar.broker.systopic.SystemTopicClient;
+import org.apache.pulsar.broker.transaction.buffer.matadata.v2.TransactionBufferSnapshotIndexes;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.events.EventType;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.util.FutureUtil;
+
+public class SystemTopicBaseTxnBufferSnapshotSegmentService extends
+        SystemTopicBaseTxnBufferSnapshotBaseService<TransactionBufferSnapshotIndexes.TransactionBufferSnapshot> {
+    private final NamespaceEventsSystemTopicFactory namespaceEventsSystemTopicFactory;
+
+    public SystemTopicBaseTxnBufferSnapshotSegmentService(PulsarClient client) {
+        super(new ConcurrentHashMap<>());
+        this.namespaceEventsSystemTopicFactory = new NamespaceEventsSystemTopicFactory(client);
+    }
+
+    @Override
+    protected CompletableFuture<SystemTopicClient<TransactionBufferSnapshotIndexes.TransactionBufferSnapshot>>
+    getTransactionBufferSystemTopicClient(
+            TopicName topicName) {
+        TopicName systemTopicName = NamespaceEventsSystemTopicFactory
+                .getSystemTopicName(topicName.getNamespaceObject(), EventType.TRANSACTION_BUFFER_SNAPSHOT_SEGMENT);
+        if (systemTopicName == null) {
+            return FutureUtil.failedFuture(
+                    new PulsarClientException.InvalidTopicNameException(
+                            "Can't create SystemTopicBaseTxnBufferSnapshotSegmentService, "
+                                    + "because the topicName is null!"));
+        }
+        return CompletableFuture.completedFuture(clients.computeIfAbsent(systemTopicName,
+                (v) -> namespaceEventsSystemTopicFactory
+                        .createTransactionBufferSnapshotSegmentSystemTopicClient(topicName.getNamespaceObject(),
+                                this)));
+    }
+
+
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBaseTxnBufferSnapshotService.java

@@ -18,38 +18,29 @@
  */
 package org.apache.pulsar.broker.service;
 
-import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory;
 import org.apache.pulsar.broker.systopic.SystemTopicClient;
-import org.apache.pulsar.broker.systopic.SystemTopicClient.Reader;
-import org.apache.pulsar.broker.systopic.SystemTopicClient.Writer;
-import org.apache.pulsar.broker.systopic.TransactionBufferSystemTopicClient;
 import org.apache.pulsar.broker.transaction.buffer.matadata.TransactionBufferSnapshot;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException.InvalidTopicNameException;
 import org.apache.pulsar.common.events.EventType;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.util.FutureUtil;
 
-public class SystemTopicBaseTxnBufferSnapshotService implements TransactionBufferSnapshotService {
-
-    private final Map<TopicName, SystemTopicClient<TransactionBufferSnapshot>> clients;
+public class SystemTopicBaseTxnBufferSnapshotService
+        extends SystemTopicBaseTxnBufferSnapshotBaseService<TransactionBufferSnapshot> {
 
     private final NamespaceEventsSystemTopicFactory namespaceEventsSystemTopicFactory;
 
     public SystemTopicBaseTxnBufferSnapshotService(PulsarClient client) {
+        super(new ConcurrentHashMap<>());
         this.namespaceEventsSystemTopicFactory = new NamespaceEventsSystemTopicFactory(client);
-        this.clients = new ConcurrentHashMap<>();
     }
 
     @Override
-    public CompletableFuture<Writer<TransactionBufferSnapshot>> createWriter(TopicName topicName) {
-        return getTransactionBufferSystemTopicClient(topicName).thenCompose(SystemTopicClient::newWriterAsync);
-    }
-
-    private CompletableFuture<SystemTopicClient<TransactionBufferSnapshot>> getTransactionBufferSystemTopicClient(
+    protected CompletableFuture<SystemTopicClient<TransactionBufferSnapshot>> getTransactionBufferSystemTopicClient(
             TopicName topicName) {
         TopicName systemTopicName = NamespaceEventsSystemTopicFactory
                 .getSystemTopicName(topicName.getNamespaceObject(), EventType.TRANSACTION_BUFFER_SNAPSHOT);
@@ -62,25 +53,4 @@ private CompletableFuture<SystemTopicClient<TransactionBufferSnapshot>> getTrans
                 (v) -> namespaceEventsSystemTopicFactory
                         .createTransactionBufferSystemTopicClient(topicName.getNamespaceObject(), this)));
     }
-
-    @Override
-    public CompletableFuture<Reader<TransactionBufferSnapshot>> createReader(TopicName topicName) {
-        return getTransactionBufferSystemTopicClient(topicName).thenCompose(SystemTopicClient::newReaderAsync);
-    }
-
-    @Override
-    public void removeClient(TopicName topicName,
-                                          TransactionBufferSystemTopicClient transactionBufferSystemTopicClient) {
-        if (transactionBufferSystemTopicClient.getReaders().size() == 0
-                && transactionBufferSystemTopicClient.getWriters().size() == 0) {
-            clients.remove(topicName);
-        }
-    }
-
-    @Override
-    public void close() throws Exception {
-        for (Map.Entry<TopicName, SystemTopicClient<TransactionBufferSnapshot>> entry : clients.entrySet()) {
-            entry.getValue().close();
-        }
-    }
 }