Support tiered storage on 2.9.2

bouncy-castle/bc/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>bouncy-castle-parent</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
     <relativePath>..</relativePath>
   </parent>
 

bouncy-castle/bcfips-include-test/pom.xml

@@ -24,7 +24,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>bouncy-castle-parent</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
     <relativePath>..</relativePath>
   </parent>
 

bouncy-castle/bcfips/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>bouncy-castle-parent</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
     <relativePath>..</relativePath>
   </parent>
 

bouncy-castle/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>pulsar</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
     <relativePath>..</relativePath>
   </parent>
 

buildtools/pom.xml

@@ -31,7 +31,7 @@
 
   <groupId>org.apache.pulsar</groupId>
   <artifactId>buildtools</artifactId>
-  <version>2.9.2</version>
+  <version>2.9.2-offload-SNAPSHOT</version>
   <packaging>jar</packaging>
   <name>Pulsar Build Tools</name>
 

deployment/terraform-ansible/deploy-pulsar.yaml

@@ -39,7 +39,7 @@
         zookeeper_servers: "{{ groups['zookeeper']|map('extract', hostvars, ['ansible_default_ipv4', 'address'])|map('regex_replace', '^(.*)$', '\\1:2181') | join(',') }}"
         service_url: "{{ pulsar_service_url }}"
         http_url: "{{ pulsar_web_url }}"
-        pulsar_version: "2.9.2"
+        pulsar_version: "2.9.2-offload-SNAPSHOT"
     - name: Download Pulsar binary package
       unarchive:
         src: https://www.apache.org/dyn/mirrors/mirrors.cgi?action=download&filename=pulsar/pulsar-{{ pulsar_version }}/apache-pulsar-{{ pulsar_version }}-bin.tar.gz

distribution/io/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>distribution</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
     <relativePath>..</relativePath>
   </parent>
 

distribution/offloaders/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>distribution</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
     <relativePath>..</relativePath>
   </parent>
 

distribution/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>pulsar</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
     <relativePath>..</relativePath>
   </parent>
 

distribution/server/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>distribution</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
     <relativePath>..</relativePath>
   </parent>
 

docker/grafana/pom.xml

@@ -23,7 +23,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>docker-images</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
   <artifactId>grafana-docker-image</artifactId>

docker/pom.xml

@@ -26,7 +26,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>pulsar</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
   </parent>
   <artifactId>docker-images</artifactId>
   <name>Apache Pulsar :: Docker Images</name>

docker/pulsar-all/pom.xml

@@ -23,7 +23,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>docker-images</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
   <artifactId>pulsar-all-docker-image</artifactId>

docker/pulsar/pom.xml

@@ -23,7 +23,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>docker-images</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
   <artifactId>pulsar-docker-image</artifactId>

jclouds-shaded/pom.xml

@@ -26,7 +26,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>pulsar</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
     <relativePath>..</relativePath>
   </parent>
 

kafka-connect-avro-converter-shaded/pom.xml

@@ -26,7 +26,7 @@
     <parent>
         <artifactId>pulsar</artifactId>
         <groupId>org.apache.pulsar</groupId>
-        <version>2.9.2</version>
+        <version>2.9.2-offload-SNAPSHOT</version>
         <relativePath>..</relativePath>
     </parent>
 

managed-ledger/pom.xml

@@ -25,7 +25,7 @@
   <parent>
     <groupId>org.apache.pulsar</groupId>
     <artifactId>pulsar</artifactId>
-    <version>2.9.2</version>
+    <version>2.9.2-offload-SNAPSHOT</version>
     <relativePath>..</relativePath>
   </parent>
 
@@ -60,6 +60,12 @@
       <version>${project.version}</version>
     </dependency>
 
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-broker-common</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>pulsar-metadata</artifactId>

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

@@ -143,6 +143,12 @@ interface OffloadCallback {
         void offloadFailed(ManagedLedgerException exception, Object ctx);
     }
 
+    interface OffloadServiceCallback {
+        void offloadComplete(Object ctx);
+
+        void offloadFailed(ManagedLedgerException exception, Object ctx);
+    }
+
     interface UpdatePropertiesCallback {
         void updatePropertiesComplete(Map<String, String> properties, Object ctx);
 

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

@@ -20,9 +20,15 @@
 
 import java.io.IOException;
 import java.util.Map;
+import org.apache.bookkeeper.client.api.BookKeeper;
 import org.apache.bookkeeper.common.annotation.InterfaceAudience.LimitedPrivate;
 import org.apache.bookkeeper.common.annotation.InterfaceStability.Evolving;
+import org.apache.bookkeeper.common.util.OrderedExecutor;
 import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
 import org.apache.pulsar.common.protocol.schema.SchemaStorage;
 
@@ -31,7 +37,7 @@
  */
 @LimitedPrivate
 @Evolving
-public interface LedgerOffloaderFactory<T extends LedgerOffloader> {
+public interface LedgerOffloaderFactory<T> {
 
     /**
      * Check whether the provided driver <tt>driverName</tt> is supported.
@@ -72,4 +78,17 @@ default T create(OffloadPoliciesImpl offloadPolicies,
             throws IOException {
         return create(offloadPolicies, userMetadata, scheduler);
     }
+
+
+    default T create(ServiceConfiguration conf,
+                     OffloadPoliciesImpl offloadPolicies,
+                     Map<String, String> userMetadata,
+                     PulsarClient pulsarClient,
+                     PulsarAdmin pulsarAdmin,
+                     BookKeeper bkc,
+                     OrderedExecutor executor,
+                     OrderedScheduler scheduler,
+                     StatsLogger statsLogger) throws IOException {
+        return create(offloadPolicies, userMetadata, scheduler);
+    }
 }

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

@@ -503,6 +503,16 @@ ManagedCursor newNonDurableCursor(Position startPosition, String subscriptionNam
      */
     void asyncOffloadPrefix(Position pos, OffloadCallback callback, Object ctx);
 
+    /**
+     * Start offload service.
+     * @param topicName
+     * @param operationType
+     * @param callback
+     * @param ctx
+     */
+    void asyncOffloadService(String topicName, String operationType,
+                             AsyncCallbacks.OffloadServiceCallback callback, Object ctx);
+
     /**
      * Get the slowest consumer.
      *

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

@@ -33,6 +33,7 @@
 import org.apache.bookkeeper.common.annotation.InterfaceStability;
 import org.apache.bookkeeper.mledger.impl.NullLedgerOffloader;
 
+import org.apache.bookkeeper.mledger.impl.NullOffloadService;
 import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
 import org.apache.pulsar.common.util.collections.ConcurrentOpenLongPairRangeSet;
 
@@ -74,6 +75,7 @@ public class ManagedLedgerConfig {
     private Class<? extends EnsemblePlacementPolicy>  bookKeeperEnsemblePlacementPolicyClassName;
     private Map<String, Object> bookKeeperEnsemblePlacementPolicyProperties;
     private LedgerOffloader ledgerOffloader = NullLedgerOffloader.INSTANCE;
+    private OffloadService offloadService = NullOffloadService.INSTANCE;
     private int newEntriesCheckDelayInMillis = 10;
     private Clock clock = Clock.systemUTC();
     private ManagedLedgerInterceptor managedLedgerInterceptor;
@@ -509,6 +511,28 @@ public ManagedLedgerConfig setLedgerOffloader(LedgerOffloader offloader) {
         return this;
     }
 
+    /**
+     * Get offload service which will be used to offload ledgers to longterm storage.
+     *
+     * The default offloader throws an exception on any attempt to offload.
+     *
+     * @return offloadService
+     */
+    public OffloadService getOffloadService() {
+        return offloadService;
+    }
+
+    /**
+     * Set offload service to use to offloading ledgers to long term storage.
+     *
+     * @param offloadService the offload service to use.
+     * @return
+     */
+    public ManagedLedgerConfig setOffloadService(OffloadService offloadService) {
+        this.offloadService = offloadService;
+        return this;
+    }
+
     /**
      * Get clock to use to time operations
      *

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

@@ -0,0 +1,80 @@
+/**
+ * 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.bookkeeper.mledger;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
+
+/**
+ * The offload service is used for offload the message from a topic.
+ * Each topic will have individual offload service.
+ */
+public interface OffloadService {
+
+    /**
+     * Initialize the offload service.
+     * @return
+     */
+    CompletableFuture<Void> initialize();
+
+    /**
+     * Trigger offload process for the topic. The offload process can keep running or stop in specific condition.
+     * @param topic
+     * @return
+     */
+    CompletableFuture<Void> offload(String topic);
+
+    /**
+     * Close the offload service.
+     * @return
+     */
+    CompletableFuture<Void> closeAsync();
+
+    String getOffloadDriverName();
+
+    /**
+     * Get offload policies of this LedgerOffloader
+     *
+     * @return offload policies
+     */
+    OffloadPoliciesImpl getOffloadPolicies();
+
+    /**
+     * Get related ledger's ReadHandle.
+     * @param ledgerId
+     * @return
+     */
+    CompletableFuture<ReadHandle> readOffloaded(long ledgerId, String managedLedgerName);
+
+    /**
+     * Get related ledger's ReadHandle.
+     * @param ledgerId
+     * @param executor
+     * @param conf
+     * @return
+     */
+    CompletableFuture<ReadHandle> readOffloaded(long ledgerId, String managedLedgerName,
+                                                OrderedExecutor executor, ServiceConfiguration conf);
+
+    void close();
+}
+