You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bookkeeper.apache.org by GitBox <gi...@apache.org> on 2022/09/20 02:07:11 UTC
[GitHub] [bookkeeper] zymap commented on a diff in pull request #3457: BP-56: Support data migration
zymap commented on code in PR #3457:
URL: https://github.com/apache/bookkeeper/pull/3457#discussion_r974062764
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookies/ReplicasMigrationCommand.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.tools.cli.commands.bookies;
+
+import com.beust.jcommander.Parameter;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
+import org.apache.bookkeeper.http.HttpRouter;
+import org.apache.bookkeeper.http.HttpServer;
+import org.apache.bookkeeper.meta.MetadataDrivers;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.tools.cli.helpers.BookieCommand;
+import org.apache.bookkeeper.tools.framework.CliFlags;
+import org.apache.bookkeeper.tools.framework.CliSpec;
+import org.apache.bookkeeper.util.HttpClientUtils;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Migrate the data of the specified replica to other bookie nodes.
+ * */
+public class ReplicasMigrationCommand extends BookieCommand<ReplicasMigrationCommand.ReplicasMigrationFlags> {
Review Comment:
Please add test
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookies/ReplicasMigrationCommand.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.tools.cli.commands.bookies;
+
+import com.beust.jcommander.Parameter;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
+import org.apache.bookkeeper.http.HttpRouter;
+import org.apache.bookkeeper.http.HttpServer;
+import org.apache.bookkeeper.meta.MetadataDrivers;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.tools.cli.helpers.BookieCommand;
+import org.apache.bookkeeper.tools.framework.CliFlags;
+import org.apache.bookkeeper.tools.framework.CliSpec;
+import org.apache.bookkeeper.util.HttpClientUtils;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Migrate the data of the specified replica to other bookie nodes.
+ * */
+public class ReplicasMigrationCommand extends BookieCommand<ReplicasMigrationCommand.ReplicasMigrationFlags> {
+ static final Logger LOG = LoggerFactory.getLogger(ReplicasMigrationCommand.class);
+ private static final String NAME = "replicasMigration";
+ private static final String DESC = "Migrate the data of the specified replica to other bookie nodes";
+ private static final String ALL = "ALL";
+ private final String seperator = ",";
+ public ReplicasMigrationCommand() {
+ this(new ReplicasMigrationFlags());
+ }
+ private ReplicasMigrationCommand(ReplicasMigrationFlags flags) {
+ super(CliSpec.<ReplicasMigrationFlags>newBuilder()
+ .withName(NAME)
+ .withDescription(DESC)
+ .withFlags(flags)
+ .build());
+ }
+
+ /**
+ * Flags for replicasMigration command.
+ */
+ @Accessors(fluent = true)
+ @Setter
+ public static class ReplicasMigrationFlags extends CliFlags {
+
+ @Parameter(names = { "-b", "--bookieIds" }, description = "Bookies corresponding to the migrated replica, "
+ + "eg: bookieId1,bookieId2,bookieId3")
+ private List<String> bookieIdsOfMigratedReplica;
+
+ @Parameter(names = { "-l", "--ledgerIds" }, description = "The ledgerIds corresponding to the migrated replica,"
+ + "eg: ledgerId1,ledgerId2,ledgerId3. The default is empty, which means that all replica data on these "
+ + "bookie nodes will be migrated to other bookie nodes")
+ private List<Long> ledgerIdsOfMigratedReplica = Collections.emptyList();
+
+ @Parameter(names = { "-r", "--readOnly" }, description = "Whether to set the bookie nodes of the migrated data"
+ + " to readOnly, the default is false")
+ private boolean readOnly = false;
+ }
+
+ @Override
+ public boolean apply(ServerConfiguration conf, ReplicasMigrationFlags cmdFlags) {
+ try (BookKeeperAdmin admin = new BookKeeperAdmin(new ClientConfiguration(conf))) {
+ if (cmdFlags.readOnly) {
+ if (!conf.isHttpServerEnabled()) {
Review Comment:
It is a server configuration, the client doesn't know whether the HTTP server is enabled or not.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieShell.java:
##########
@@ -1951,6 +1953,77 @@ public int runCmd(CommandLine cmdLine) throws Exception {
}
}
+ /**
+ * Migrate the specified ledger data on some bookie to other bookies.
+ * */
+ class ReplicasMigrationCmd extends MyCommand {
+
+ ReplicasMigrationCmd() {
+ super(CMD_REPLICASMIGRATION);
+ Option bookiesOpt = new Option("b", "bookieIds", true,
+ "Bookies corresponding to the migrated replica");
+ Option ledgersOpt = new Option("l", "ledgerIds", true,
Review Comment:
IIUC, in most cases, you want to migrate all the data from one bookie to another bookie, not part of the data. So why do we require the ledgerIds? Shouldn't it be optional?
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookies/ReplicasMigrationCommand.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.tools.cli.commands.bookies;
+
+import com.beust.jcommander.Parameter;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
+import org.apache.bookkeeper.http.HttpRouter;
+import org.apache.bookkeeper.http.HttpServer;
+import org.apache.bookkeeper.meta.MetadataDrivers;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.tools.cli.helpers.BookieCommand;
+import org.apache.bookkeeper.tools.framework.CliFlags;
+import org.apache.bookkeeper.tools.framework.CliSpec;
+import org.apache.bookkeeper.util.HttpClientUtils;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Migrate the data of the specified replica to other bookie nodes.
+ * */
+public class ReplicasMigrationCommand extends BookieCommand<ReplicasMigrationCommand.ReplicasMigrationFlags> {
+ static final Logger LOG = LoggerFactory.getLogger(ReplicasMigrationCommand.class);
+ private static final String NAME = "replicasMigration";
+ private static final String DESC = "Migrate the data of the specified replica to other bookie nodes";
+ private static final String ALL = "ALL";
+ private final String seperator = ",";
+ public ReplicasMigrationCommand() {
+ this(new ReplicasMigrationFlags());
+ }
+ private ReplicasMigrationCommand(ReplicasMigrationFlags flags) {
+ super(CliSpec.<ReplicasMigrationFlags>newBuilder()
+ .withName(NAME)
+ .withDescription(DESC)
+ .withFlags(flags)
+ .build());
+ }
+
+ /**
+ * Flags for replicasMigration command.
+ */
+ @Accessors(fluent = true)
+ @Setter
+ public static class ReplicasMigrationFlags extends CliFlags {
+
+ @Parameter(names = { "-b", "--bookieIds" }, description = "Bookies corresponding to the migrated replica, "
+ + "eg: bookieId1,bookieId2,bookieId3")
+ private List<String> bookieIdsOfMigratedReplica;
+
+ @Parameter(names = { "-l", "--ledgerIds" }, description = "The ledgerIds corresponding to the migrated replica,"
+ + "eg: ledgerId1,ledgerId2,ledgerId3. The default is empty, which means that all replica data on these "
+ + "bookie nodes will be migrated to other bookie nodes")
+ private List<Long> ledgerIdsOfMigratedReplica = Collections.emptyList();
+
+ @Parameter(names = { "-r", "--readOnly" }, description = "Whether to set the bookie nodes of the migrated data"
+ + " to readOnly, the default is false")
+ private boolean readOnly = false;
+ }
+
+ @Override
+ public boolean apply(ServerConfiguration conf, ReplicasMigrationFlags cmdFlags) {
+ try (BookKeeperAdmin admin = new BookKeeperAdmin(new ClientConfiguration(conf))) {
+ if (cmdFlags.readOnly) {
+ if (!conf.isHttpServerEnabled()) {
+ throw new RuntimeException("Please enable http service first, config httpServerEnabled is false!");
+ }
+ Set<BookieId> toSwitchBookieIdsSet = new HashSet<>();
+ for (String bookieIdStr : cmdFlags.bookieIdsOfMigratedReplica) {
+ toSwitchBookieIdsSet.add(BookieId.parse(bookieIdStr));
+ }
+
+ Collection<BookieId> readOnlyBookies = admin.getReadOnlyBookies();
+
+ Set<BookieId> switchedBookies = new HashSet<>();
+ if (!switchToReadonly(admin, toSwitchBookieIdsSet, switchedBookies, true)) {
Review Comment:
We can remove the read-only bookies from the switching bookie list first, then switch the remaining to the read-only.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicasMigrationWorker.java:
##########
@@ -0,0 +1,237 @@
+/**
+ * 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.replication;
+
+import static org.apache.bookkeeper.replication.ReplicationStats.NUM_ENTRIES_UNABLE_TO_READ_FOR_MIGRATION;
+
+import java.io.UnsupportedEncodingException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import org.apache.bookkeeper.bookie.BookieThread;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.client.LedgerChecker;
+import org.apache.bookkeeper.client.LedgerFragment;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.meta.MetadataClientDriver;
+import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.stats.Counter;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.stats.annotations.StatsDoc;
+import org.apache.bookkeeper.util.BookKeeperConstants;
+import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Migrate replica data to other bookie nodes.
+ */
+public class ReplicasMigrationWorker implements Runnable {
Review Comment:
MigrationReplicas? Please be consistent with all the places.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookies/ReplicasMigrationCommand.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.tools.cli.commands.bookies;
+
+import com.beust.jcommander.Parameter;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
+import org.apache.bookkeeper.http.HttpRouter;
+import org.apache.bookkeeper.http.HttpServer;
+import org.apache.bookkeeper.meta.MetadataDrivers;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.tools.cli.helpers.BookieCommand;
+import org.apache.bookkeeper.tools.framework.CliFlags;
+import org.apache.bookkeeper.tools.framework.CliSpec;
+import org.apache.bookkeeper.util.HttpClientUtils;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Migrate the data of the specified replica to other bookie nodes.
+ * */
+public class ReplicasMigrationCommand extends BookieCommand<ReplicasMigrationCommand.ReplicasMigrationFlags> {
+ static final Logger LOG = LoggerFactory.getLogger(ReplicasMigrationCommand.class);
+ private static final String NAME = "replicasMigration";
+ private static final String DESC = "Migrate the data of the specified replica to other bookie nodes";
+ private static final String ALL = "ALL";
+ private final String seperator = ",";
+ public ReplicasMigrationCommand() {
+ this(new ReplicasMigrationFlags());
+ }
+ private ReplicasMigrationCommand(ReplicasMigrationFlags flags) {
+ super(CliSpec.<ReplicasMigrationFlags>newBuilder()
+ .withName(NAME)
+ .withDescription(DESC)
+ .withFlags(flags)
+ .build());
+ }
+
+ /**
+ * Flags for replicasMigration command.
+ */
+ @Accessors(fluent = true)
+ @Setter
+ public static class ReplicasMigrationFlags extends CliFlags {
+
+ @Parameter(names = { "-b", "--bookieIds" }, description = "Bookies corresponding to the migrated replica, "
+ + "eg: bookieId1,bookieId2,bookieId3")
+ private List<String> bookieIdsOfMigratedReplica;
+
+ @Parameter(names = { "-l", "--ledgerIds" }, description = "The ledgerIds corresponding to the migrated replica,"
+ + "eg: ledgerId1,ledgerId2,ledgerId3. The default is empty, which means that all replica data on these "
+ + "bookie nodes will be migrated to other bookie nodes")
+ private List<Long> ledgerIdsOfMigratedReplica = Collections.emptyList();
+
+ @Parameter(names = { "-r", "--readOnly" }, description = "Whether to set the bookie nodes of the migrated data"
+ + " to readOnly, the default is false")
+ private boolean readOnly = false;
+ }
+
+ @Override
+ public boolean apply(ServerConfiguration conf, ReplicasMigrationFlags cmdFlags) {
+ try (BookKeeperAdmin admin = new BookKeeperAdmin(new ClientConfiguration(conf))) {
+ if (cmdFlags.readOnly) {
+ if (!conf.isHttpServerEnabled()) {
+ throw new RuntimeException("Please enable http service first, config httpServerEnabled is false!");
+ }
+ Set<BookieId> toSwitchBookieIdsSet = new HashSet<>();
+ for (String bookieIdStr : cmdFlags.bookieIdsOfMigratedReplica) {
+ toSwitchBookieIdsSet.add(BookieId.parse(bookieIdStr));
+ }
+
+ Collection<BookieId> readOnlyBookies = admin.getReadOnlyBookies();
+
+ Set<BookieId> switchedBookies = new HashSet<>();
+ if (!switchToReadonly(admin, toSwitchBookieIdsSet, switchedBookies, true)) {
+ LOG.warn("Some bookie nodes that fail to set readonly, "
+ + "and the successful bookies fall back to the previous state!");
+ Set<BookieId> fallbackSuccessfulBookies = new HashSet<>();
+ switchedBookies.removeAll(readOnlyBookies);
+ if (!switchToReadonly(admin, switchedBookies, fallbackSuccessfulBookies, false)) {
+ switchedBookies.removeAll(fallbackSuccessfulBookies);
+ LOG.error(String.format("Fallback failed! Failed nodes:%s", switchedBookies));
+ }
+ return false;
+ }
+ }
+
+ return replicasMigration(admin, conf, cmdFlags);
+ } catch (Exception e) {
+ throw new UncheckedExecutionException(e.getMessage(), e);
+ }
+ }
+
+ private boolean switchToReadonly(BookKeeperAdmin admin,
+ Set<BookieId> toSwitchBookieIdsSet,
+ Set<BookieId> switchedBookieIds,
+ boolean toReadOnly) throws BKException {
+ for (BookieId bookieId : toSwitchBookieIdsSet) {
+ BookieServiceInfo bookieServiceInfo = admin.getBookieServiceInfo(bookieId);
+ for (BookieServiceInfo.Endpoint endpoint : bookieServiceInfo.getEndpoints()) {
+ if (endpoint.getProtocol().equals("http")) {
+ String host = endpoint.getHost();
+ int port = endpoint.getPort();
+ String url = String.format("http://%s:%s/%s", host, port, HttpRouter.BOOKIE_STATE_READONLY);
+ String readOnly = toReadOnly ? "true" : "false";
+ CloseableHttpResponse response = HttpClientUtils.doPutRequest(url, readOnly, "UTF-8");
+ int responseCode = response.getStatusLine().getStatusCode();
+ if (responseCode != HttpServer.StatusCode.OK.getValue()) {
+ LOG.error(String.format("Set readonly to %s failed! bookieId:%s, responseCode:%s ",
+ readOnly, bookieId, responseCode));
+ return false;
+ }
+
+ switchedBookieIds.add(bookieId);
+ LOG.info(String.format("Set readonly to %s success! bookieId:%s, responseCode:%s",
+ readOnly, bookieId, responseCode));
+ }
+ }
+ }
+ return true;
+ }
+
+ private boolean replicasMigration(BookKeeperAdmin admin, ServerConfiguration conf, ReplicasMigrationFlags flags)
+ throws BKException, InterruptedException {
+ try {
+ // 1. Build bookieIdsSet adn ledgerIdsToMigrate
+ Set<BookieId> bookieIdsSet = new HashSet<>();
+ for (String bookieIdStr : flags.bookieIdsOfMigratedReplica) {
+ bookieIdsSet.add(BookieId.parse(bookieIdStr));
+ }
+ HashSet<Long> ledgerIdsToMigrate = new HashSet<>();
+ for (Long ledgerId : flags.ledgerIdsOfMigratedReplica) {
+ ledgerIdsToMigrate.add(ledgerId);
+ }
+
+ // 2. Build toMigratedLedgerAndBookieMap
+ Map<Long, Set<BookieId>> toMigratedLedgerAndBookieMap = new HashMap<>();
+ Map<BookieId, Set<Long>> bookieLedgerMaps = admin.listLedgers(bookieIdsSet);
+ for (Map.Entry<BookieId, Set<Long>> bookieIdSetEntry : bookieLedgerMaps.entrySet()) {
+ Set<Long> ledgersOnBookieId = bookieIdSetEntry.getValue();
+ if (!ledgerIdsToMigrate.isEmpty()) {
+ ledgersOnBookieId.retainAll(ledgerIdsToMigrate);
+ }
+ for (Long ledgerId : ledgersOnBookieId) {
+ Set<BookieId> bookieIdSet = toMigratedLedgerAndBookieMap.getOrDefault(ledgerId, new HashSet<>());
+ bookieIdSet.add(bookieIdSetEntry.getKey());
+ toMigratedLedgerAndBookieMap.put(ledgerId, bookieIdSet);
+ }
+ }
+
+ // 3. Submit ledger replicas to be migrated
+ MetadataDrivers.runFunctionWithMetadataBookieDriver(conf, driver -> {
+ try {
+ driver.submitToMigrateReplicas(toMigratedLedgerAndBookieMap);
+ } catch (Throwable e) {
+ LOG.warn(String.format("The migration task of Replicas=%s submit failed!",
+ toMigratedLedgerAndBookieMap), e);
+ }
+ return true;
Review Comment:
This method looks like don't need returned value?
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicasMigrationWorker.java:
##########
@@ -0,0 +1,237 @@
+/**
+ * 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.replication;
+
+import static org.apache.bookkeeper.replication.ReplicationStats.NUM_ENTRIES_UNABLE_TO_READ_FOR_MIGRATION;
+
+import java.io.UnsupportedEncodingException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import org.apache.bookkeeper.bookie.BookieThread;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.client.LedgerChecker;
+import org.apache.bookkeeper.client.LedgerFragment;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.meta.MetadataClientDriver;
+import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.stats.Counter;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.stats.annotations.StatsDoc;
+import org.apache.bookkeeper.util.BookKeeperConstants;
+import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Migrate replica data to other bookie nodes.
+ */
+public class ReplicasMigrationWorker implements Runnable {
+ private static final Logger LOG = LoggerFactory.getLogger(ReplicasMigrationWorker.class);
+ private final BookKeeperAdmin admin;
+ private final BiConsumer<Long, Long> onReadEntryFailureCallback;
+ private final MetadataClientDriver metadataClientDriver;
+ private final String rootPath;
+ private final String replicasMigrationPath;
+ private final String lock = "locked";
+ private final LedgerChecker ledgerChecker;
+ private final Thread workerThread;
+ private final long backoffMs = 100;
+ private volatile boolean workerRunning = false;
+ private final String seperator = ",";
+ private final String advertisedAddress;
+
+ @StatsDoc(
+ name = NUM_ENTRIES_UNABLE_TO_READ_FOR_MIGRATION,
+ help = "the number of entries ReplicasMigrationWorker unable to read"
+ )
+ private final Counter numEntriesUnableToReadForMigration;
+
+ public ReplicasMigrationWorker(final ServerConfiguration conf, BookKeeper bkc, StatsLogger statsLogger) {
+ this.advertisedAddress = conf.getAdvertisedAddress();
+ this.rootPath = ZKMetadataDriverBase.resolveZkLedgersRootPath(conf);
+ this.replicasMigrationPath = getReplicasMigrationPath(rootPath);
+ this.ledgerChecker = new LedgerChecker(bkc);
+ this.admin = new BookKeeperAdmin(bkc, statsLogger, new ClientConfiguration(conf));
+ this.metadataClientDriver = bkc.getMetadataClientDriver();
+ this.numEntriesUnableToReadForMigration = statsLogger
+ .getCounter(NUM_ENTRIES_UNABLE_TO_READ_FOR_MIGRATION);
+ this.onReadEntryFailureCallback = (ledgerid, entryid) -> {
+ numEntriesUnableToReadForMigration.inc();
+ };
+ this.workerThread = new BookieThread(this, "ReplicasMigrationWorker");
+ }
+
+ public static String getReplicasMigrationPath(String rootPath) {
+ return String.format("%s/%s", rootPath, BookKeeperConstants.MIGRATION_REPLICAS);
+ }
+
+ public static String getLockForMigrationReplicasPath(String replicasMigrationPath, long ledgerId, String lock) {
+ return String.format("%s/%s/%s", replicasMigrationPath, ledgerId, lock);
+ }
+
+ public static String getLedgerForMigrationReplicasPath(String replicasMigrationPath, long ledgerId) {
+ return String.format("%s/%s", replicasMigrationPath, ledgerId);
+ }
+
+ static class Replicas {
+ long ledgerId;
+ Set<BookieId> bookieIds;
+
+ public Replicas(long ledgerId, Set<BookieId> bookieIds) {
+ this.ledgerId = ledgerId;
+ this.bookieIds = bookieIds;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("<ledgerId=%s,bookies=<%s>", ledgerId, bookieIds);
+ }
+ }
+
+ private Replicas getReplicasToMigrate(List<String> ledgerIds) throws InterruptedException, KeeperException {
+ Iterator<String> iterator = ledgerIds.iterator();
+ while (iterator.hasNext()) {
+ try {
+ String ledgerId = iterator.next();
+ String ledgerForMigrationReplicasPath = getLedgerForMigrationReplicasPath(replicasMigrationPath,
+ Long.parseLong(ledgerId));
+ if (!metadataClientDriver.exists(ledgerForMigrationReplicasPath)) {
+ // The ledger migration is completed
+ iterator.remove();
+ continue;
+ }
+ String bookieIdStr = metadataClientDriver.
+ getOwnerBookiesMigrationReplicas(ledgerForMigrationReplicasPath);
+
+ String lockForMigrationReplicasPath = getLockForMigrationReplicasPath(replicasMigrationPath,
+ Long.parseLong(ledgerId), lock);
+ metadataClientDriver.lockMigrationReplicas(lockForMigrationReplicasPath, advertisedAddress);
+ String[] migrationBookieIds = bookieIdStr.split(seperator);
+ Set<BookieId> bookieIds = new HashSet<>();
+ for (int i = 0; i < migrationBookieIds.length; i++) {
+ bookieIds.add(BookieId.parse(migrationBookieIds[i]));
+ }
+ return new Replicas(Long.parseLong(ledgerId), bookieIds);
+ } catch (KeeperException.NodeExistsException nee) {
+ // do nothing, someone each could have locked the ledger
+ } catch (UnsupportedEncodingException e) {
+ LOG.error("The encoding is not supported!", e);
+ }
+ }
+ return null;
+ }
+
+ private Set<LedgerFragment> getFragmentsOnMigrationBookies(LedgerHandle lh, Set<BookieId> migrationBookieIds) {
+ return ledgerChecker.getFragmentsOnMigrationBookies(lh, migrationBookieIds);
+ }
+
+ private static void waitBackOffTime(long backoffMs) {
+ try {
+ Thread.sleep(backoffMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private void releaseLock(Replicas replicas) {
+ try {
+ metadataClientDriver.deleteZkPath(getLockForMigrationReplicasPath(replicasMigrationPath,
+ replicas.ledgerId, lock));
+ LOG.info(String.format("Release lock for ledgerId %s success!", replicas.ledgerId));
+ } catch (KeeperException.NoNodeException e) {
+ // do nothing,already release lock
+ } catch (Exception e) {
+ LOG.error(String.format("Release lock for ledgerId %s failed!", replicas.ledgerId));
+ }
+ }
+
+ /**
+ * Start the migration replicas worker.
+ */
+ public void start() {
+ this.workerThread.start();
+ }
+
+ @Override
+ public void run() {
+ workerRunning = true;
+ List<String> toMigrateLedgerIds = Collections.emptyList();
+ while (workerRunning) {
+ // 1. build migrating ledgers
+ try {
+ if (toMigrateLedgerIds.isEmpty()) {
+ toMigrateLedgerIds = metadataClientDriver.listLedgersOfMigrationReplicas(replicasMigrationPath);
+ }
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ } catch (Throwable e) {
+ LOG.error("list migrating ledgers failed!", e);
+ }
+
+ // 2. Get a replica ledger to replicate or wait back off
+ Replicas replicas = null;
+ try {
+ replicas = getReplicasToMigrate(toMigrateLedgerIds);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ } catch (Throwable e) {
+ LOG.error("Get migrating replicas failed!", e);
+ }
+ if (null == replicas) {
+ waitBackOffTime(backoffMs);
+ continue;
+ }
+
+ // 3. Get the fragment containing the migration bookie
+ LOG.info(String.format("Start migrate replicas(%s)!", replicas));
+ try (LedgerHandle lh = admin.openLedgerNoRecovery(replicas.ledgerId)) {
+ Set<LedgerFragment> fragments = getFragmentsOnMigrationBookies(lh, replicas.bookieIds);
+ if (fragments.size() < 1) {
+ //3.The replication has been completed, delete the ledgerId directory and release lock
+ releaseLock(replicas);
Review Comment:
It will release the lock twice.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/util/BookKeeperConstants.java:
##########
@@ -40,6 +40,7 @@ public class BookKeeperConstants {
public static final String AVAILABLE_NODE = "available";
public static final String COOKIE_NODE = "cookies";
public static final String UNDER_REPLICATION_NODE = "underreplication";
+ public static final String MIGRATION_REPLICAS = "migrationReplicas";
Review Comment:
```suggestion
public static final String MIGRATION_REPLICAS = "migratedReplicas";
```
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverBase.java:
##########
@@ -239,6 +251,7 @@ protected void initialize(AbstractConfiguration<?> conf,
}
disableHealthCheckPath = ledgersRootPath + "/" + DISABLE_HEALTH_CHECK;
+ replicasMigrationPath = ledgersRootPath + "/" + MIGRATION_REPLICAS;
Review Comment:
Can we make the replicas migration as a single class like how underreplication does? We can create a MigrationManager to handle that.
https://github.com/apache/bookkeeper/blob/master/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java#L88
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverBase.java:
##########
@@ -289,6 +302,52 @@ public void processResult(int rc, String path, Object ctx, String name) {
return createResult;
}
+ private String getLedgerForReplicasMigrationPath(long ledgerId) {
+ return String.format("%s/%s", replicasMigrationPath, ledgerId);
Review Comment:
Because it is similar to replication, we can refer to how replication located the ledger in the zookeeper. And we can consistent the ledgerId format with that.
https://github.com/apache/bookkeeper/blob/master/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java#L252
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MetadataClientDriver.java:
##########
@@ -106,6 +110,50 @@ interface SessionStateListener {
*/
void setSessionStateListener(SessionStateListener sessionStateListener);
+ /**
+ * Get the list of ledgers corresponding to the replica to be migrated.
+ * */
+ default List<String> listLedgersOfMigrationReplicas(String migrationReplicasPath)
+ throws InterruptedException, KeeperException {
+ return new ArrayList<>();
Review Comment:
I saw other methods throw an unsupported exception, we should do the same thing on this.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookies/ReplicasMigrationCommand.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.tools.cli.commands.bookies;
+
+import com.beust.jcommander.Parameter;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.BookieServiceInfo;
+import org.apache.bookkeeper.http.HttpRouter;
+import org.apache.bookkeeper.http.HttpServer;
+import org.apache.bookkeeper.meta.MetadataDrivers;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.tools.cli.helpers.BookieCommand;
+import org.apache.bookkeeper.tools.framework.CliFlags;
+import org.apache.bookkeeper.tools.framework.CliSpec;
+import org.apache.bookkeeper.util.HttpClientUtils;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Migrate the data of the specified replica to other bookie nodes.
+ * */
+public class ReplicasMigrationCommand extends BookieCommand<ReplicasMigrationCommand.ReplicasMigrationFlags> {
+ static final Logger LOG = LoggerFactory.getLogger(ReplicasMigrationCommand.class);
+ private static final String NAME = "replicasMigration";
+ private static final String DESC = "Migrate the data of the specified replica to other bookie nodes";
+ private static final String ALL = "ALL";
+ private final String seperator = ",";
+ public ReplicasMigrationCommand() {
+ this(new ReplicasMigrationFlags());
+ }
+ private ReplicasMigrationCommand(ReplicasMigrationFlags flags) {
+ super(CliSpec.<ReplicasMigrationFlags>newBuilder()
+ .withName(NAME)
+ .withDescription(DESC)
+ .withFlags(flags)
+ .build());
+ }
+
+ /**
+ * Flags for replicasMigration command.
+ */
+ @Accessors(fluent = true)
+ @Setter
+ public static class ReplicasMigrationFlags extends CliFlags {
+
+ @Parameter(names = { "-b", "--bookieIds" }, description = "Bookies corresponding to the migrated replica, "
+ + "eg: bookieId1,bookieId2,bookieId3")
+ private List<String> bookieIdsOfMigratedReplica;
+
+ @Parameter(names = { "-l", "--ledgerIds" }, description = "The ledgerIds corresponding to the migrated replica,"
+ + "eg: ledgerId1,ledgerId2,ledgerId3. The default is empty, which means that all replica data on these "
+ + "bookie nodes will be migrated to other bookie nodes")
+ private List<Long> ledgerIdsOfMigratedReplica = Collections.emptyList();
+
+ @Parameter(names = { "-r", "--readOnly" }, description = "Whether to set the bookie nodes of the migrated data"
+ + " to readOnly, the default is false")
+ private boolean readOnly = false;
+ }
+
+ @Override
+ public boolean apply(ServerConfiguration conf, ReplicasMigrationFlags cmdFlags) {
+ try (BookKeeperAdmin admin = new BookKeeperAdmin(new ClientConfiguration(conf))) {
+ if (cmdFlags.readOnly) {
+ if (!conf.isHttpServerEnabled()) {
+ throw new RuntimeException("Please enable http service first, config httpServerEnabled is false!");
+ }
+ Set<BookieId> toSwitchBookieIdsSet = new HashSet<>();
+ for (String bookieIdStr : cmdFlags.bookieIdsOfMigratedReplica) {
+ toSwitchBookieIdsSet.add(BookieId.parse(bookieIdStr));
+ }
+
+ Collection<BookieId> readOnlyBookies = admin.getReadOnlyBookies();
+
+ Set<BookieId> switchedBookies = new HashSet<>();
+ if (!switchToReadonly(admin, toSwitchBookieIdsSet, switchedBookies, true)) {
+ LOG.warn("Some bookie nodes that fail to set readonly, "
+ + "and the successful bookies fall back to the previous state!");
+ Set<BookieId> fallbackSuccessfulBookies = new HashSet<>();
+ switchedBookies.removeAll(readOnlyBookies);
+ if (!switchToReadonly(admin, switchedBookies, fallbackSuccessfulBookies, false)) {
Review Comment:
Is it necessary to retry it? Because it is a command line tool, rerunning the command should be ok?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@bookkeeper.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org