[feat] PIP-454: Metadata Store Migration Framework (implementation)

distribution/server/src/assemble/LICENSE.bin.txt

@@ -498,8 +498,8 @@ The Apache Software License, Version 2.0
   * Prometheus
     - io.prometheus-simpleclient_httpserver-0.16.0.jar
   * Oxia
-    - io.github.oxia-db-oxia-client-api-0.7.2.jar
-    - io.github.oxia-db-oxia-client-0.7.2.jar
+    - io.github.oxia-db-oxia-client-api-0.7.4.jar
+    - io.github.oxia-db-oxia-client-0.7.4.jar
   * OpenHFT
     - net.openhft-zero-allocation-hashing-0.16.jar
   * Java JSON WebTokens

pom.xml

@@ -299,7 +299,7 @@ flexible messaging model and an intuitive client API.</description>
     <apache-http-client.version>4.5.13</apache-http-client.version>
     <apache-httpcomponents.version>4.4.15</apache-httpcomponents.version>
     <jetcd.version>0.7.7</jetcd.version>
-    <oxia.version>0.7.2</oxia.version>
+    <oxia.version>0.7.4</oxia.version>
     <snakeyaml.version>2.0</snakeyaml.version>
     <ant.version>1.10.12</ant.version>
     <seancfoley.ipaddress.version>5.5.0</seancfoley.ipaddress.version>

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

@@ -53,6 +53,7 @@
 import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
 import org.apache.pulsar.metadata.bookkeeper.PulsarMetadataBookieDriver;
 import org.apache.pulsar.metadata.bookkeeper.PulsarMetadataClientDriver;
+import org.apache.pulsar.metadata.impl.DualMetadataStore;
 import org.apache.pulsar.metadata.impl.MetadataStoreFactoryImpl;
 import org.apache.pulsar.metadata.impl.ZKMetadataStore;
 import org.slf4j.Logger;
@@ -315,7 +316,7 @@ private static void initializeCluster(Arguments arguments, int bundleNumberForDe
             }
         }
 
-        if (localStore instanceof ZKMetadataStore && configStore instanceof ZKMetadataStore) {
+        if (localStore instanceof DualMetadataStore && configStore instanceof DualMetadataStore) {
             String uriStr;
             if (arguments.existingBkMetadataServiceUri != null) {
                 uriStr = arguments.existingBkMetadataServiceUri;

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

@@ -0,0 +1,116 @@
+/*
+ * 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.admin.impl;
+
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiParam;
+import io.swagger.annotations.ApiResponse;
+import io.swagger.annotations.ApiResponses;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.Response;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.admin.AdminResource;
+import org.apache.pulsar.broker.web.RestException;
+import org.apache.pulsar.common.migration.MigrationState;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.apache.pulsar.metadata.coordination.impl.MigrationCoordinator;
+import org.apache.pulsar.metadata.impl.DualMetadataStore;
+
+/**
+ * Admin resource for metadata store migration operations.
+ */
+@Slf4j
+public class MetadataMigrationBase extends AdminResource {
+
+    @GET
+    @Path("/status")
+    @ApiOperation(value = "Get current migration status", response = MigrationState.class)
+    @ApiResponses(value = {
+            @ApiResponse(code = 200, message = "Migration status retrieved successfully"),
+            @ApiResponse(code = 500, message = "Internal server error")
+    })
+    public MigrationState getStatus() {
+        validateSuperUserAccess();
+
+        try {
+            var ogr = pulsar().getLocalMetadataStore().get(MigrationState.MIGRATION_FLAG_PATH).get();
+            if (ogr.isPresent()) {
+                return ObjectMapperFactory.getMapper().reader().readValue(ogr.get().getValue(), MigrationState.class);
+            } else {
+                return MigrationState.NOT_STARTED;
+            }
+        } catch (Exception e) {
+            log.error("Failed to get migration status", e);
+            throw new RestException(e);
+        }
+    }
+
+    @POST
+    @Path("/start")
+    @ApiOperation(value = "Start metadata store migration")
+    @ApiResponses(value = {
+            @ApiResponse(code = 204, message = "Migration started successfully"),
+            @ApiResponse(code = 400, message = "Invalid target URL"),
+            @ApiResponse(code = 409, message = "Migration already in progress"),
+            @ApiResponse(code = 500, message = "Internal server error")
+    })
+    public void startMigration(
+            @ApiParam(value = "Target metadata store URL", required = true)
+            @QueryParam("target")
+            String targetUrl) {
+        validateSuperUserAccess();
+
+        if (targetUrl == null || targetUrl.trim().isEmpty()) {
+            throw new RestException(Response.Status.BAD_REQUEST, "Target URL is required");
+        }
+
+        try {
+            // Check if metadata store is wrapped with DualMetadataStore
+            if (!(pulsar().getLocalMetadataStore() instanceof DualMetadataStore)) {
+                throw new RestException(Response.Status.BAD_REQUEST, "Metadata store is not configured for migration. "
+                        + "Please ensure you're using a supported source metadata store (e.g., ZooKeeper).");
+            }
+
+            // Create coordinator
+            MigrationCoordinator coordinator = new MigrationCoordinator(pulsar().getLocalMetadataStore(), targetUrl);
+
+            // Start migration in background thread
+            pulsar().getExecutor().submit(() -> {
+                try {
+                    log.info("Starting metadata migration to: {}", targetUrl);
+                    coordinator.startMigration();
+                    log.info("Metadata migration completed successfully");
+                } catch (Exception e) {
+                    log.error("Metadata migration failed", e);
+                }
+            });
+
+            log.info("Migration initiated to target: {}", targetUrl);
+
+        } catch (RestException e) {
+            throw e;
+        } catch (Exception e) {
+            log.error("Failed to start migration", e);
+            throw new RestException(e);
+        }
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/MetadataMigration.java

@@ -0,0 +1,34 @@
+/*
+ * 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.admin.v2;
+
+import io.swagger.annotations.Api;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+import org.apache.pulsar.broker.admin.impl.MetadataMigrationBase;
+
+/**
+ * REST API for metadata store migration operations.
+ */
+@Path("/metadata/migration")
+@Api(value = "/metadata/migration", description = "Metadata store migration admin APIs", tags = "metadata-migration")
+@Produces(MediaType.APPLICATION_JSON)
+public class MetadataMigration extends MetadataMigrationBase {
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateMetadataStoreTableViewImpl.java

@@ -36,7 +36,7 @@
 import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.metadata.api.MetadataStoreException;
 import org.apache.pulsar.metadata.api.MetadataStoreTableView;
-import org.apache.pulsar.metadata.impl.AbstractMetadataStore;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
 import org.apache.pulsar.metadata.tableview.impl.MetadataStoreTableViewImpl;
 
 @Slf4j
@@ -65,7 +65,7 @@ public void start(PulsarService pulsar,
         init(pulsar);
         conflictResolver = new ServiceUnitStateDataConflictResolver();
         conflictResolver.setStorageType(MetadataStore);
-        if (!(pulsar.getLocalMetadataStore() instanceof AbstractMetadataStore)
+        if (!(pulsar.getLocalMetadataStore() instanceof MetadataStoreExtended)
             && !MetadataSessionExpiredPolicy.shutdown.equals(pulsar.getConfig().getZookeeperSessionExpiredPolicy())) {
             String errorMsg = String.format("Your current metadata store [%s] does not support the registration of "
                     + "session event listeners. Please set \"zookeeperSessionExpiredPolicy\" to \"shutdown\";"

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/CanReconnectZKClientPulsarServiceBaseTest.java

@@ -35,7 +35,9 @@
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
 import org.apache.pulsar.common.policies.data.TopicType;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
 import org.apache.pulsar.metadata.api.extended.SessionEvent;
+import org.apache.pulsar.metadata.impl.DualMetadataStore;
 import org.apache.pulsar.metadata.impl.ZKMetadataStore;
 import org.apache.pulsar.tests.TestRetrySupport;
 import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
@@ -88,9 +90,11 @@ protected void startBrokers() throws Exception {
         pulsar = new PulsarService(config);
         pulsar.start();
         broker = pulsar.getBrokerService();
-        ZKMetadataStore zkMetadataStore = (ZKMetadataStore) pulsar.getLocalMetadataStore();
-        localZkOfBroker = zkMetadataStore.getZkClient();
-        zkMetadataStore.registerSessionListener(n -> {
+        MetadataStoreExtended store = pulsar.getLocalMetadataStore();
+        if (store instanceof DualMetadataStore dms) {
+            localZkOfBroker = ((ZKMetadataStore) dms.getSourceStore()).getZkClient();
+        }
+        store.registerSessionListener(n -> {
             log.info("Received session event: {}", n);
             sessionEvent = n;
         });

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java

@@ -116,6 +116,7 @@
 import org.apache.pulsar.common.schema.SchemaInfo;
 import org.apache.pulsar.common.schema.SchemaType;
 import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.metadata.impl.DualMetadataStore;
 import org.awaitility.Awaitility;
 import org.awaitility.reflect.WhiteboxImpl;
 import org.glassfish.jersey.client.JerseyClient;
@@ -1547,8 +1548,9 @@ public void testCloseTopicAfterStartReplicationFailed() throws Exception {
                 (PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
 
         // We inject an error to make "start replicator" to fail.
+        DualMetadataStore dms = (DualMetadataStore) pulsar1.getConfigurationMetadataStore();
         AsyncLoadingCache<String, Boolean> existsCache =
-                WhiteboxImpl.getInternalState(pulsar1.getConfigurationMetadataStore(), "existsCache");
+                WhiteboxImpl.getInternalState(dms.getSourceStore(), "existsCache");
         String path = "/admin/partitioned-topics/" + TopicName.get(topicName).getPersistenceNamingEncoding();
         existsCache.put(path, CompletableFuture.completedFuture(true));
 

pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/MetadataMigration.java

@@ -0,0 +1,42 @@
+/*
+ * 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.client.admin;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.common.migration.MigrationState;
+
+/**
+ * Handle cluster metadata migrations.
+ */
+public interface MetadataMigration {
+
+    /**
+     * Start metadata store migration.
+     *
+     * @return
+     */
+    CompletableFuture<Void> start(String targetUrl);
+
+    /**
+     * Get current migration status.
+     *
+     * @return
+     */
+    CompletableFuture<MigrationState> status();
+}

pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java

@@ -174,6 +174,8 @@ static PulsarAdminBuilder builder() {
      */
     Transactions transactions();
 
+    MetadataMigration metadataMigration();
+
     /**
      * Close the PulsarAdminClient and release all the resources.
      *

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/migration/MigrationPhase.java

@@ -0,0 +1,52 @@
+/*
+ * 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.common.migration;
+
+/**
+ * Represents the different phases of metadata store migration.
+ */
+public enum MigrationPhase {
+    /**
+     * No migration in progress. Operating normally on source store only.
+     */
+    NOT_STARTED,
+
+    /**
+     * Migration preparation phase. All brokers and bookies are recreating
+     * their ephemeral nodes in the target store.
+     */
+    PREPARATION,
+
+    /**
+     * Data copy phase. The migration coordinator is copying persistent
+     * data from source to target store.
+     */
+    COPYING,
+
+    /**
+     * Migration completed. All services are using target store. Source
+     * store can be decommissioned after configuration update and restart.
+     */
+    COMPLETED,
+
+    /**
+     * Migration has failed. System has rolled-back to used the old metadata store.
+     */
+    FAILED,
+}