You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by av...@apache.org on 2020/10/06 20:24:30 UTC
[hadoop-ozone] 03/08: HDDS-4141. Implement Finalize command in
Ozone Manager client. (#1400)
This is an automated email from the ASF dual-hosted git repository.
avijayan pushed a commit to branch HDDS-3698-upgrade
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit 9b55f696625c8bf2e0c667d9cc78dfa5dbc391cf
Author: Istvan Fajth <pi...@cloudera.com>
AuthorDate: Wed Sep 9 21:02:24 2020 +0200
HDDS-4141. Implement Finalize command in Ozone Manager client. (#1400)
---
.../main/java/org/apache/hadoop/ozone/OmUtils.java | 2 +
.../ozone/om/protocol/OzoneManagerProtocol.java | 64 ++++++
...OzoneManagerProtocolClientSideTranslatorPB.java | 44 ++++
.../src/main/proto/OmClientProtocol.proto | 36 ++++
hadoop-ozone/ozone-manager/pom.xml | 5 +
.../hadoop/ozone/om/OmMetadataManagerImpl.java | 14 ++
.../org/apache/hadoop/ozone/om/OzoneManager.java | 69 +++++++
.../ozone/om/ratis/OzoneManagerDoubleBuffer.java | 13 +-
.../om/ratis/utils/OzoneManagerRatisUtils.java | 6 +
.../upgrade/OMFinalizeUpgradeProgressRequest.java | 87 ++++++++
.../request/upgrade/OMFinalizeUpgradeRequest.java | 80 ++++++++
.../ozone/om/request/upgrade/package-info.java | 23 +++
.../hadoop/ozone/om/response/CleanupTableInfo.java | 10 +-
.../upgrade/OMFinalizeUpgradeProgressResponse.java | 45 +++++
.../upgrade/OMFinalizeUpgradeResponse.java | 43 ++++
.../ozone/om/response/upgrade/package-info.java | 23 +++
.../hadoop/ozone/om/TestOmMetadataManager.java | 10 +
.../ozone/om/response/TestCleanupTableInfo.java | 60 ++++--
.../ozone/admin/om/FinalizeUpgradeSubCommand.java | 221 +++++++++++++++++++++
.../org/apache/hadoop/ozone/admin/om/OMAdmin.java | 39 +++-
20 files changed, 873 insertions(+), 21 deletions(-)
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java
index 2a34580..67bd2a0 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java
@@ -271,6 +271,8 @@ public final class OmUtils {
case AddAcl:
case PurgeKeys:
case RecoverTrash:
+ case FinalizeUpgradeProgress:
+ case FinalizeUpgrade:
return false;
default:
LOG.error("CmdType {} is not categorized as readOnly or not.", cmdType);
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java
index 79cc926..8c0d686 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java
@@ -49,6 +49,7 @@ import org.apache.hadoop.ozone.om.helpers.ServiceInfo;
import org.apache.hadoop.ozone.om.helpers.ServiceInfoEx;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneAclInfo;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UpgradeFinalizationStatus;
import org.apache.hadoop.ozone.security.OzoneDelegationTokenSelector;
import org.apache.hadoop.ozone.security.acl.OzoneObj;
import org.apache.hadoop.security.KerberosInfo;
@@ -312,6 +313,69 @@ public interface OzoneManagerProtocol
ServiceInfoEx getServiceInfo() throws IOException;
+ /**
+ * Initiate metadata upgrade finalization.
+ * This method when called, initiates finalization of Ozone Manager metadata
+ * during an upgrade. The status returned contains the status
+ * - ALREADY_FINALIZED with empty message list when the software layout
+ * version and the metadata layout version are equal
+ * - STARTING_FINALIZATION with empty message list when the finalization
+ * has been started successfully
+ * - If a finalization is already in progress, then the method throws an
+ * {@link OMException} with a result code INVALID_REQUEST
+ *
+ *
+ * The leader Ozone Manager initiates finalization of the followers via
+ * the Raft protocol in other Ozone Managers, and reports progress to the
+ * client via the {@link #queryUpgradeFinalizationProgress(String, boolean)}
+ * call.
+ *
+ * The follower Ozone Managers reject this request and directs the client to
+ * the leader.
+ *
+ * @param upgradeClientID String identifier of the upgrade finalizer client
+ * @return the finalization status.
+ * @throws IOException
+ * when finalization is failed, or this Ozone Manager is not the
+ * leader.
+ * @throws OMException
+ * when finalization is already in progress.
+ */
+ UpgradeFinalizationStatus finalizeUpgrade(
+ String upgradeClientID
+ ) throws IOException;
+
+ /**
+ * Queries the current status of finalization.
+ * This method when called, returns the status messages from the finalization
+ * progress, if any. The status returned is
+ * - FINALIZATION_IN_PROGRESS, and the messages since the last query if the
+ * finalization is still running
+ * - FINALIZATION_DONE with a message list containing the messages since
+ * the last query, if the finalization ended but the messages were not
+ * yet emitted to the client.
+ * - ALREADY_FINALIZED with an empty message list otherwise
+ * - If finalization is not in progress, but software layout version and
+ * metadata layout version are different, the method will throw an
+ * {@link OMException} with a result code INVALID_REQUEST
+ * - If during finalization an other client with different ID than the one
+ * initiated finalization is calling the method, then an
+ * {@link OMException} with a result code INVALID_REQUEST is thrown,
+ * unless the request is forced by a new client, in which case the new
+ * client takes over the old client and the old client should exit.
+ *
+ * @param upgradeClientID String identifier of the upgrade finalizer client
+ * @param force set force takeover of output monitoring
+ * @return the finalization status and status messages.
+ * @throws IOException
+ * if there was a problem during the query
+ * @throws OMException
+ * if finalization is needed but not yet started
+ */
+ UpgradeFinalizationStatus queryUpgradeFinalizationProgress(
+ String upgradeClientID, boolean takeover
+ ) throws IOException;
+
/*
* S3 Specific functionality that is supported by Ozone Manager.
*/
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
index 6fea681..919c622 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java
@@ -78,6 +78,10 @@ import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DeleteK
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DeleteKeyRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DeleteKeysRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DeleteVolumeRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.FinalizeUpgradeProgressRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.FinalizeUpgradeProgressResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.FinalizeUpgradeRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.FinalizeUpgradeResponse;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetAclRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetAclResponse;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetDelegationTokenResponseProto;
@@ -135,6 +139,7 @@ import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetAclR
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetBucketPropertyRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetVolumePropertyRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UpgradeFinalizationStatus;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.VolumeInfo;
import org.apache.hadoop.ozone.protocolPB.OMPBHelper;
import org.apache.hadoop.ozone.security.OzoneTokenIdentifier;
@@ -1073,6 +1078,45 @@ public final class OzoneManagerProtocolClientSideTranslatorPB
resp.getCaCertificate());
}
+ @Override
+ public UpgradeFinalizationStatus finalizeUpgrade(
+ String upgradeClientID
+ ) throws IOException {
+ FinalizeUpgradeRequest req = FinalizeUpgradeRequest.newBuilder()
+ .setUpgradeClientId(upgradeClientID)
+ .build();
+
+ OMRequest omRequest = createOMRequest(Type.FinalizeUpgrade)
+ .setFinalizeUpgradeRequest(req)
+ .build();
+
+ FinalizeUpgradeResponse response =
+ handleError(submitRequest(omRequest)).getFinalizeUpgradeResponse();
+
+ return response.getStatus();
+ }
+
+ @Override
+ public UpgradeFinalizationStatus queryUpgradeFinalizationProgress(
+ String upgradeClientID, boolean takeover
+ ) throws IOException {
+ FinalizeUpgradeProgressRequest req = FinalizeUpgradeProgressRequest
+ .newBuilder()
+ .setUpgradeClientId(upgradeClientID)
+ .setTakeover(takeover)
+ .build();
+
+ OMRequest omRequest = createOMRequest(Type.FinalizeUpgradeProgress)
+ .setFinalizeUpgradeProgressRequest(req)
+ .build();
+
+ FinalizeUpgradeProgressResponse response =
+ handleError(submitRequest(omRequest))
+ .getFinalizeUpgradeProgressResponse();
+
+ return response.getStatus();
+ }
+
/**
* Get a valid Delegation Token.
*
diff --git a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
index e30f775..d1e2971 100644
--- a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
+++ b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
@@ -71,6 +71,8 @@ enum Type {
ServiceList = 51;
DBUpdates = 53;
+ FinalizeUpgrade = 54;
+ FinalizeUpgradeProgress = 55;
GetDelegationToken = 61;
RenewDelegationToken = 62;
@@ -138,6 +140,8 @@ message OMRequest {
optional ServiceListRequest serviceListRequest = 51;
optional DBUpdatesRequest dbUpdatesRequest = 53;
+ optional FinalizeUpgradeRequest finalizeUpgradeRequest = 54;
+ optional FinalizeUpgradeProgressRequest finalizeUpgradeProgressRequest = 55;
optional hadoop.common.GetDelegationTokenRequestProto getDelegationTokenRequest = 61;
optional hadoop.common.RenewDelegationTokenRequestProto renewDelegationTokenRequest= 62;
@@ -211,6 +215,8 @@ message OMResponse {
optional ServiceListResponse ServiceListResponse = 51;
optional DBUpdatesResponse dbUpdatesResponse = 52;
+ optional FinalizeUpgradeResponse finalizeUpgradeResponse = 54;
+ optional FinalizeUpgradeProgressResponse finalizeUpgradeProgressResponse = 55;
optional GetDelegationTokenResponseProto getDelegationTokenResponse = 61;
optional RenewDelegationTokenResponseProto renewDelegationTokenResponse = 62;
@@ -1014,6 +1020,36 @@ message DBUpdatesResponse {
repeated bytes data = 2;
}
+
+message UpgradeFinalizationStatus {
+ enum Status {
+ ALREADY_FINALIZED = 1;
+ STARTING_FINALIZATION = 2;
+ FINALIZATION_IN_PROGRESS = 3;
+ FINALIZATION_DONE = 4;
+ FINALIZATION_REQUIRED = 5;
+ }
+ required Status status = 1;
+ repeated string messages = 2;
+}
+
+message FinalizeUpgradeRequest {
+ required string upgradeClientId = 1;
+}
+
+message FinalizeUpgradeResponse {
+ required UpgradeFinalizationStatus status = 1;
+}
+
+message FinalizeUpgradeProgressRequest {
+ required string upgradeClientId = 1;
+ optional bool takeover = 2;
+}
+
+message FinalizeUpgradeProgressResponse {
+ required UpgradeFinalizationStatus status = 1;
+}
+
message ServicePort {
enum Type {
RPC = 1;
diff --git a/hadoop-ozone/ozone-manager/pom.xml b/hadoop-ozone/ozone-manager/pom.xml
index 7891666..4c9a901 100644
--- a/hadoop-ozone/ozone-manager/pom.xml
+++ b/hadoop-ozone/ozone-manager/pom.xml
@@ -121,6 +121,11 @@ https://maven.apache.org/xsd/maven-4.0.0.xsd">
<artifactId>jmockit</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>hamcrest-all</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.reflections</groupId>
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
index da7e985..56a9629 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
@@ -143,6 +143,20 @@ public class OmMetadataManagerImpl implements OMMetadataManager {
public static final String TRANSACTION_INFO_TABLE =
"transactionInfoTable";
+ public static final String[] ALL_TABLES = new String[] {
+ USER_TABLE,
+ VOLUME_TABLE,
+ BUCKET_TABLE,
+ KEY_TABLE,
+ DELETED_TABLE,
+ OPEN_KEY_TABLE,
+ MULTIPARTINFO_TABLE,
+ S3_SECRET_TABLE,
+ DELEGATION_TOKEN_TABLE,
+ PREFIX_TABLE,
+ TRANSACTION_INFO_TABLE
+ };
+
private DBStore store;
private final OzoneManagerLock lock;
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
index 5af09ee..1809827 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java
@@ -41,6 +41,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Random;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
@@ -149,6 +150,7 @@ import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRoleI
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneAclInfo;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ServicePort;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UserVolumeInfo;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UpgradeFinalizationStatus;
import org.apache.hadoop.ozone.protocolPB.OzoneManagerProtocolServerSideTranslatorPB;
import org.apache.hadoop.ozone.security.OzoneBlockTokenSecretManager;
import org.apache.hadoop.ozone.security.OzoneDelegationTokenSecretManager;
@@ -2600,6 +2602,73 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
return new ServiceInfoEx(getServiceList(), caCertPem);
}
+ private final List<String> finalizationMsgs = new ArrayList<>();
+ private UpgradeFinalizationStatus.Status finalizationStatus =
+ UpgradeFinalizationStatus.Status.FINALIZATION_REQUIRED;
+
+ @Override
+ public UpgradeFinalizationStatus finalizeUpgrade(String upgradeClientID)
+ throws IOException {
+ if (!finalizationStatus
+ .equals(UpgradeFinalizationStatus.Status.FINALIZATION_REQUIRED)){
+ throw new OMException("Finalization is not needed.", INVALID_REQUEST);
+ }
+ finalizationStatus = UpgradeFinalizationStatus.Status.STARTING_FINALIZATION;
+ UpgradeFinalizationStatus status = UpgradeFinalizationStatus.newBuilder()
+ .setStatus(finalizationStatus)
+ .build();
+ LOG.info("FinalizeUpgrade initiated by client: {}.", upgradeClientID);
+ if (isLeader()) {
+ finalizationMsgs.add("Finalization started.");
+ finalizationStatus =
+ UpgradeFinalizationStatus.Status.FINALIZATION_IN_PROGRESS;
+
+ new Thread(() -> {
+ LOG.info("Finalization thread started.");
+ int i = 0;
+ Random random = new Random(0xafaf);
+ while (i < 50) {
+ int rand = random.nextInt(Math.min(10, 50 - i)) + 1;
+ synchronized (finalizationMsgs) {
+ LOG.info("Emitting {} messages", rand);
+ for (int j = 0; j < rand; j++) {
+ LOG.info("Upgrade MSG: {} - added.", "Message " + i + ".");
+ finalizationMsgs.add("Message " + i + ".");
+ i++;
+ }
+ }
+ try {
+ int sleep = random.nextInt(1200);
+ LOG.info("Sleeping {}ms before emit messages again.", sleep);
+ Thread.sleep(sleep);
+ } catch (InterruptedException e) {
+ LOG.info("Finalization thread interrupted.", e);
+ return;
+ }
+ }
+ LOG.info("Finalization done.");
+ finalizationStatus = UpgradeFinalizationStatus.Status.FINALIZATION_DONE;
+ }, "Finalization-Thread").start();
+ }
+ return status;
+ }
+
+ @Override
+ public UpgradeFinalizationStatus queryUpgradeFinalizationProgress(
+ String upgradeClientID, boolean takeover
+ ) throws IOException {
+ UpgradeFinalizationStatus.Builder builder =
+ UpgradeFinalizationStatus.newBuilder();
+ builder.setStatus(finalizationStatus);
+ List<String> msgs = new ArrayList<>();
+ synchronized (finalizationMsgs) {
+ msgs.addAll(finalizationMsgs);
+ finalizationMsgs.clear();
+ }
+ builder.addAllMessages(msgs);
+ return builder.build();
+ }
+
@Override
/**
* {@inheritDoc}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerDoubleBuffer.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerDoubleBuffer.java
index f1c144e..68d359e 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerDoubleBuffer.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerDoubleBuffer.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.ozone.om.ratis;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -37,6 +38,8 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.function.SupplierWithIOException;
import org.apache.hadoop.hdds.tracing.TracingUtil;
+import org.apache.hadoop.hdds.utils.db.DBColumnFamilyDefinition;
+import org.apache.hadoop.ozone.om.codec.OMDBDefinition;
import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
import org.apache.hadoop.util.Time;
@@ -404,7 +407,15 @@ public final class OzoneManagerDoubleBuffer {
CleanupTableInfo cleanupTableInfo =
responseClass.getAnnotation(CleanupTableInfo.class);
if (cleanupTableInfo != null) {
- String[] cleanupTables = cleanupTableInfo.cleanupTables();
+ String[] cleanupTables;
+ if (cleanupTableInfo.cleanupAll()){
+ cleanupTables = Arrays
+ .stream(new OMDBDefinition().getColumnFamilies())
+ .map(DBColumnFamilyDefinition::getTableName)
+ .toArray(String[]::new);
+ } else {
+ cleanupTables = cleanupTableInfo.cleanupTables();
+ }
for (String table : cleanupTables) {
cleanupEpochs.computeIfAbsent(table, list -> new ArrayList<>())
.add(entry.getTrxLogIndex());
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java
index 681c0da..f43dfba 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java
@@ -57,6 +57,8 @@ import org.apache.hadoop.ozone.om.request.s3.security.S3GetSecretRequest;
import org.apache.hadoop.ozone.om.request.security.OMCancelDelegationTokenRequest;
import org.apache.hadoop.ozone.om.request.security.OMGetDelegationTokenRequest;
import org.apache.hadoop.ozone.om.request.security.OMRenewDelegationTokenRequest;
+import org.apache.hadoop.ozone.om.request.upgrade.OMFinalizeUpgradeProgressRequest;
+import org.apache.hadoop.ozone.om.request.upgrade.OMFinalizeUpgradeRequest;
import org.apache.hadoop.ozone.om.request.volume.OMVolumeCreateRequest;
import org.apache.hadoop.ozone.om.request.volume.OMVolumeDeleteRequest;
import org.apache.hadoop.ozone.om.request.volume.OMVolumeSetOwnerRequest;
@@ -160,6 +162,10 @@ public final class OzoneManagerRatisUtils {
return new S3GetSecretRequest(omRequest);
case RecoverTrash:
return new OMTrashRecoverRequest(omRequest);
+ case FinalizeUpgrade:
+ return new OMFinalizeUpgradeRequest(omRequest);
+ case FinalizeUpgradeProgress:
+ return new OMFinalizeUpgradeProgressRequest(omRequest);
default:
throw new IllegalStateException("Unrecognized write command " +
"type request" + cmdType);
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMFinalizeUpgradeProgressRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMFinalizeUpgradeProgressRequest.java
new file mode 100644
index 0000000..3cb9210
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMFinalizeUpgradeProgressRequest.java
@@ -0,0 +1,87 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hadoop.ozone.om.request.upgrade;
+
+import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
+import org.apache.hadoop.ozone.om.request.OMClientRequest;
+import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.om.response.upgrade.OMFinalizeUpgradeProgressResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.FinalizeUpgradeProgressRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.FinalizeUpgradeProgressResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UpgradeFinalizationStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * Handles finalizeUpgradeProgress request that serves to query the status
+ * of the async finalization progress.
+ */
+public class OMFinalizeUpgradeProgressRequest extends OMClientRequest {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(OMFinalizeUpgradeProgressRequest.class);
+
+ public OMFinalizeUpgradeProgressRequest(OMRequest omRequest) {
+ super(omRequest);
+ }
+
+ @Override public OMClientResponse validateAndUpdateCache(
+ OzoneManager ozoneManager, long transactionLogIndex,
+ OzoneManagerDoubleBufferHelper ozoneManagerDoubleBufferHelper) {
+
+ LOG.info("Finalization progress check's validateAndUpdateCache"
+ + "called and started.");
+ LOG.trace("Request: {}", getOmRequest());
+ OzoneManagerProtocolProtos.OMResponse.Builder responseBuilder =
+ OmResponseUtil.getOMResponseBuilder(getOmRequest());
+ responseBuilder
+ .setCmdType(OzoneManagerProtocolProtos.Type.FinalizeUpgradeProgress);
+ OMClientResponse response = null;
+
+ try {
+ FinalizeUpgradeProgressRequest finalizeUpgradeProgressRequest =
+ getOmRequest().getFinalizeUpgradeProgressRequest();
+ String upgradeClientID =
+ finalizeUpgradeProgressRequest.getUpgradeClientId();
+ boolean takeover = finalizeUpgradeProgressRequest.getTakeover();
+
+ UpgradeFinalizationStatus status =
+ ozoneManager
+ .queryUpgradeFinalizationProgress(upgradeClientID, takeover);
+
+ FinalizeUpgradeProgressResponse omResponse =
+ FinalizeUpgradeProgressResponse.newBuilder()
+ .setStatus(status)
+ .build();
+
+ responseBuilder.setFinalizeUpgradeProgressResponse(omResponse);
+ response = new OMFinalizeUpgradeProgressResponse(responseBuilder.build());
+ LOG.trace("Returning response: {}", response);
+ } catch (IOException e) {
+ response = new OMFinalizeUpgradeProgressResponse(
+ createErrorOMResponse(responseBuilder, e));
+ }
+
+ return response;
+ }
+}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMFinalizeUpgradeRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMFinalizeUpgradeRequest.java
new file mode 100644
index 0000000..772eae7
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMFinalizeUpgradeRequest.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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hadoop.ozone.om.request.upgrade;
+
+import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
+import org.apache.hadoop.ozone.om.request.OMClientRequest;
+import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.om.response.upgrade.OMFinalizeUpgradeResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.FinalizeUpgradeRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.FinalizeUpgradeResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UpgradeFinalizationStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * Handles finalizeUpgrade request.
+ */
+public class OMFinalizeUpgradeRequest extends OMClientRequest {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(OMFinalizeUpgradeRequest.class);
+
+ public OMFinalizeUpgradeRequest(OMRequest omRequest) {
+ super(omRequest);
+ }
+
+ @Override
+ public OMClientResponse validateAndUpdateCache(
+ OzoneManager ozoneManager, long transactionLogIndex,
+ OzoneManagerDoubleBufferHelper ozoneManagerDoubleBufferHelper) {
+ LOG.info("Finalization's validateAndUpdateCache called and started.");
+ LOG.trace("Request: {}", getOmRequest());
+ OMResponse.Builder responseBuilder =
+ OmResponseUtil.getOMResponseBuilder(getOmRequest());
+ responseBuilder.setCmdType(OzoneManagerProtocolProtos.Type.FinalizeUpgrade);
+ OMClientResponse response = null;
+
+ try {
+ FinalizeUpgradeRequest request =
+ getOmRequest().getFinalizeUpgradeRequest();
+
+ String upgradeClientID = request.getUpgradeClientId();
+
+ UpgradeFinalizationStatus status =
+ ozoneManager.finalizeUpgrade(upgradeClientID);
+
+ FinalizeUpgradeResponse omResponse =
+ FinalizeUpgradeResponse.newBuilder().setStatus(status).build();
+ responseBuilder.setFinalizeUpgradeResponse(omResponse);
+ response = new OMFinalizeUpgradeResponse(responseBuilder.build());
+ LOG.trace("Returning response: {}", response);
+ } catch (IOException e) {
+ response = new OMFinalizeUpgradeResponse(
+ createErrorOMResponse(responseBuilder, e));
+ }
+
+ return response;
+ }
+}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/package-info.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/package-info.java
new file mode 100644
index 0000000..d785d90
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/package-info.java
@@ -0,0 +1,23 @@
+/*
+ * 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 contains classes related to upgrade finalization requests.
+ */
+package org.apache.hadoop.ozone.om.request.upgrade;
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/CleanupTableInfo.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/CleanupTableInfo.java
index e456423..39416ac 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/CleanupTableInfo.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/CleanupTableInfo.java
@@ -40,5 +40,13 @@ public @interface CleanupTableInfo {
* during cleanup table cache.
* @return list of table names.
*/
- String[] cleanupTables();
+ String[] cleanupTables() default {};
+
+ /**
+ * If all tables are affected, like at update finalization, one can specify
+ * cleanupAll=true, instead of the list of all tables. In this case the
+ * cleanupTable property has to be defined as an empty array (the default).
+ * @return whether to cleanup all tables.
+ */
+ boolean cleanupAll() default false;
}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/upgrade/OMFinalizeUpgradeProgressResponse.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/upgrade/OMFinalizeUpgradeProgressResponse.java
new file mode 100644
index 0000000..f07e275
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/upgrade/OMFinalizeUpgradeProgressResponse.java
@@ -0,0 +1,45 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hadoop.ozone.om.response.upgrade;
+
+import org.apache.hadoop.hdds.utils.db.BatchOperation;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
+import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+
+import java.io.IOException;
+
+/**
+ * Response for finalizeUpgradeProgress request.
+ */
+// yepp this will not be a write request, adding a table here to the annotation
+// just to pass tests related to this annotation.
+@CleanupTableInfo(cleanupTables = { OmMetadataManagerImpl.USER_TABLE })
+public class OMFinalizeUpgradeProgressResponse extends OMClientResponse {
+ public OMFinalizeUpgradeProgressResponse(
+ OzoneManagerProtocolProtos.OMResponse omResponse) {
+ super(omResponse);
+ }
+
+ @Override protected void addToDBBatch(OMMetadataManager omMetadataManager,
+ BatchOperation batchOperation) throws IOException {
+
+ }
+}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/upgrade/OMFinalizeUpgradeResponse.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/upgrade/OMFinalizeUpgradeResponse.java
new file mode 100644
index 0000000..04e6a24
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/upgrade/OMFinalizeUpgradeResponse.java
@@ -0,0 +1,43 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hadoop.ozone.om.response.upgrade;
+
+import org.apache.hadoop.hdds.utils.db.BatchOperation;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+
+import java.io.IOException;
+
+/**
+ * Response for finalizeUpgrade request.
+ */
+@CleanupTableInfo(cleanupAll = true)
+public class OMFinalizeUpgradeResponse extends OMClientResponse {
+ public OMFinalizeUpgradeResponse(
+ OzoneManagerProtocolProtos.OMResponse omResponse) {
+ super(omResponse);
+ }
+
+ @Override
+ protected void addToDBBatch(OMMetadataManager omMetadataManager,
+ BatchOperation batchOperation) throws IOException {
+
+ }
+}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/upgrade/package-info.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/upgrade/package-info.java
new file mode 100644
index 0000000..a2e7415
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/upgrade/package-info.java
@@ -0,0 +1,23 @@
+/*
+ * 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 contains classes related to upgrade finalization responses.
+ */
+package org.apache.hadoop.ozone.om.response.upgrade;
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java
index 7c2d258..71193c9 100644
--- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmMetadataManager.java
@@ -44,6 +44,8 @@ import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_OPEN_KEY_EXPIRE_THRE
import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_OPEN_KEY_EXPIRE_THRESHOLD_SECONDS_DEFAULT;
import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_DB_DIRS;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.junit.Assert.assertThat;
/**
* Tests OzoneManager MetadataManager.
@@ -612,4 +614,12 @@ public class TestOmMetadataManager {
}
}
+ @Test
+ public void testAllTablesAreProperInOMMetadataManagerImpl() {
+ String[] tablesByDefinition = OmMetadataManagerImpl.ALL_TABLES;
+
+ Set<String> tablesInManager = omMetadataManager.listTableNames();
+
+ assertThat(tablesInManager, containsInAnyOrder(tablesByDefinition));
+ }
}
\ No newline at end of file
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestCleanupTableInfo.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestCleanupTableInfo.java
index f66e3a3..f813000 100644
--- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestCleanupTableInfo.java
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestCleanupTableInfo.java
@@ -22,14 +22,18 @@ import org.apache.hadoop.hdds.server.ServerUtils;
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.reflections.Reflections;
import java.io.File;
+import java.util.Arrays;
import java.util.Set;
+import static org.junit.Assert.assertTrue;
+
/**
* This tests check whether {@link OMClientResponse} have defined
* {@link CleanupTableInfo} annotation.
@@ -39,31 +43,53 @@ public class TestCleanupTableInfo {
@Rule
public TemporaryFolder folder = new TemporaryFolder();
- @Test
- public void checkAnnotationAndTableName() throws Exception {
- OzoneConfiguration conf = new OzoneConfiguration();
+ private OzoneConfiguration conf = new OzoneConfiguration();
+
+ @Before
+ public void setupMetaManager() throws Exception {
File newFolder = folder.newFolder();
if (!newFolder.exists()) {
Assert.assertTrue(newFolder.mkdirs());
}
ServerUtils.setOzoneMetaDirPath(conf, newFolder.toString());
- OMMetadataManager omMetadataManager = new OmMetadataManagerImpl(conf);
+ }
- Set<String> tables = omMetadataManager.listTableNames();
- Reflections reflections = new Reflections(
- "org.apache.hadoop.ozone.om.response");
- Set<Class<? extends OMClientResponse>> subTypes =
- reflections.getSubTypesOf(OMClientResponse.class);
- subTypes.forEach(aClass -> {
- Assert.assertTrue(aClass + "does not have annotation of" +
- " CleanupTableInfo",
+ @Test
+ public void checkAllWriteResponseHasCleanupTableAnnotation() {
+ getResponseClasses().forEach(aClass -> {
+ Assert.assertTrue(
+ aClass + "does not have annotation of CleanupTableInfo",
aClass.isAnnotationPresent(CleanupTableInfo.class));
- String[] cleanupTables =
- aClass.getAnnotation(CleanupTableInfo.class).cleanupTables();
- Assert.assertTrue(cleanupTables.length >=1);
- for (String tableName : cleanupTables) {
- Assert.assertTrue(tables.contains(tableName));
+ });
+ }
+
+ @Test
+ public void checkWriteResponseIsAnnotatedWithKnownTableNames()
+ throws Exception {
+ OMMetadataManager omMetadataManager = new OmMetadataManagerImpl(conf);
+ Set<String> tables = omMetadataManager.listTableNames();
+
+ getResponseClasses().forEach(aClass -> {
+
+ CleanupTableInfo annotation =
+ aClass.getAnnotation(CleanupTableInfo.class);
+ String[] cleanupTables = annotation.cleanupTables();
+ boolean cleanupAll = annotation.cleanupAll();
+
+ if (cleanupTables.length >= 1) {
+ assertTrue(
+ Arrays.stream(cleanupTables).allMatch(tables::contains)
+ );
+ } else {
+ assertTrue(cleanupAll);
}
+
});
}
+
+ private Set<Class<? extends OMClientResponse>> getResponseClasses() {
+ Reflections reflections =
+ new Reflections("org.apache.hadoop.ozone.om.response");
+ return reflections.getSubTypesOf(OMClientResponse.class);
+ }
}
diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/FinalizeUpgradeSubCommand.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/FinalizeUpgradeSubCommand.java
new file mode 100644
index 0000000..b35c621
--- /dev/null
+++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/FinalizeUpgradeSubCommand.java
@@ -0,0 +1,221 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hadoop.ozone.admin.om;
+
+import org.apache.hadoop.hdds.cli.HddsVersionProvider;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
+import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UpgradeFinalizationStatus;
+import picocli.CommandLine;
+
+import java.io.IOException;
+import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.INVALID_REQUEST;
+
+/**
+ * Handler of ozone admin om finalizeUpgrade command.
+ */
+@CommandLine.Command(
+ name = "finalizeupgrade",
+ description = "Finalizes Ozone Manager's metadata changes and enables new "
+ + "features after a software upgrade.\n"
+ + "It is possible to specify the service ID for an HA environment, "
+ + "or the Ozone manager host in a non-HA environment, if none provided "
+ + "the default from configuration is being used if not ambiguous.",
+ mixinStandardHelpOptions = true,
+ versionProvider = HddsVersionProvider.class
+)
+public class FinalizeUpgradeSubCommand implements Callable<Void> {
+
+ @CommandLine.ParentCommand
+ private OMAdmin parent;
+
+ @CommandLine.Option(
+ names = {"-id", "--service-id"},
+ description = "Ozone Manager Service ID"
+ )
+ private String omServiceId;
+
+ @CommandLine.Option(
+ names = {"-host", "--service-host"},
+ description = "Ozone Manager Host"
+ )
+ private String omHost;
+
+ @CommandLine.Option(
+ names = {"--takeover"},
+ description = "Forces takeover of monitoring from an other client, if "
+ + "finalization has already been started and did not finished yet."
+ )
+ private boolean force;
+
+ @Override
+ public Void call() throws Exception {
+ boolean forceHA = false;
+ OzoneManagerProtocol client =
+ parent.createOmClient(omServiceId, omHost, forceHA);
+ String upgradeClientID = "Upgrade-Client-" + UUID.randomUUID().toString();
+ try {
+ UpgradeFinalizationStatus status =
+ client.finalizeUpgrade(upgradeClientID);
+ if (isFinalized(status)){
+ System.out.println("Upgrade has already been finalized.");
+ emitExitMsg();
+ return null;
+ } else if (!isStarting(status)){
+ System.err.println("Invalid response from Ozone Manager.");
+ System.err.println(
+ "Current finalization status is: " + status.getStatus()
+ );
+ throw new IOException("Exiting...");
+ }
+ } catch (OMException e) {
+ handleInvalidRequestAfterInitiatingFinalization(e);
+ }
+ monitorAndWaitFinalization(client, upgradeClientID);
+ return null;
+ }
+
+ private void monitorAndWaitFinalization(OzoneManagerProtocol client,
+ String upgradeClientID) throws ExecutionException {
+ ExecutorService exec = Executors.newSingleThreadExecutor();
+ Future<?> monitor =
+ exec.submit(new UpgradeMonitor(client, upgradeClientID, force));
+ try {
+ monitor.get();
+ emitFinishedMsg();
+ } catch (CancellationException|InterruptedException e) {
+ emitCancellationMsg();
+ } catch (ExecutionException e) {
+ emitGeneralErrorMsg();
+ throw e;
+ } finally {
+ exec.shutdown();
+ }
+ }
+
+ private void handleInvalidRequestAfterInitiatingFinalization(
+ OMException e) throws IOException {
+ if (e.getResult().equals(INVALID_REQUEST)) {
+ if (force) {
+ return;
+ }
+ System.err.println("Finalization is already in progress, it is not"
+ + "possible to initiate it again.");
+ e.printStackTrace(System.err);
+ System.err.println("If you want to track progress from a new client"
+ + "for any reason, use --takeover, and the status update will be"
+ + "received by the new client. Note that with forcing to monitor"
+ + "progress from a new client, the old one initiated the upgrade"
+ + "will not be able to monitor the progress further and exit.");
+ throw new IOException("Exiting...");
+ } else {
+ throw e;
+ }
+ }
+
+ private static class UpgradeMonitor implements Callable<Void> {
+
+ private OzoneManagerProtocol client;
+ private String upgradeClientID;
+ private boolean force;
+
+ UpgradeMonitor(
+ OzoneManagerProtocol client,
+ String upgradeClientID,
+ boolean force
+ ) {
+ this.client = client;
+ this.upgradeClientID = upgradeClientID;
+ this.force = force;
+ }
+
+ @Override
+ public Void call() throws Exception {
+ boolean finished = false;
+ while (!finished) {
+ Thread.sleep(500);
+ // do not check for exceptions, if one happens during monitoring we
+ // should report it and exit.
+ UpgradeFinalizationStatus status =
+ client.queryUpgradeFinalizationProgress(upgradeClientID, force);
+ // this can happen after trying to takeover the request after the fact
+ // when there is already nothing to take over.
+ if (isFinalized(status)) {
+ System.out.println("Finalization already finished.");
+ emitExitMsg();
+ return null;
+ }
+ if (isInprogress(status) || isDone(status)) {
+ status.getMessagesList().stream().forEachOrdered(System.out::println);
+ }
+ if (isDone(status)) {
+ emitExitMsg();
+ finished = true;
+ }
+ }
+ return null;
+ }
+
+ }
+ private static void emitExitMsg() {
+ System.out.println("Exiting...");
+ }
+
+ private static boolean isFinalized(UpgradeFinalizationStatus status) {
+ return status.getStatus()
+ .equals(UpgradeFinalizationStatus.Status.ALREADY_FINALIZED);
+ }
+
+ private static boolean isDone(UpgradeFinalizationStatus status) {
+ return status.getStatus()
+ .equals(UpgradeFinalizationStatus.Status.FINALIZATION_DONE);
+ }
+
+ private static boolean isInprogress(UpgradeFinalizationStatus status) {
+ return status.getStatus()
+ .equals(UpgradeFinalizationStatus.Status.FINALIZATION_IN_PROGRESS);
+ }
+
+ private static boolean isStarting(UpgradeFinalizationStatus status) {
+ return status.getStatus()
+ .equals(UpgradeFinalizationStatus.Status.STARTING_FINALIZATION);
+ }
+
+ private static void emitGeneralErrorMsg() {
+ System.err.println("Finalization was not successful.");
+ }
+
+ private static void emitFinishedMsg() {
+ System.out.println("Finalization of Ozone Manager's metadata upgrade "
+ + "finished.");
+ }
+
+ private static void emitCancellationMsg() {
+ System.err.println("Finalization command was cancelled. Note that, this"
+ + "will not cancel finalization in Ozone Manager. Progress can be"
+ + "monitored in the Ozone Manager's log.");
+ }
+}
diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/OMAdmin.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/OMAdmin.java
index f9321ab..317c464 100644
--- a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/OMAdmin.java
+++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/OMAdmin.java
@@ -34,6 +34,7 @@ import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolClientSideTrans
import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB;
import org.apache.hadoop.security.UserGroupInformation;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SERVICE_IDS_KEY;
import org.apache.ratis.protocol.ClientId;
import org.kohsuke.MetaInfServices;
@@ -41,6 +42,8 @@ import picocli.CommandLine;
import picocli.CommandLine.Model.CommandSpec;
import picocli.CommandLine.Spec;
+import java.util.Collection;
+
/**
* Subcommand for admin operations related to OM.
*/
@@ -50,6 +53,7 @@ import picocli.CommandLine.Spec;
mixinStandardHelpOptions = true,
versionProvider = HddsVersionProvider.class,
subcommands = {
+ FinalizeUpgradeSubCommand.class,
GetServiceRolesSubcommand.class
})
@MetaInfServices(SubcommandWithParent.class)
@@ -86,13 +90,28 @@ public class OMAdmin extends GenericCli implements SubcommandWithParent {
}
public OzoneManagerProtocolClientSideTranslatorPB createOmClient(
- String omServiceID) throws Exception {
+ String omServiceID
+ ) throws Exception {
+ return createOmClient(omServiceID, null, true);
+ }
+
+ public OzoneManagerProtocolClientSideTranslatorPB createOmClient(
+ String omServiceID,
+ String omHost,
+ boolean forceHA
+ ) throws Exception {
OzoneConfiguration conf = parent.getOzoneConf();
+ if (omHost != null && !omHost.isEmpty()) {
+ omServiceID = null;
+ conf.set(OZONE_OM_ADDRESS_KEY, omHost);
+ } else if (omServiceID == null || omServiceID.isEmpty()) {
+ omServiceID = getTheOnlyConfiguredOmServiceIdOrThrow();
+ }
UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
RPC.setProtocolEngine(conf, OzoneManagerProtocolPB.class,
ProtobufRpcEngine.class);
String clientId = ClientId.randomId().toString();
- if (OmUtils.isOmHAServiceId(conf, omServiceID)) {
+ if (!forceHA || (forceHA && OmUtils.isOmHAServiceId(conf, omServiceID))) {
OmTransport omTransport = new Hadoop3OmTransportFactory()
.createOmTransport(conf, ugi, omServiceID);
return new OzoneManagerProtocolClientSideTranslatorPB(omTransport,
@@ -106,6 +125,22 @@ public class OMAdmin extends GenericCli implements SubcommandWithParent {
}
}
+ private String getTheOnlyConfiguredOmServiceIdOrThrow() {
+ if (getConfiguredServiceIds().size() != 1) {
+ throw new IllegalArgumentException("There is no Ozone Manager service ID"
+ + "specified, but there are either zero, or more than one service "
+ + "configured. Please specify the service ID to be finalized.");
+ }
+ return getConfiguredServiceIds().iterator().next();
+ }
+
+ private Collection<String> getConfiguredServiceIds() {
+ OzoneConfiguration conf = parent.getOzoneConf();
+ Collection<String> omServiceIds =
+ conf.getTrimmedStringCollection(OZONE_OM_SERVICE_IDS_KEY);
+ return omServiceIds;
+ }
+
@Override
public Class<?> getParentType() {
return OzoneAdmin.class;
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org