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 2021/01/04 19:23:11 UTC
[ozone] branch HDDS-3698-upgrade updated: HDDS-4569. Add pre append
gate and marker file to OM prepare state. (#1705)
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/ozone.git
The following commit(s) were added to refs/heads/HDDS-3698-upgrade by this push:
new 17a126d HDDS-4569. Add pre append gate and marker file to OM prepare state. (#1705)
17a126d is described below
commit 17a126da1775d45f8843b47985bb5deb0ea3e928
Author: Ethan Rose <33...@users.noreply.github.com>
AuthorDate: Mon Jan 4 14:22:54 2021 -0500
HDDS-4569. Add pre append gate and marker file to OM prepare state. (#1705)
---
.../java/org/apache/hadoop/ozone/OzoneConsts.java | 4 +
.../server/ratis/ContainerStateMachine.java | 7 +-
.../transport/server/ratis/XceiverServerRatis.java | 7 +-
.../ozone/container/ContainerTestHelper.java | 23 +-
.../hadoop/ozone/om/exceptions/OMException.java | 3 +-
.../ozone/om/ha/OMFailoverProxyProvider.java | 12 +
.../apache/hadoop/ozone/container/TestHelper.java | 22 +-
.../ozone/freon/TestHadoopDirTreeGenerator.java | 4 +-
.../ozone/freon/TestHadoopNestedDirGenerator.java | 4 +-
.../ozone/freon/TestOzoneClientKeyGenerator.java | 4 +-
.../hadoop/ozone/om/TestOMRatisSnapshots.java | 2 +-
.../hadoop/ozone/om/TestOzoneManagerPrepare.java | 349 +++++++++++++--------
.../src/main/proto/OmClientProtocol.proto | 1 +
.../hadoop/ozone/om/ratis/OMTransactionInfo.java | 2 +-
.../org/apache/hadoop/ozone/om/OzoneManager.java | 53 ++--
.../hadoop/ozone/om/OzoneManagerPrepareState.java | 282 +++++++++++++++++
.../hadoop/ozone/om/ratis/OMRatisSnapshotInfo.java | 2 +-
.../ozone/om/ratis/OzoneManagerRatisServer.java | 28 +-
.../ozone/om/ratis/OzoneManagerStateMachine.java | 33 +-
.../ozone/om/request/upgrade/OMPrepareRequest.java | 39 ++-
.../protocolPB/OzoneManagerRequestHandler.java | 31 +-
.../om/ratis/TestOzoneManagerRatisServer.java | 4 +-
.../om/ratis/TestOzoneManagerStateMachine.java | 94 ++++++
.../om/upgrade/TestOzoneManagerPrepareState.java | 269 ++++++++++++++++
pom.xml | 2 +-
25 files changed, 1041 insertions(+), 240 deletions(-)
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
index a7aca16..e1d1b0d 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
@@ -381,4 +381,8 @@ public final class OzoneConsts {
// An on-disk transient marker file used when replacing DB with checkpoint
public static final String DB_TRANSIENT_MARKER = "dbInconsistentMarker";
+
+ // An on-disk marker file used to indicate that the OM is in prepare and
+ // should remain prepared even after a restart.
+ public static final String PREPARE_MARKER = "prepareMarker";
}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java
index 1a87ce5..42373bd 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java
@@ -71,7 +71,6 @@ import org.apache.ratis.protocol.RaftGroupMemberId;
import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.protocol.exceptions.StateMachineException;
import org.apache.ratis.server.RaftServer;
-import org.apache.ratis.server.impl.RaftServerProxy;
import org.apache.ratis.server.protocol.TermIndex;
import org.apache.ratis.server.raftlog.RaftLog;
import org.apache.ratis.server.storage.RaftStorage;
@@ -220,8 +219,7 @@ public class ContainerStateMachine extends BaseStateMachine {
private long loadSnapshot(SingleFileSnapshotInfo snapshot)
throws IOException {
if (snapshot == null) {
- TermIndex empty =
- TermIndex.newTermIndex(0, RaftLog.INVALID_LOG_INDEX);
+ TermIndex empty = TermIndex.valueOf(0, RaftLog.INVALID_LOG_INDEX);
LOG.info("{}: The snapshot info is null. Setting the last applied index" +
"to:{}", gid, empty);
setLastAppliedTermIndex(empty);
@@ -421,9 +419,8 @@ public class ContainerStateMachine extends BaseStateMachine {
long startTime) {
final WriteChunkRequestProto write = requestProto.getWriteChunk();
RaftServer server = ratisServer.getServer();
- Preconditions.checkState(server instanceof RaftServerProxy);
try {
- if (((RaftServerProxy) server).getImpl(gid).isLeader()) {
+ if (server.getDivision(gid).getInfo().isLeader()) {
stateMachineDataCache.put(entryIndex, write.getData());
}
} catch (InterruptedException ioe) {
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java
index 9bc2eb2..eca0b1c 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java
@@ -100,7 +100,6 @@ import org.apache.ratis.rpc.RpcType;
import org.apache.ratis.rpc.SupportedRpcType;
import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.RaftServerConfigKeys;
-import org.apache.ratis.server.impl.RaftServerProxy;
import org.apache.ratis.server.protocol.TermIndex;
import org.apache.ratis.util.SizeInBytes;
import org.apache.ratis.util.TimeDuration;
@@ -460,9 +459,9 @@ public final class XceiverServerRatis implements XceiverServerSpi {
}
server.start();
- int realPort =
- ((RaftServerProxy) server).getServerRpc().getInetSocketAddress()
- .getPort();
+ int realPort = server.getServerRpc()
+ .getInetSocketAddress()
+ .getPort();
if (port == 0) {
LOG.info("{} {} is started using port {}", getClass().getSimpleName(),
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ContainerTestHelper.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ContainerTestHelper.java
index 5b25dba..24c598a 100644
--- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ContainerTestHelper.java
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ContainerTestHelper.java
@@ -51,8 +51,7 @@ import org.apache.hadoop.security.token.Token;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import org.apache.ratis.protocol.RaftGroupId;
-import org.apache.ratis.server.impl.RaftServerImpl;
-import org.apache.ratis.server.impl.RaftServerProxy;
+import org.apache.ratis.server.RaftServer;
import org.apache.ratis.statemachine.StateMachine;
import org.junit.Assert;
import org.slf4j.Logger;
@@ -583,35 +582,35 @@ public final class ContainerTestHelper {
return String.format("%1$" + length + "s", string);
}
- private static RaftServerImpl getRaftServerImpl(HddsDatanodeService dn,
- Pipeline pipeline) throws Exception {
+ private static RaftServer.Division getRaftServerDivision(
+ HddsDatanodeService dn, Pipeline pipeline) throws Exception {
if (!pipeline.getNodes().contains(dn.getDatanodeDetails())) {
throw new IllegalArgumentException("Pipeline:" + pipeline.getId() +
" not exist in datanode:" + dn.getDatanodeDetails().getUuid());
}
- XceiverServerSpi server = dn.getDatanodeStateMachine().
+ XceiverServerSpi serverSpi = dn.getDatanodeStateMachine().
getContainer().getWriteChannel();
- RaftServerProxy proxy =
- (RaftServerProxy) (((XceiverServerRatis) server).getServer());
+ RaftServer server = (((XceiverServerRatis) serverSpi).getServer());
RaftGroupId groupId =
- pipeline == null ? proxy.getGroupIds().iterator().next() :
+ pipeline == null ? server.getGroupIds().iterator().next() :
RatisHelper.newRaftGroup(pipeline).getGroupId();
- return proxy.getImpl(groupId);
+
+ return server.getDivision(groupId);
}
public static StateMachine getStateMachine(HddsDatanodeService dn,
Pipeline pipeline) throws Exception {
- return getRaftServerImpl(dn, pipeline).getStateMachine();
+ return getRaftServerDivision(dn, pipeline).getStateMachine();
}
public static boolean isRatisLeader(HddsDatanodeService dn, Pipeline pipeline)
throws Exception {
- return getRaftServerImpl(dn, pipeline).isLeader();
+ return getRaftServerDivision(dn, pipeline).getInfo().isLeader();
}
public static boolean isRatisFollower(HddsDatanodeService dn,
Pipeline pipeline) throws Exception {
- return getRaftServerImpl(dn, pipeline).isFollower();
+ return getRaftServerDivision(dn, pipeline).getInfo().isFollower();
}
}
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/exceptions/OMException.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/exceptions/OMException.java
index c1e8148..2c4d89d 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/exceptions/OMException.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/exceptions/OMException.java
@@ -237,6 +237,7 @@ public class OMException extends IOException {
REMOVE_UPGRADE_TO_LAYOUT_VERSION_FAILED,
UPDATE_LAYOUT_VERSION_FAILED,
LAYOUT_FEATURE_FINALIZATION_FAILED,
- PREPARE_FAILED
+ PREPARE_FAILED,
+ NOT_SUPPORTED_OPERATION_WHEN_PREPARED
}
}
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProvider.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProvider.java
index 0d698fd..a452dc1 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProvider.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProvider.java
@@ -49,12 +49,14 @@ import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.ozone.OmUtils;
import org.apache.hadoop.ozone.OzoneConfigKeys;
import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException;
import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException;
import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolClientSideTranslatorPB;
import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.ratis.protocol.exceptions.StateMachineException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -468,6 +470,16 @@ public class OMFailoverProxyProvider implements
return false;
}
}
+ } else if (ex instanceof StateMachineException) {
+ StateMachineException smEx = (StateMachineException) ex;
+ Throwable cause = smEx.getCause();
+ if (cause instanceof OMException) {
+ OMException omEx = (OMException) cause;
+ // Do not failover if the operation was blocked because the OM was
+ // prepared.
+ return omEx.getResult() !=
+ OMException.ResultCodes.NOT_SUPPORTED_OPERATION_WHEN_PREPARED;
+ }
}
return true;
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/TestHelper.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/TestHelper.java
index 12ffce6..8c18262 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/TestHelper.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/TestHelper.java
@@ -47,8 +47,7 @@ import org.apache.hadoop.ozone.container.common.transport.server.ratis.XceiverSe
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.ratis.protocol.RaftGroupId;
-import org.apache.ratis.server.impl.RaftServerImpl;
-import org.apache.ratis.server.impl.RaftServerProxy;
+import org.apache.ratis.server.RaftServer;
import org.apache.ratis.statemachine.StateMachine;
import org.junit.Assert;
@@ -299,30 +298,27 @@ public final class TestHelper {
return getStateMachine(cluster.getHddsDatanodes().get(0), null);
}
- private static RaftServerImpl getRaftServerImpl(HddsDatanodeService dn,
- Pipeline pipeline) throws Exception {
- XceiverServerSpi server = dn.getDatanodeStateMachine().
+ private static RaftServer.Division getRaftServerDivision(
+ HddsDatanodeService dn, Pipeline pipeline) throws Exception {
+ XceiverServerSpi serverSpi = dn.getDatanodeStateMachine().
getContainer().getWriteChannel();
- RaftServerProxy proxy =
- (RaftServerProxy) (((XceiverServerRatis) server).getServer());
+ RaftServer server = (((XceiverServerRatis) serverSpi).getServer());
RaftGroupId groupId =
- pipeline == null ? proxy.getGroupIds().iterator().next() :
+ pipeline == null ? server.getGroupIds().iterator().next() :
RatisHelper.newRaftGroup(pipeline).getGroupId();
- return proxy.getImpl(groupId);
+ return server.getDivision(groupId);
}
public static StateMachine getStateMachine(HddsDatanodeService dn,
Pipeline pipeline) throws Exception {
- return getRaftServerImpl(dn, pipeline).getStateMachine();
+ return getRaftServerDivision(dn, pipeline).getStateMachine();
}
public static HddsDatanodeService getDatanodeService(OmKeyLocationInfo info,
- MiniOzoneCluster cluster)
- throws IOException {
+ MiniOzoneCluster cluster) throws IOException {
DatanodeDetails dnDetails = info.getPipeline().
getFirstNode();
return cluster.getHddsDatanodes().get(cluster.
getHddsDatanodeIndex(dnDetails));
}
-
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopDirTreeGenerator.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopDirTreeGenerator.java
index 6f5b113..6ad5b8d 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopDirTreeGenerator.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopDirTreeGenerator.java
@@ -26,7 +26,7 @@ import org.apache.hadoop.ozone.client.ObjectStore;
import org.apache.hadoop.ozone.client.OzoneClientFactory;
import org.apache.hadoop.ozone.client.OzoneVolume;
import org.apache.hadoop.test.GenericTestUtils;
-import org.apache.ratis.server.impl.RaftServerImpl;
+import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.raftlog.RaftLog;
import org.junit.Assert;
import org.junit.Before;
@@ -53,7 +53,7 @@ public class TestHadoopDirTreeGenerator {
path = GenericTestUtils
.getTempPath(TestOzoneClientKeyGenerator.class.getSimpleName());
GenericTestUtils.setLogLevel(RaftLog.LOG, Level.DEBUG);
- GenericTestUtils.setLogLevel(RaftServerImpl.LOG, Level.DEBUG);
+ GenericTestUtils.setLogLevel(RaftServer.LOG, Level.DEBUG);
File baseDir = new File(path);
baseDir.mkdirs();
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopNestedDirGenerator.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopNestedDirGenerator.java
index 69731ea..42f6fd8 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopNestedDirGenerator.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestHadoopNestedDirGenerator.java
@@ -26,7 +26,7 @@ import org.apache.hadoop.ozone.client.ObjectStore;
import org.apache.hadoop.ozone.client.OzoneClientFactory;
import org.apache.hadoop.ozone.client.OzoneVolume;
import org.apache.hadoop.test.GenericTestUtils;
-import org.apache.ratis.server.impl.RaftServerImpl;
+import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.raftlog.RaftLog;
import java.util.LinkedList;
import org.junit.Assert;
@@ -57,7 +57,7 @@ public class TestHadoopNestedDirGenerator {
path = GenericTestUtils
.getTempPath(TestOzoneClientKeyGenerator.class.getSimpleName());
GenericTestUtils.setLogLevel(RaftLog.LOG, Level.DEBUG);
- GenericTestUtils.setLogLevel(RaftServerImpl.LOG, Level.DEBUG);
+ GenericTestUtils.setLogLevel(RaftServer.LOG, Level.DEBUG);
File baseDir = new File(path);
baseDir.mkdirs();
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestOzoneClientKeyGenerator.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestOzoneClientKeyGenerator.java
index 055f114..c3af704 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestOzoneClientKeyGenerator.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestOzoneClientKeyGenerator.java
@@ -25,7 +25,7 @@ import org.apache.hadoop.ozone.MiniOzoneCluster;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.commons.io.FileUtils;
-import org.apache.ratis.server.impl.RaftServerImpl;
+import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.raftlog.RaftLog;
import org.junit.Before;
import org.junit.Test;
@@ -48,7 +48,7 @@ public class TestOzoneClientKeyGenerator {
path = GenericTestUtils
.getTempPath(TestOzoneClientKeyGenerator.class.getSimpleName());
GenericTestUtils.setLogLevel(RaftLog.LOG, Level.DEBUG);
- GenericTestUtils.setLogLevel(RaftServerImpl.LOG, Level.DEBUG);
+ GenericTestUtils.setLogLevel(RaftServer.LOG, Level.DEBUG);
File baseDir = new File(path);
baseDir.mkdirs();
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMRatisSnapshots.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMRatisSnapshots.java
index 1465d95..ac0beb6 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMRatisSnapshots.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMRatisSnapshots.java
@@ -155,7 +155,7 @@ public class TestOMRatisSnapshots {
OMTransactionInfo omTransactionInfo =
OMTransactionInfo.readTransactionInfo(leaderOM.getMetadataManager());
TermIndex leaderOMTermIndex =
- TermIndex.newTermIndex(omTransactionInfo.getTerm(),
+ TermIndex.valueOf(omTransactionInfo.getTerm(),
omTransactionInfo.getTransactionIndex());
long leaderOMSnaphsotIndex = leaderOMTermIndex.getIndex();
long leaderOMSnapshotTermIndex = leaderOMTermIndex.getTerm();
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerPrepare.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerPrepare.java
index e5104a8..9fe873a 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerPrepare.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerPrepare.java
@@ -22,24 +22,28 @@ import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
import java.io.File;
import java.nio.file.Paths;
+import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import org.apache.hadoop.hdds.client.ReplicationFactor;
import org.apache.hadoop.hdds.client.ReplicationType;
import org.apache.hadoop.ozone.MiniOzoneHAClusterImpl;
import org.apache.hadoop.ozone.client.ObjectStore;
-import org.apache.hadoop.ozone.client.OzoneClient;
-import org.apache.hadoop.ozone.client.OzoneClientFactory;
import org.apache.hadoop.ozone.client.OzoneVolume;
import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
import org.apache.hadoop.ozone.container.ContainerTestHelper;
import org.apache.hadoop.ozone.container.TestHelper;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
-import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol;
import org.apache.hadoop.ozone.om.ratis.OMTransactionInfo;
import org.apache.hadoop.test.LambdaTestUtils;
import org.junit.Assert;
@@ -49,11 +53,24 @@ import org.junit.Test;
* Test OM prepare against actual mini cluster.
*/
public class TestOzoneManagerPrepare extends TestOzoneManagerHA {
-
- private final String keyPrefix = "key";
- private final int timeoutMillis = 30000;
- private final static Long PREPARE_FLUSH_WAIT_TIMEOUT_SECONDS = 120L;
- private final static Long PREPARE_FLUSH_INTERVAL_SECONDS = 5L;
+ private static final String BUCKET = "bucket";
+ private static final String VOLUME = "volume";
+ private static final String KEY_PREFIX = "key";
+
+ // Maximum time to wait for conditions involving Ratis logs.
+ private static final int WAIT_TIMEOUT_MILLIS = 120000;
+ private final static long PREPARE_FLUSH_WAIT_TIMEOUT_SECONDS = 120L;
+ private final static long PREPARE_FLUSH_INTERVAL_SECONDS = 5L;
+
+ private MiniOzoneHAClusterImpl cluster;
+ private ClientProtocol clientProtocol;
+ private ObjectStore store;
+
+ public void setup() throws Exception {
+ cluster = getCluster();
+ store = getObjectStore();
+ clientProtocol = store.getClientProxy();
+ }
/**
* Calls prepare on all OMs when they have no transaction information.
@@ -61,19 +78,10 @@ public class TestOzoneManagerPrepare extends TestOzoneManagerHA {
*/
@Test
public void testPrepareWithoutTransactions() throws Exception {
- MiniOzoneHAClusterImpl cluster = getCluster();
- ClientProtocol ozClient = OzoneClientFactory.getRpcClient(getConf())
- .getObjectStore().getClientProxy();
- long prepareRequestLogIndex =
- ozClient.getOzoneManagerClient().prepareOzoneManager(
- PREPARE_FLUSH_WAIT_TIMEOUT_SECONDS, PREPARE_FLUSH_INTERVAL_SECONDS);
-
- // Prepare response processing is included in the snapshot,
- // giving index of 1.
- Assert.assertEquals(1, prepareRequestLogIndex);
- for (OzoneManager om: cluster.getOzoneManagersList()) {
- waitAndCheckPrepared(om, prepareRequestLogIndex);
- }
+ setup();
+ long prepareIndex = submitPrepareRequest();
+ assertClusterPrepared(prepareIndex);
+ assertRatisLogsCleared();
}
/**
@@ -82,48 +90,14 @@ public class TestOzoneManagerPrepare extends TestOzoneManagerHA {
*/
@Test
public void testPrepareWithTransactions() throws Exception {
- MiniOzoneHAClusterImpl cluster = getCluster();
- OzoneClient ozClient = OzoneClientFactory.getRpcClient(getConf());
-
- String volumeName = UUID.randomUUID().toString();
- String bucketName = UUID.randomUUID().toString();
- ObjectStore store = ozClient.getObjectStore();
-
- store.createVolume(volumeName);
- OzoneVolume volume = store.getVolume(volumeName);
- volume.createBucket(bucketName);
-
- Set<String> writtenKeys = new HashSet<>();
- for (int i = 1; i <= 10; i++) {
- String keyName = keyPrefix + i;
- writeTestData(store, volumeName, bucketName, keyName);
- writtenKeys.add(keyName);
- }
-
- // Make sure all OMs have logs from writing data, so we can check that
- // they are purged after prepare.
- for (OzoneManager om: cluster.getOzoneManagersList()) {
- LambdaTestUtils.await(timeoutMillis, 1000,
- () -> logFilesPresentInRatisPeer(om));
- }
-
- OzoneManagerProtocol ozoneManagerClient =
- ozClient.getObjectStore().getClientProxy().getOzoneManagerClient();
- long prepareRequestLogIndex =
- ozoneManagerClient.prepareOzoneManager(
- PREPARE_FLUSH_WAIT_TIMEOUT_SECONDS, PREPARE_FLUSH_INTERVAL_SECONDS);
+ setup();
+ Set<String> writtenKeys = writeKeysAndWaitForLogs(50);
+ long prepareIndex = submitPrepareRequest();
// Make sure all OMs are prepared and all OMs still have their data.
- for (OzoneManager om: cluster.getOzoneManagersList()) {
- waitAndCheckPrepared(om, prepareRequestLogIndex);
- List<OmKeyInfo> keys = om.getMetadataManager().listKeys(volumeName,
- bucketName, null, keyPrefix, 100);
-
- Assert.assertEquals(writtenKeys.size(), keys.size());
- for (OmKeyInfo keyInfo: keys) {
- Assert.assertTrue(writtenKeys.contains(keyInfo.getKeyName()));
- }
- }
+ assertClusterPrepared(prepareIndex);
+ assertRatisLogsCleared();
+ assertKeysWritten(writtenKeys);
}
/**
@@ -136,79 +110,121 @@ public class TestOzoneManagerPrepare extends TestOzoneManagerHA {
* Checks that third OM received all transactions and is prepared.
* @throws Exception
*/
-// @Test
+ // TODO: This test should be passing after HDDS-4610 and RATIS-1241
+ // @Test
public void testPrepareDownedOM() throws Exception {
+ setup();
// Index of the OM that will be shut down during this test.
final int shutdownOMIndex = 2;
-
- MiniOzoneHAClusterImpl cluster = getCluster();
- OzoneClient ozClient = OzoneClientFactory.getRpcClient(getConf());
-
- String volumeName = UUID.randomUUID().toString();
- String bucketName = UUID.randomUUID().toString();
- ObjectStore store = ozClient.getObjectStore();
+ List<OzoneManager> runningOms = cluster.getOzoneManagersList();
// Create keys with all 3 OMs up.
- store.createVolume(volumeName);
- OzoneVolume volume = store.getVolume(volumeName);
- volume.createBucket(bucketName);
-
- Set<String> writtenKeys = new HashSet<>();
- for (int i = 1; i <= 50; i++) {
- String keyName = keyPrefix + i;
- writeTestData(store, volumeName, bucketName, keyName);
- writtenKeys.add(keyName);
- }
-
- // Make sure all OMs have logs from writing data, so we can check that
- // they are purged after prepare.
- for (OzoneManager om: cluster.getOzoneManagersList()) {
- LambdaTestUtils.await(timeoutMillis, 1000,
- () -> logFilesPresentInRatisPeer(om));
- }
+ Set<String> writtenKeys = writeKeysAndWaitForLogs(10, runningOms);
// Shut down one OM.
cluster.stopOzoneManager(shutdownOMIndex);
OzoneManager downedOM = cluster.getOzoneManager(shutdownOMIndex);
Assert.assertFalse(downedOM.isRunning());
+ Assert.assertEquals(runningOms.remove(shutdownOMIndex), downedOM);
// Write keys with the remaining OMs up.
- for (int i = 51; i <= 100; i++) {
- String keyName = keyPrefix + i;
- writeTestData(store, volumeName, bucketName, keyName);
- writtenKeys.add(keyName);
- }
+ writtenKeys.addAll(
+ writeKeysAndWaitForLogs(10, runningOms));
- OzoneManagerProtocol ozoneManagerClient =
- ozClient.getObjectStore().getClientProxy().getOzoneManagerClient();
- long prepareIndex = ozoneManagerClient.prepareOzoneManager(
- PREPARE_FLUSH_WAIT_TIMEOUT_SECONDS, PREPARE_FLUSH_INTERVAL_SECONDS);
+ long prepareIndex = submitPrepareRequest();
// Check that the two live OMs are prepared.
- for (OzoneManager om: cluster.getOzoneManagersList()) {
- if (om != downedOM) {
- // Follower may still be applying transactions.
- waitAndCheckPrepared(om, prepareIndex);
- }
- }
+ assertClusterPrepared(prepareIndex, runningOms);
// Restart the downed OM and wait for it to catch up.
// Since prepare was the last Ratis transaction, it should have all data
// it missed once it receives the prepare transaction.
cluster.restartOzoneManager(downedOM, true);
- LambdaTestUtils.await(timeoutMillis, 2000,
- () -> checkPrepared(downedOM, prepareIndex));
+ runningOms.add(shutdownOMIndex, downedOM);
// Make sure all OMs are prepared and still have data.
- for (OzoneManager om: cluster.getOzoneManagersList()) {
- List<OmKeyInfo> readKeys = om.getMetadataManager().listKeys(volumeName,
- bucketName, null, keyPrefix, 100);
+ assertClusterPrepared(prepareIndex, runningOms);
+ assertKeysWritten(writtenKeys, runningOms);
+ }
+
+ // TODO: This test should be passing after HDDS-4610 and RATIS-1241
+ // @Test
+ public void testPrepareWithRestart() throws Exception {
+ setup();
+ writeKeysAndWaitForLogs(10);
+ long prepareIndex = submitPrepareRequest();
+ assertClusterPrepared(prepareIndex);
+
+ // Restart all ozone managers.
+ cluster.restartOzoneManager();
+
+ // No check for cleared logs, since Ratis meta transactions may slip in
+ // on restart.
+ assertClusterPrepared(prepareIndex);
+ }
+
+ /**
+ * Issues requests on ten different threads, for which one is a prepare and
+ * the rest are create volume. We cannot be sure of the exact order that
+ * the requests will execute, so this test checks that the cluster ends in
+ * a prepared state, and that create volume requests either succeed, or fail
+ * indicating the cluster was prepared before they were encountered.
+ * @throws Exception
+ */
+ @Test
+ public void testPrepareWithMultipleThreads() throws Exception {
+ setup();
+ final int numThreads = 10;
+ final int prepareTaskIndex = 5;
+
+ ExecutorService executorService = Executors.newFixedThreadPool(numThreads);
+ // For the prepare task, the future will return a log index.
+ // For the create volume tasks, 0 (dummy value) will be returned.
+ List<Future<Long>> tasks = new ArrayList<>();
+
+ for (int i = 0; i < numThreads; i++) {
+ Callable<Long> task;
+ if (i == prepareTaskIndex) {
+ task = this::submitPrepareRequest;
+ } else {
+ String volumeName = VOLUME + i;
+ task = () -> {
+ clientProtocol.createVolume(volumeName);
+ return 0L;
+ };
+ }
+ tasks.add(executorService.submit(task));
+ }
- Assert.assertEquals(writtenKeys.size(), readKeys.size());
- for (OmKeyInfo keyInfo: readKeys) {
- Assert.assertTrue(writtenKeys.contains(keyInfo.getKeyName()));
+ // For each task, wait for it to complete and check its result.
+ for (int i = 0; i < numThreads; i++) {
+ Future<Long> future = tasks.get(i);
+
+ if (i == prepareTaskIndex) {
+ assertClusterPrepared(future.get());
+ assertRatisLogsCleared();
+ } else {
+ try {
+ // If this throws an exception, it should be an OMException
+ // indicating failure because the cluster was already prepared.
+ // If no exception is thrown, the volume should be created.
+ future.get();
+ String volumeName = VOLUME + i;
+ Assert.assertTrue(clientProtocol.listVolumes(volumeName, "", 1)
+ .stream()
+ .anyMatch((vol) -> vol.getName().equals(volumeName)));
+ } catch (ExecutionException ex) {
+ Throwable cause = ex.getCause();
+ Assert.assertTrue(cause instanceof OMException);
+ Assert.assertEquals(
+ OMException.ResultCodes.NOT_SUPPORTED_OPERATION_WHEN_PREPARED,
+ ((OMException) cause).getResult());
+ }
}
}
+
+ // In the above loop, we have waited for all threads to terminate.
+ executorService.shutdown();
}
private boolean logFilesPresentInRatisPeer(OzoneManager om) {
@@ -228,9 +244,37 @@ public class TestOzoneManagerPrepare extends TestOzoneManagerHA {
return false;
}
- private void writeTestData(ObjectStore store, String volumeName,
- String bucketName, String keyName)
- throws Exception {
+ private Set<String> writeKeysAndWaitForLogs(int numKeys) throws Exception {
+ return writeKeysAndWaitForLogs(numKeys, cluster.getOzoneManagersList());
+ }
+
+ private Set<String> writeKeysAndWaitForLogs(int numKeys,
+ List<OzoneManager> ozoneManagers) throws Exception {
+
+ store.createVolume(VOLUME);
+ OzoneVolume volume = store.getVolume(VOLUME);
+ volume.createBucket(BUCKET);
+
+ Set<String> writtenKeys = new HashSet<>();
+ for (int i = 1; i <= numKeys; i++) {
+ String keyName = KEY_PREFIX + i;
+ writeTestData(VOLUME, BUCKET, keyName);
+ writtenKeys.add(keyName);
+ }
+
+ // Make sure all OMs have logs from writing data, so we can check that
+ // they are purged after prepare.
+ for (OzoneManager om: ozoneManagers) {
+ LambdaTestUtils.await(WAIT_TIMEOUT_MILLIS, 1000,
+ () -> logFilesPresentInRatisPeer(om));
+ }
+
+ return writtenKeys;
+ }
+
+ private void writeTestData(String volumeName,
+ String bucketName, String keyName) throws Exception {
+
String keyString = UUID.randomUUID().toString();
byte[] data = ContainerTestHelper.getFixedLengthString(
keyString, 100).getBytes(UTF_8);
@@ -241,23 +285,74 @@ public class TestOzoneManagerPrepare extends TestOzoneManagerHA {
keyStream.close();
}
- private void waitAndCheckPrepared(OzoneManager om,
- long prepareRequestLogIndex) throws Exception {
- // Log files are deleted after the snapshot is taken,
- // So once log files have been deleted, OM should be prepared.
- LambdaTestUtils.await(timeoutMillis, 1000,
- () -> !logFilesPresentInRatisPeer(om));
- Assert.assertTrue(checkPrepared(om, prepareRequestLogIndex));
+ private void assertKeysWritten(Set<String> expectedKeys) throws Exception {
+ assertKeysWritten(expectedKeys, cluster.getOzoneManagersList());
+ }
+
+ private void assertKeysWritten(Set<String> expectedKeys,
+ List<OzoneManager> ozoneManagers) throws Exception {
+ for (OzoneManager om: ozoneManagers) {
+ List<OmKeyInfo> keys = om.getMetadataManager().listKeys(VOLUME,
+ BUCKET, null, KEY_PREFIX, 100);
+
+ Assert.assertEquals(expectedKeys.size(), keys.size());
+ for (OmKeyInfo keyInfo: keys) {
+ Assert.assertTrue(expectedKeys.contains(keyInfo.getKeyName()));
+ }
+ }
+ }
+
+ private long submitPrepareRequest() throws Exception {
+ return clientProtocol.getOzoneManagerClient()
+ .prepareOzoneManager(PREPARE_FLUSH_WAIT_TIMEOUT_SECONDS,
+ PREPARE_FLUSH_INTERVAL_SECONDS);
}
- private boolean checkPrepared(OzoneManager om, long prepareRequestLogIndex)
+ private void assertClusterPrepared(long preparedIndex) throws Exception {
+ assertClusterPrepared(preparedIndex, cluster.getOzoneManagersList());
+ }
+
+ private void assertClusterPrepared(long preparedIndex,
+ List<OzoneManager> ozoneManagers) throws Exception {
+
+ for (OzoneManager om : ozoneManagers) {
+ // Wait for each OM to be running and transaction info to match to know
+ // it is prepared.
+ LambdaTestUtils.await(WAIT_TIMEOUT_MILLIS,
+ 1000, () -> {
+ if (!om.isRunning()) {
+ return false;
+ } else {
+ OMTransactionInfo txnInfo = om.getMetadataManager()
+ .getTransactionInfoTable().get(TRANSACTION_INFO_KEY);
+ return txnInfo.getTransactionIndex() == preparedIndex;
+ }
+ });
+ }
+
+ // Submitting a read request should pass.
+ clientProtocol.listVolumes(VOLUME, "", 100);
+
+ // Submitting write request should fail.
+ try {
+ clientProtocol.createVolume("vol");
+ Assert.fail("Write request should fail when OM is in prepare mode.");
+ } catch (OMException ex) {
+ Assert.assertEquals(
+ OMException.ResultCodes.NOT_SUPPORTED_OPERATION_WHEN_PREPARED,
+ ex.getResult());
+ }
+ }
+
+ private void assertRatisLogsCleared() throws Exception {
+ assertRatisLogsCleared(cluster.getOzoneManagersList());
+ }
+
+ private void assertRatisLogsCleared(List<OzoneManager> ozoneManagers)
throws Exception {
- // If no transactions have been persisted to the DB, transaction info
- // will be null, not zero.
- // This will cause a null pointer exception if we use
- // OzoneManager#getRatisSnapshotIndex to get the index from the DB.
- OMTransactionInfo txnInfo = om.getMetadataManager()
- .getTransactionInfoTable().get(TRANSACTION_INFO_KEY);
- return (txnInfo.getTransactionIndex() == prepareRequestLogIndex);
+ for (OzoneManager om: ozoneManagers) {
+ LambdaTestUtils.await(WAIT_TIMEOUT_MILLIS, 1000,
+ () -> !logFilesPresentInRatisPeer(om));
+ }
}
}
\ No newline at end of file
diff --git a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
index 1d29d2a..d18354f 100644
--- a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
+++ b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
@@ -336,6 +336,7 @@ enum Status {
UPDATE_LAYOUT_VERSION_FAILED = 69;
LAYOUT_FEATURE_FINALIZATION_FAILED = 70;
PREPARE_FAILED = 71;
+ NOT_SUPPORTED_OPERATION_WHEN_PREPARED = 72;
}
/**
diff --git a/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/ratis/OMTransactionInfo.java b/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/ratis/OMTransactionInfo.java
index 1628883..6c4b1b6 100644
--- a/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/ratis/OMTransactionInfo.java
+++ b/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/ratis/OMTransactionInfo.java
@@ -74,7 +74,7 @@ public final class OMTransactionInfo {
}
public TermIndex getTermIndex() {
- return TermIndex.newTermIndex(term, transactionIndex);
+ return TermIndex.valueOf(term, transactionIndex);
}
/**
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 e5cee22..5f8f8aa 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
@@ -220,6 +220,7 @@ import static org.apache.hadoop.ozone.OzoneConsts.DB_TRANSIENT_MARKER;
import static org.apache.hadoop.ozone.OzoneConsts.OM_METRICS_FILE;
import static org.apache.hadoop.ozone.OzoneConsts.OM_METRICS_TEMP_FILE;
import static org.apache.hadoop.ozone.OzoneConsts.RPC_PORT;
+import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ENABLE_FILESYSTEM_PATHS;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ENABLE_FILESYSTEM_PATHS_DEFAULT;
@@ -337,14 +338,14 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
private final boolean useRatisForReplication;
private boolean isNativeAuthorizerEnabled;
- private boolean isPrepared;
private ExitManager exitManager;
+ private final OzoneManagerPrepareState prepareState;
+
private enum State {
INITIALIZED,
RUNNING,
- PREPARING_FOR_UPGRADE,
STOPPED
}
@@ -352,10 +353,8 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
private State omState;
private Thread emptier;
- // TODO: Utilize the forUpgrade parameter to remove the marker file and
- // take the OM out of prepare mode on startup.
@SuppressWarnings("methodlength")
- private OzoneManager(OzoneConfiguration conf, boolean forUpgrade)
+ private OzoneManager(OzoneConfiguration conf)
throws IOException, AuthenticationException {
super(OzoneVersionInfo.OZONE_VERSION_INFO);
Preconditions.checkNotNull(conf);
@@ -400,6 +399,7 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
ResultCodes.OM_NOT_INITIALIZED);
}
omMetaDir = OMStorage.getOmDbDir(configuration);
+
this.isAclEnabled = conf.getBoolean(OZONE_ACL_ENABLED,
OZONE_ACL_ENABLED_DEFAULT);
this.scmBlockSize = (long) conf.getStorageSize(OZONE_SCM_BLOCK_SIZE,
@@ -464,6 +464,25 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
instantiateServices();
+ // If the prepare marker file is present and its index matches the last
+ // transaction index in the OM DB, turn on the in memory flag to
+ // put the ozone manager in prepare mode, disallowing write requests.
+ // This must be done after metadataManager is instantiated by
+ // instantiateServices and before the RPC server is started.
+ OMTransactionInfo txnInfo = metadataManager.getTransactionInfoTable()
+ .get(TRANSACTION_INFO_KEY);
+ prepareState = new OzoneManagerPrepareState(configuration);
+ if (txnInfo != null) {
+ // Only puts OM in prepare mode if a marker file matching the txn index
+ // is found.
+ prepareState.restorePrepare(txnInfo.getTransactionIndex());
+ } else {
+ // If we have no transaction info in the DB, then no prepare request
+ // could have been received, since the request would update the txn
+ // index in the DB.
+ prepareState.cancelPrepare();
+ }
+
// Create special volume s3v which is required for S3G.
addS3GVolumeToDB();
@@ -504,10 +523,6 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
ShutdownHookManager.get().addShutdownHook(shutdownHook,
SHUTDOWN_HOOK_PRIORITY);
omState = State.INITIALIZED;
-
- // TODO: When marker file is added, check for that on startup to
- // determine prepare mode.
- this.isPrepared = false;
}
private void logVersionMismatch(OzoneConfiguration conf, ScmInfo scmInfo) {
@@ -941,12 +956,7 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
*/
public static OzoneManager createOm(OzoneConfiguration conf)
throws IOException, AuthenticationException {
- return new OzoneManager(conf, false);
- }
-
- public static OzoneManager createOmUpgradeMode(OzoneConfiguration conf)
- throws IOException, AuthenticationException {
- return new OzoneManager(conf, true);
+ return new OzoneManager(conf);
}
/**
@@ -1220,12 +1230,7 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
startJVMPauseMonitor();
setStartTime();
- if (!isPrepared) {
- omState = State.RUNNING;
- } else {
- omState = State.PREPARING_FOR_UPGRADE;
- LOG.info("Started OM services in upgrade mode.");
- }
+ omState = State.RUNNING;
}
/**
@@ -3390,7 +3395,7 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
// TODO: We should only return the snpashotIndex to the leader.
// Should be fixed after RATIS-586
- TermIndex newTermIndex = TermIndex.newTermIndex(term, lastAppliedIndex);
+ TermIndex newTermIndex = TermIndex.valueOf(term, lastAppliedIndex);
return newTermIndex;
}
@@ -3825,4 +3830,8 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
public OmLayoutVersionManager getVersionManager() {
return versionManager;
}
+
+ public OzoneManagerPrepareState getPrepareState() {
+ return prepareState;
+ }
}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManagerPrepareState.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManagerPrepareState.java
new file mode 100644
index 0000000..a79531f
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManagerPrepareState.java
@@ -0,0 +1,282 @@
+/*
+ * 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.hadoop.ozone.om;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.server.ServerUtils;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.om.ratis.OzoneManagerStateMachine;
+import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse.PrepareStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+
+/**
+ * Controls the prepare state of the {@link OzoneManager} containing the
+ * instance. When prepared, an ozone manager should have no Ratis logs
+ * remaining, disallow all write requests except prepare and cancel prepare,
+ * and have a marker file present on disk that will cause it to remain prepared
+ * on restart.
+ */
+public final class OzoneManagerPrepareState {
+ public static final long NO_PREPARE_INDEX = -1;
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(OzoneManagerPrepareState.class);
+
+ private boolean prepareGateEnabled;
+ private long prepareIndex;
+ private PrepareStatus status;
+ private final ConfigurationSource conf;
+
+ public OzoneManagerPrepareState(ConfigurationSource conf) {
+ prepareGateEnabled = false;
+ prepareIndex = NO_PREPARE_INDEX;
+ status = PrepareStatus.PREPARE_NOT_STARTED;
+ this.conf = conf;
+ }
+
+ /**
+ * Turns on the prepare gate flag, clears the prepare index, and moves the
+ * prepare status to {@link PrepareStatus#PREPARE_IN_PROGRESS}.
+ *
+ * Turning on the prepare gate flag will enable a gate in the
+ * {@link OzoneManagerStateMachine#preAppendTransaction} (called on leader
+ * OM only) and {@link OzoneManagerRatisServer#submitRequest}
+ * (called on all OMs) that block write requests from reaching the OM and
+ * fail them with error responses to the client.
+ */
+ public synchronized void enablePrepareGate() {
+ prepareGateEnabled = true;
+ prepareIndex = NO_PREPARE_INDEX;
+ status = PrepareStatus.PREPARE_IN_PROGRESS;
+ }
+
+ /**
+ * Removes the prepare marker file, clears the prepare index, turns off
+ * the prepare gate, and moves the prepare status to
+ * {@link PrepareStatus#PREPARE_NOT_STARTED}.
+ * This can be called from any state to clear the current prepare state.
+ *
+ * @throws IOException If the prepare marker file exists but cannot be
+ * deleted.
+ */
+ public synchronized void cancelPrepare()
+ throws IOException {
+ deletePrepareMarkerFile();
+ prepareIndex = NO_PREPARE_INDEX;
+ prepareGateEnabled = false;
+ status = PrepareStatus.PREPARE_NOT_STARTED;
+ }
+
+ /**
+ * Enables the prepare gate, writes the prepare marker file, sets the in
+ * memory prepare index, and
+ * moves the prepare status to {@link PrepareStatus#PREPARE_COMPLETED}.
+ * This can be called from any state to move the OM into prepare mode.
+ *
+ * @param index The log index to prepare the OM on.
+ * @throws IOException If the marker file cannot be written.
+ */
+ public synchronized void finishPrepare(long index) throws IOException {
+ finishPrepare(index, true);
+ }
+
+ private void finishPrepare(long index, boolean writeFile) throws IOException {
+ // Enabling the prepare gate is idempotent, and may have already been
+ // performed if we are the leader.If we are a follower, we must ensure this
+ // is run now case we become the leader.
+ enablePrepareGate();
+
+ if (writeFile) {
+ writePrepareMarkerFile(index);
+ }
+ prepareIndex = index;
+ status = PrepareStatus.PREPARE_COMPLETED;
+ }
+
+ /**
+ * Uses the on disk marker file to determine the OM's prepare state.
+ * If the marker file exists and contains an index matching {@code
+ * expectedPrepareIndex}, the necessary steps will be taken to finish
+ * preparation and the state will be moved to
+ * {@link PrepareStatus#PREPARE_COMPLETED}.
+ * Else, the status will be moved to
+ * {@link PrepareStatus#PREPARE_NOT_STARTED} and any preparation steps will
+ * be cancelled.
+ *
+ * @return The status the OM is in after this method call.
+ * @throws IOException If the marker file cannot be read, and it cannot be
+ * deleted as part of moving to the
+ * {@link PrepareStatus#PREPARE_NOT_STARTED} state.
+ */
+ public synchronized PrepareStatus restorePrepare(long expectedPrepareIndex)
+ throws IOException {
+ boolean prepareIndexRead = true;
+ long prepareMarkerIndex = NO_PREPARE_INDEX;
+
+ File prepareMarkerFile = getPrepareMarkerFile();
+ if (prepareMarkerFile.exists()) {
+ byte[] data = new byte[(int) prepareMarkerFile.length()];
+ try(FileInputStream stream = new FileInputStream(prepareMarkerFile)) {
+ stream.read(data);
+ } catch (IOException e) {
+ LOG.error("Failed to read prepare marker file {} while restoring OM.",
+ prepareMarkerFile.getAbsolutePath());
+ prepareIndexRead = false;
+ }
+
+ try {
+ prepareMarkerIndex = Long.parseLong(
+ new String(data, StandardCharsets.UTF_8));
+ } catch (NumberFormatException e) {
+ LOG.error("Failed to parse log index from prepare marker file {} " +
+ "while restoring OM.", prepareMarkerFile.getAbsolutePath());
+ prepareIndexRead = false;
+ }
+ } else {
+ // No marker file found.
+ prepareIndexRead = false;
+ }
+
+ boolean prepareRestored = false;
+ if (prepareIndexRead) {
+ if (prepareMarkerIndex != expectedPrepareIndex) {
+ LOG.error("Failed to restore OM prepare state, because the expected " +
+ "prepare index {} does not match the index {} written to the " +
+ "marker file.", expectedPrepareIndex, prepareMarkerIndex);
+ } else {
+ // Prepare state can only be restored if we read the expected index
+ // from the marker file.
+ prepareRestored = true;
+ }
+ }
+
+ if (prepareRestored) {
+ // Do not rewrite the marker file, since we verified it already exists.
+ finishPrepare(prepareMarkerIndex, false);
+ } else {
+ // If the potentially faulty marker file cannot be deleted,
+ // propagate the IOException.
+ // If there is no marker file, this call sets the in memory state only.
+ cancelPrepare();
+ }
+
+ return status;
+ }
+
+ /**
+ * If the prepare gate is enabled, always returns true.
+ * If the prepare gate is disabled, returns true only if {@code
+ * requestType} is {@code Prepare} or {@code CancelPrepare}. Returns false
+ * otherwise.
+ */
+ public synchronized boolean requestAllowed(Type requestType) {
+ boolean requestAllowed = true;
+
+ if (prepareGateEnabled) {
+ // TODO: Also return true for cancel prepare when it is implemented.
+ requestAllowed = (requestType == Type.Prepare);
+ }
+
+ return requestAllowed;
+ }
+
+ /**
+ * @return the current log index and status of preparation.
+ * Both fields are returned together to provide a consistent view of the
+ * state, which would not be guaranteed if they had to be retrieved through
+ * separate getters.
+ */
+ public synchronized State getState() {
+ return new State(prepareIndex, status);
+ }
+
+ /**
+ * Creates a prepare marker file inside the OM metadata directory which
+ * contains the log index {@code index}.
+ * If a marker file already exists, it will be overwritten.
+ */
+ private void writePrepareMarkerFile(long index) throws IOException {
+ File markerFile = getPrepareMarkerFile();
+ File parentDir = markerFile.getParentFile();
+ Files.createDirectories(parentDir.toPath());
+
+ try(FileOutputStream stream = new FileOutputStream(markerFile)) {
+ stream.write(Long.toString(index).getBytes(StandardCharsets.UTF_8));
+ }
+
+ LOG.info("Prepare marker file written with log index {} to file {}", index,
+ markerFile.getAbsolutePath());
+ }
+
+ private void deletePrepareMarkerFile()
+ throws IOException {
+ File markerFile = getPrepareMarkerFile();
+ if (markerFile.exists()) {
+ Files.delete(markerFile.toPath());
+ LOG.info("Deleted prepare marker file: {}", markerFile.getAbsolutePath());
+ } else {
+ LOG.info("Request to delete prepare marker file that does not exist: {}",
+ markerFile.getAbsolutePath());
+ }
+ }
+
+ /**
+ * Returns a {@link File} object representing the prepare marker file,
+ * which may or may not actually exist on disk.
+ * This method should be used for testing only.
+ */
+ @VisibleForTesting
+ public File getPrepareMarkerFile() {
+ File markerFileDir = new File(ServerUtils.getOzoneMetaDirPath(conf),
+ OMStorage.STORAGE_DIR_CURRENT);
+ return new File(markerFileDir, OzoneConsts.PREPARE_MARKER);
+ }
+
+ /**
+ * The current state of preparation is defined by the status and the
+ * prepare index that are currently set.
+ */
+ public static class State {
+ private final long index;
+ private final PrepareStatus status;
+
+ public State(long index, PrepareStatus status) {
+ this.index = index;
+ this.status = status;
+ }
+
+ public long getIndex() {
+ return index;
+ }
+
+ public PrepareStatus getStatus() {
+ return status;
+ }
+ }
+}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OMRatisSnapshotInfo.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OMRatisSnapshotInfo.java
index b893414..91c388e 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OMRatisSnapshotInfo.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OMRatisSnapshotInfo.java
@@ -52,7 +52,7 @@ public class OMRatisSnapshotInfo implements SnapshotInfo {
@Override
public TermIndex getTermIndex() {
- return TermIndex.newTermIndex(term, snapshotIndex);
+ return TermIndex.valueOf(term, snapshotIndex);
}
@Override
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerRatisServer.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerRatisServer.java
index 467764b..f552f48 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerRatisServer.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerRatisServer.java
@@ -50,6 +50,7 @@ import org.apache.hadoop.ozone.om.helpers.OMRatisHelper;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
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.Status;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
@@ -128,10 +129,29 @@ public final class OzoneManagerRatisServer {
* @throws ServiceException
*/
public OMResponse submitRequest(OMRequest omRequest) throws ServiceException {
- RaftClientRequest raftClientRequest =
- createWriteRaftClientRequest(omRequest);
- RaftClientReply raftClientReply = submitRequestToRatis(raftClientRequest);
- return processReply(omRequest, raftClientReply);
+ // In prepare mode, only prepare and cancel requests are allowed to go
+ // through.
+ if (ozoneManager.getPrepareState().requestAllowed(omRequest.getCmdType())) {
+ RaftClientRequest raftClientRequest =
+ createWriteRaftClientRequest(omRequest);
+ RaftClientReply raftClientReply = submitRequestToRatis(raftClientRequest);
+
+ return processReply(omRequest, raftClientReply);
+ } else {
+ LOG.info("Rejecting write request on OM {} because it is in prepare " +
+ "mode: {}", ozoneManager.getOMNodeId(),
+ omRequest.getCmdType().name());
+
+ String message = "Cannot apply write request " +
+ omRequest.getCmdType().name() + " when OM is in prepare mode.";
+ OMResponse.Builder omResponse = OMResponse.newBuilder()
+ .setMessage(message)
+ .setStatus(Status.NOT_SUPPORTED_OPERATION_WHEN_PREPARED)
+ .setCmdType(omRequest.getCmdType())
+ .setTraceID(omRequest.getTraceID())
+ .setSuccess(false);
+ return omResponse.build();
+ }
}
/**
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
index aaf94e9..82e2fa3 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
@@ -33,8 +33,10 @@ import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.OzoneManagerPrepareState;
import org.apache.hadoop.ozone.om.exceptions.OMException;
import org.apache.hadoop.ozone.om.helpers.OMRatisHelper;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
.OMRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
@@ -48,6 +50,7 @@ import org.apache.ratis.protocol.Message;
import org.apache.ratis.protocol.RaftClientRequest;
import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.protocol.RaftPeerId;
+import org.apache.ratis.protocol.exceptions.StateMachineException;
import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.protocol.TermIndex;
import org.apache.ratis.server.storage.RaftStorage;
@@ -193,6 +196,32 @@ public class OzoneManagerStateMachine extends BaseStateMachine {
return handleStartTransactionRequests(raftClientRequest, omRequest);
}
+ @Override
+ public TransactionContext preAppendTransaction(TransactionContext trx)
+ throws IOException {
+ OMRequest request = OMRatisHelper.convertByteStringToOMRequest(
+ trx.getStateMachineLogEntry().getLogData());
+ OzoneManagerProtocolProtos.Type cmdType = request.getCmdType();
+
+ // In prepare mode, only prepare and cancel requests are allowed to go
+ // through.
+ OzoneManagerPrepareState prepareState = ozoneManager.getPrepareState();
+ if (prepareState.requestAllowed(cmdType)) {
+ if (cmdType == OzoneManagerProtocolProtos.Type.Prepare) {
+ prepareState.enablePrepareGate();
+ }
+ // TODO: Add cancel prepare here after it is implemented.
+ return trx;
+ } else {
+ String message = "Cannot apply write request " +
+ request.getCmdType().name() + " when OM is in prepare mode.";
+ OMException cause = new OMException(message,
+ OMException.ResultCodes.NOT_SUPPORTED_OPERATION_WHEN_PREPARED);
+ // Indicate that the leader should not step down because of this failure.
+ throw new StateMachineException(message, cause, false);
+ }
+ }
+
/*
* Apply a committed log entry to the state machine.
*/
@@ -317,7 +346,7 @@ public class OzoneManagerStateMachine extends BaseStateMachine {
getLifeCycle().startAndTransition(() -> {
this.ozoneManagerDoubleBuffer = buildDoubleBufferForRatis();
handler.updateDoubleBuffer(ozoneManagerDoubleBuffer);
- this.setLastAppliedTermIndex(TermIndex.newTermIndex(
+ this.setLastAppliedTermIndex(TermIndex.valueOf(
newLastAppliedSnapShotTermIndex, newLastAppliedSnaphsotIndex));
});
}
@@ -510,7 +539,7 @@ public class OzoneManagerStateMachine extends BaseStateMachine {
OMTransactionInfo.readTransactionInfo(
ozoneManager.getMetadataManager());
if (omTransactionInfo != null) {
- setLastAppliedTermIndex(TermIndex.newTermIndex(
+ setLastAppliedTermIndex(TermIndex.valueOf(
omTransactionInfo.getTerm(),
omTransactionInfo.getTransactionIndex()));
snapshotInfo.updateTermIndex(omTransactionInfo.getTerm(),
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMPrepareRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMPrepareRequest.java
index 87b7657..a4a26c4 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMPrepareRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMPrepareRequest.java
@@ -35,8 +35,7 @@ import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRespo
import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
-import org.apache.ratis.server.impl.RaftServerImpl;
-import org.apache.ratis.server.impl.RaftServerProxy;
+import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.raftlog.RaftLog;
import org.apache.ratis.statemachine.StateMachine;
import org.slf4j.Logger;
@@ -67,7 +66,8 @@ public class OMPrepareRequest extends OMClientRequest {
OzoneManager ozoneManager, long transactionLogIndex,
OzoneManagerDoubleBufferHelper ozoneManagerDoubleBufferHelper) {
- LOG.info("Received prepare request with log index {}", transactionLogIndex);
+ LOG.info("OM {} Received prepare request with log index {}",
+ ozoneManager.getOMNodeId(), transactionLogIndex);
OMRequest omRequest = getOmRequest();
OzoneManagerProtocolProtos.PrepareRequestArgs args =
@@ -99,9 +99,9 @@ public class OMPrepareRequest extends OMClientRequest {
ozoneManagerDoubleBufferHelper.add(response, transactionLogIndex);
OzoneManagerRatisServer omRatisServer = ozoneManager.getOmRatisServer();
- RaftServerProxy server = (RaftServerProxy) omRatisServer.getServer();
- RaftServerImpl serverImpl =
- server.getImpl(omRatisServer.getRaftGroup().getGroupId());
+ RaftServer.Division division =
+ omRatisServer.getServer()
+ .getDivision(omRatisServer.getRaftGroup().getGroupId());
// Wait for outstanding double buffer entries to flush to disk,
// so they will not be purged from the log before being persisted to
@@ -110,11 +110,18 @@ public class OMPrepareRequest extends OMClientRequest {
// already, once this index reaches the state machine, we know all
// transactions have been flushed.
waitForLogIndex(transactionLogIndex,
- ozoneManager.getMetadataManager(), serverImpl,
+ ozoneManager.getMetadataManager(), division,
flushTimeout, flushCheckInterval);
- takeSnapshotAndPurgeLogs(serverImpl);
- // TODO: Create marker file with txn index.
+ // TODO: After the snapshot index update fix, the prepare index will
+ // always be 1 more than the prepare txn index, because the Ratis
+ // entry to commit the prepare request will be included in the snapshot.
+ long prepareIndex = takeSnapshotAndPurgeLogs(division);
+
+ // Save transaction log index to a marker file, so if the OM restarts,
+ // it will remain in prepare mode on that index as long as the file
+ // exists.
+ ozoneManager.getPrepareState().finishPrepare(prepareIndex);
LOG.info("OM {} prepared at log index {}. Returning response {}",
ozoneManager.getOMNodeId(),
@@ -142,7 +149,7 @@ public class OMPrepareRequest extends OMClientRequest {
* disk, and to be updated in memory in Ratis.
*/
private static void waitForLogIndex(long indexToWaitFor,
- OMMetadataManager metadataManager, RaftServerImpl server,
+ OMMetadataManager metadataManager, RaftServer.Division division,
Duration flushTimeout, Duration flushCheckInterval)
throws InterruptedException, IOException {
@@ -158,7 +165,7 @@ public class OMPrepareRequest extends OMClientRequest {
OMTransactionInfo dbTxnInfo = metadataManager
.getTransactionInfoTable().get(TRANSACTION_INFO_KEY);
long ratisTxnIndex =
- server.getStateMachine().getLastAppliedTermIndex().getIndex();
+ division.getStateMachine().getLastAppliedTermIndex().getIndex();
// Ratis may apply meta transactions after the prepare request, causing
// its in memory index to always be greater than the DB index.
@@ -189,15 +196,16 @@ public class OMPrepareRequest extends OMClientRequest {
/**
* Take a snapshot of the state machine at the last index, and purge ALL logs.
- * @param impl RaftServerImpl instance
+ * @param division Raft server division.
+ * @return The index the snapshot was taken on.
* @throws IOException on Error.
*/
- public static long takeSnapshotAndPurgeLogs(RaftServerImpl impl)
+ public static long takeSnapshotAndPurgeLogs(RaftServer.Division division)
throws IOException {
- StateMachine stateMachine = impl.getStateMachine();
+ StateMachine stateMachine = division.getStateMachine();
long snapshotIndex = stateMachine.takeSnapshot();
- RaftLog raftLog = impl.getState().getLog();
+ RaftLog raftLog = division.getRaftLog();
long raftLogIndex = raftLog.getLastEntryTermIndex().getIndex();
// We can have a case where the log has a meta transaction after the
@@ -213,6 +221,7 @@ public class OMPrepareRequest extends OMClientRequest {
CompletableFuture<Long> purgeFuture =
raftLog.syncWithSnapshot(snapshotIndex);
+
try {
Long purgeIndex = purgeFuture.get();
if (purgeIndex != snapshotIndex) {
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
index e0e7abb..eb5c1c7 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList;
import org.apache.hadoop.hdds.utils.db.SequenceNumberNotFoundException;
import org.apache.hadoop.ozone.OzoneAcl;
import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.OzoneManagerPrepareState;
import org.apache.hadoop.ozone.om.exceptions.OMException;
import org.apache.hadoop.ozone.om.helpers.DBUpdates;
import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
@@ -74,9 +75,7 @@ import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Multipa
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.MultipartUploadListPartsResponse;
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.PrepareStatusRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse.PrepareStatus;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ServiceListRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ServiceListResponse;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status;
@@ -98,8 +97,6 @@ import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.
import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.MultipartUploadInfo;
import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneAclInfo;
import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PartInfo;
-import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse.PrepareStatus.PREPARE_COMPLETED;
-import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse.PrepareStatus.PREPARE_IN_PROGRESS;
import org.apache.hadoop.ozone.upgrade.UpgradeFinalizer.StatusAndMessages;
import org.slf4j.Logger;
@@ -220,8 +217,7 @@ public class OzoneManagerRequestHandler implements RequestHandler {
.setFinalizeUpgradeProgressResponse(upgradeProgressResponse);
break;
case PrepareStatus:
- PrepareStatusResponse prepareStatusResponse =
- getPrepareStatus(request.getPrepareStatusRequest());
+ PrepareStatusResponse prepareStatusResponse = getPrepareStatus();
responseBuilder.setPrepareStatusResponse(prepareStatusResponse);
break;
default:
@@ -633,23 +629,12 @@ public class OzoneManagerRequestHandler implements RequestHandler {
.build();
}
- private PrepareStatusResponse getPrepareStatus(PrepareStatusRequest request)
- throws IOException {
- // TODO After HDDS-4569,
- // When there is a global "prepared" state in OM, we can return
- // PREPARE_NOT_STARTED instead of PREPARE_IN_PROGRESS appropriately.
- PrepareStatus prepareStatus = null;
- long txnID = request.getTxnID();
- long ratisSnapshotIndex = impl.getRatisSnapshotIndex();
- if (ratisSnapshotIndex != txnID) {
- LOG.info("Last Txn Id = {}, PrepareStatusRequest Txn Id = {}",
- ratisSnapshotIndex, request.getTxnID());
- prepareStatus = PREPARE_IN_PROGRESS;
- } else {
- prepareStatus = PREPARE_COMPLETED;
- }
- return PrepareStatusResponse.newBuilder().setStatus(prepareStatus)
- .setCurrentTxnIndex(ratisSnapshotIndex).build();
+ private PrepareStatusResponse getPrepareStatus() {
+ OzoneManagerPrepareState.State prepareState =
+ impl.getPrepareState().getState();
+ return PrepareStatusResponse.newBuilder()
+ .setStatus(prepareState.getStatus())
+ .setCurrentTxnIndex(prepareState.getIndex()).build();
}
protected OzoneManager getOzoneManager() {
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerRatisServer.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerRatisServer.java
index f22f3db..535bbce 100644
--- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerRatisServer.java
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerRatisServer.java
@@ -103,7 +103,7 @@ public class TestOzoneManagerRatisServer {
folder.newFolder().getAbsolutePath());
omMetadataManager = new OmMetadataManagerImpl(ozoneConfiguration);
when(ozoneManager.getMetadataManager()).thenReturn(omMetadataManager);
- initialTermIndex = TermIndex.newTermIndex(0, 0);
+ initialTermIndex = TermIndex.valueOf(0, 0);
OMRatisSnapshotInfo omRatisSnapshotInfo = new OMRatisSnapshotInfo();
when(ozoneManager.getSnapshotInfo()).thenReturn(omRatisSnapshotInfo);
omRatisServer = OzoneManagerRatisServer.newOMRatisServer(conf, ozoneManager,
@@ -136,7 +136,7 @@ public class TestOzoneManagerRatisServer {
SnapshotInfo snapshotInfo =
omRatisServer.getOmStateMachine().getLatestSnapshot();
- TermIndex newSnapshotIndex = TermIndex.newTermIndex(
+ TermIndex newSnapshotIndex = TermIndex.valueOf(
snapshotInfo.getTerm(), snapshotInfo.getIndex() + 100);
omMetadataManager.getTransactionInfoTable().put(TRANSACTION_INFO_KEY,
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
index a0a21b9..48a5846 100644
--- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
@@ -21,6 +21,19 @@ import org.apache.hadoop.ozone.om.OMConfigKeys;
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.OzoneManagerPrepareState;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
+import org.apache.hadoop.ozone.om.helpers.OMRatisHelper;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareRequestArgs;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.CreateKeyRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyArgs;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse.PrepareStatus;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareRequest;
+import org.apache.ratis.proto.RaftProtos;
+import org.apache.ratis.protocol.exceptions.StateMachineException;
+import org.apache.ratis.statemachine.TransactionContext;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
@@ -42,6 +55,8 @@ public class TestOzoneManagerStateMachine {
public TemporaryFolder tempDir = new TemporaryFolder();
private OzoneManagerStateMachine ozoneManagerStateMachine;
+ private OzoneManagerPrepareState prepareState;
+
@Before
public void setup() throws Exception {
OzoneManagerRatisServer ozoneManagerRatisServer =
@@ -56,6 +71,9 @@ public class TestOzoneManagerStateMachine {
when(ozoneManager.getMetadataManager()).thenReturn(omMetadataManager);
+ prepareState = new OzoneManagerPrepareState(conf);
+ when(ozoneManager.getPrepareState()).thenReturn(prepareState);
+
when(ozoneManagerRatisServer.getOzoneManager()).thenReturn(ozoneManager);
when(ozoneManager.getSnapshotInfo()).thenReturn(
Mockito.mock(OMRatisSnapshotInfo.class));
@@ -230,4 +248,80 @@ public class TestOzoneManagerStateMachine {
Assert.assertEquals(10L,
ozoneManagerStateMachine.getLastAppliedTermIndex().getIndex());
}
+
+ @Test
+ public void testPreAppendTransaction() throws Exception {
+ // Submit write request.
+ KeyArgs args = KeyArgs
+ .newBuilder()
+ .setVolumeName("volume")
+ .setBucketName("bucket")
+ .setKeyName("key")
+ .build();
+ OMRequest createKeyRequest = OMRequest.newBuilder()
+ .setCreateKeyRequest(CreateKeyRequest.newBuilder().setKeyArgs(args))
+ .setCmdType(Type.CreateKey)
+ .setClientId("123")
+ .build();
+
+ TransactionContext submittedTrx = mockTransactionContext(createKeyRequest);
+ TransactionContext returnedTrx =
+ ozoneManagerStateMachine.preAppendTransaction(submittedTrx);
+
+ // No prepare should be triggered, and the txn should be returned unaltered.
+ Assert.assertEquals(prepareState.getState().getStatus(),
+ PrepareStatus.PREPARE_NOT_STARTED);
+ Assert.assertSame(submittedTrx, returnedTrx);
+
+ // Submit prepare request.
+ OMRequest prepareRequest = OMRequest.newBuilder()
+ .setPrepareRequest(
+ PrepareRequest.newBuilder()
+ .setArgs(PrepareRequestArgs.getDefaultInstance()))
+ .setCmdType(Type.Prepare)
+ .setClientId("123")
+ .build();
+
+ submittedTrx = mockTransactionContext(prepareRequest);
+ returnedTrx = ozoneManagerStateMachine.preAppendTransaction(submittedTrx);
+
+ // Prepare should be started, and txn should be returned unaltered.
+ Assert.assertEquals(prepareState.getState().getStatus(),
+ PrepareStatus.PREPARE_IN_PROGRESS);
+ Assert.assertSame(submittedTrx, returnedTrx);
+
+ // Submitting a write request should now fail.
+ try {
+ ozoneManagerStateMachine.preAppendTransaction(
+ mockTransactionContext(createKeyRequest));
+ Assert.fail("Expected StateMachineException to be thrown when " +
+ "submitting write request while prepared.");
+ } catch(StateMachineException smEx) {
+ Assert.assertFalse(smEx.leaderShouldStepDown());
+
+ Throwable cause = smEx.getCause();
+ Assert.assertTrue(cause instanceof OMException);
+ Assert.assertEquals(((OMException) cause).getResult(),
+ OMException.ResultCodes.NOT_SUPPORTED_OPERATION_WHEN_PREPARED);
+ }
+
+ // Should be able to prepare again without issue.
+ Assert.assertEquals(prepareState.getState().getStatus(),
+ PrepareStatus.PREPARE_IN_PROGRESS);
+ Assert.assertSame(submittedTrx, returnedTrx);
+
+ // TODO: Add test for cancel prepare once it is implemented.
+ }
+
+ private TransactionContext mockTransactionContext(OMRequest request) {
+ RaftProtos.StateMachineLogEntryProto logEntry =
+ RaftProtos.StateMachineLogEntryProto.newBuilder()
+ .setLogData(OMRatisHelper.convertRequestToByteString(request))
+ .build();
+
+ TransactionContext mockTrx = Mockito.mock(TransactionContext.class);
+ when(mockTrx.getStateMachineLogEntry()).thenReturn(logEntry);
+
+ return mockTrx;
+ }
}
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/upgrade/TestOzoneManagerPrepareState.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/upgrade/TestOzoneManagerPrepareState.java
new file mode 100644
index 0000000..7dffe33
--- /dev/null
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/upgrade/TestOzoneManagerPrepareState.java
@@ -0,0 +1,269 @@
+/*
+ * 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.hadoop.ozone.om.upgrade;
+
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse.PrepareStatus;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.om.OzoneManagerPrepareState;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.Random;
+
+public class TestOzoneManagerPrepareState {
+ @Rule
+ public TemporaryFolder folder = new TemporaryFolder();
+
+ private static final int TEST_INDEX = 5;
+ private OzoneManagerPrepareState prepareState;
+
+ @Before
+ public void setup() throws Exception {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.set(HddsConfigKeys.OZONE_METADATA_DIRS,
+ folder.getRoot().getAbsolutePath());
+
+ prepareState = new OzoneManagerPrepareState(conf);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ // Clean up marker file from previous runs.
+ prepareState.cancelPrepare();
+ }
+
+ @Test
+ public void testStartPrepare() {
+ assertPrepareNotStarted();
+ prepareState.enablePrepareGate();
+ assertPrepareInProgress();
+ }
+
+ @Test
+ public void testFinishPrepare() throws Exception {
+ // Test finish as a follower OM, where it would be called without setting
+ // the prepare gate first.
+ prepareState.finishPrepare(TEST_INDEX);
+ assertPrepareCompleted(TEST_INDEX);
+ prepareState.cancelPrepare();
+
+ // Test finish as a leader OM, where the prepare gate would be put up
+ // before prepare is finished.
+ prepareState.enablePrepareGate();
+ prepareState.finishPrepare(TEST_INDEX);
+ assertPrepareCompleted(TEST_INDEX);
+
+ // A following prepare operation should not raise errors.
+ prepareState.finishPrepare(TEST_INDEX);
+ assertPrepareCompleted(TEST_INDEX);
+ }
+
+ @Test
+ public void testCancelPrepare() throws Exception {
+ // Test cancel after start.
+ prepareState.enablePrepareGate();
+ prepareState.cancelPrepare();
+ assertPrepareNotStarted();
+
+ // Test cancel after finish.
+ prepareState.enablePrepareGate();
+ prepareState.finishPrepare(TEST_INDEX);
+ prepareState.cancelPrepare();
+ assertPrepareNotStarted();
+
+ // Test cancel after cancel.
+ prepareState.cancelPrepare();
+ prepareState.cancelPrepare();
+ assertPrepareNotStarted();
+
+ }
+
+ @Test
+ public void testRequestAllowed() {
+ assertPrepareGateDown();
+ prepareState.enablePrepareGate();
+ assertPrepareGateUp();
+ }
+
+ @Test
+ public void testRestoreCorrectIndex() throws Exception {
+ writePrepareMarkerFile(TEST_INDEX);
+ PrepareStatus status = prepareState.restorePrepare(TEST_INDEX);
+ Assert.assertEquals(PrepareStatus.PREPARE_COMPLETED, status);
+
+ assertPrepareCompleted(TEST_INDEX);
+ }
+
+ @Test
+ public void testRestoreIncorrectIndex() throws Exception {
+ writePrepareMarkerFile(TEST_INDEX);
+ PrepareStatus status = prepareState.restorePrepare(TEST_INDEX + 1);
+ Assert.assertEquals(PrepareStatus.PREPARE_NOT_STARTED, status);
+
+ assertPrepareNotStarted();
+ }
+
+ @Test
+ public void testRestoreGarbageMarkerFile() throws Exception {
+ byte[] randomBytes = new byte[10];
+ new Random().nextBytes(randomBytes);
+ writePrepareMarkerFile(randomBytes);
+
+ PrepareStatus status = prepareState.restorePrepare(TEST_INDEX);
+ Assert.assertEquals(PrepareStatus.PREPARE_NOT_STARTED, status);
+
+ assertPrepareNotStarted();
+ }
+
+ @Test
+ public void testRestoreEmptyMarkerFile() throws Exception {
+ writePrepareMarkerFile(new byte[]{});
+
+ PrepareStatus status = prepareState.restorePrepare(TEST_INDEX);
+ Assert.assertEquals(PrepareStatus.PREPARE_NOT_STARTED, status);
+
+ assertPrepareNotStarted();
+ }
+
+ @Test
+ public void testRestoreNoMarkerFile() throws Exception {
+ PrepareStatus status = prepareState.restorePrepare(TEST_INDEX);
+ Assert.assertEquals(PrepareStatus.PREPARE_NOT_STARTED, status);
+
+ assertPrepareNotStarted();
+ }
+
+ @Test
+ public void testRestoreAfterStart() throws Exception {
+ prepareState.enablePrepareGate();
+
+ // If prepare is started but never finished, no marker file is written to
+ // disk, and restoring the prepare state on an OM restart should leave it
+ // not prepared.
+ PrepareStatus status = prepareState.restorePrepare(TEST_INDEX);
+ Assert.assertEquals(PrepareStatus.PREPARE_NOT_STARTED, status);
+ assertPrepareNotStarted();
+ }
+
+ @Test
+ public void testMultipleRestores() throws Exception {
+ PrepareStatus status = prepareState.restorePrepare(TEST_INDEX);
+ Assert.assertEquals(PrepareStatus.PREPARE_NOT_STARTED, status);
+ assertPrepareNotStarted();
+
+ status = prepareState.restorePrepare(TEST_INDEX);
+ Assert.assertEquals(PrepareStatus.PREPARE_NOT_STARTED, status);
+ assertPrepareNotStarted();
+
+ prepareState.enablePrepareGate();
+ prepareState.finishPrepare(TEST_INDEX);
+
+ status = prepareState.restorePrepare(TEST_INDEX);
+ Assert.assertEquals(PrepareStatus.PREPARE_COMPLETED, status);
+ assertPrepareCompleted(TEST_INDEX);
+
+ status = prepareState.restorePrepare(TEST_INDEX);
+ Assert.assertEquals(PrepareStatus.PREPARE_COMPLETED, status);
+ assertPrepareCompleted(TEST_INDEX);
+ }
+
+ private void writePrepareMarkerFile(long index) throws IOException {
+ writePrepareMarkerFile(Long.toString(index).getBytes());
+ }
+
+ private void writePrepareMarkerFile(byte[] bytes) throws IOException {
+ File markerFile = prepareState.getPrepareMarkerFile();
+ markerFile.getParentFile().mkdirs();
+ try(FileOutputStream stream =
+ new FileOutputStream(markerFile)) {
+ stream.write(bytes);
+ }
+ }
+
+ private long readPrepareMarkerFile() throws Exception {
+ long index = 0;
+ File prepareMarkerFile = prepareState.getPrepareMarkerFile();
+ byte[] data = new byte[(int) prepareMarkerFile.length()];
+
+ try (FileInputStream stream = new FileInputStream(prepareMarkerFile)) {
+ stream.read(data);
+ index = Long.parseLong(new String(data));
+ }
+
+ return index;
+ }
+
+ private void assertPrepareNotStarted() {
+ OzoneManagerPrepareState.State state = prepareState.getState();
+ Assert.assertEquals(PrepareStatus.PREPARE_NOT_STARTED, state.getStatus());
+ Assert.assertEquals(OzoneManagerPrepareState.NO_PREPARE_INDEX,
+ state.getIndex());
+ Assert.assertFalse(prepareState.getPrepareMarkerFile().exists());
+
+ assertPrepareGateDown();
+ }
+
+ private void assertPrepareInProgress() {
+ OzoneManagerPrepareState.State state = prepareState.getState();
+ Assert.assertEquals(PrepareStatus.PREPARE_IN_PROGRESS, state.getStatus());
+ Assert.assertEquals(OzoneManagerPrepareState.NO_PREPARE_INDEX,
+ state.getIndex());
+ Assert.assertFalse(prepareState.getPrepareMarkerFile().exists());
+
+ assertPrepareGateUp();
+ }
+
+ private void assertPrepareCompleted(long index) throws Exception {
+ OzoneManagerPrepareState.State state = prepareState.getState();
+ Assert.assertEquals(PrepareStatus.PREPARE_COMPLETED,
+ state.getStatus());
+ Assert.assertEquals(index, state.getIndex());
+ Assert.assertEquals(index, readPrepareMarkerFile());
+
+ assertPrepareGateUp();
+ }
+
+ private void assertPrepareGateUp() {
+ // Once preparation has begun, only prepare and cancel prepare should be
+ // allowed.
+ for (Type cmdType: Type.values()) {
+ if (cmdType == Type.Prepare) {
+ // TODO: Add cancelPrepare to allowed request types when it is added.
+ Assert.assertTrue(prepareState.requestAllowed(cmdType));
+ } else {
+ Assert.assertFalse(prepareState.requestAllowed(cmdType));
+ }
+ }
+ }
+
+ private void assertPrepareGateDown() {
+ for (Type cmdType: Type.values()) {
+ Assert.assertTrue(prepareState.requestAllowed(cmdType));
+ }
+ }
+}
diff --git a/pom.xml b/pom.xml
index 05c34d5..16b3ac6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -79,7 +79,7 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xs
<declared.ozone.version>${ozone.version}</declared.ozone.version>
<!-- Apache Ratis version -->
- <ratis.version>1.1.0-913f5a4-SNAPSHOT</ratis.version>
+ <ratis.version>1.1.0-c5eafb9-SNAPSHOT</ratis.version>
<!-- Apache Ratis thirdparty version -->
<ratis.thirdparty.version>0.6.0-SNAPSHOT</ratis.thirdparty.version>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org