You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by ck...@apache.org on 2022/10/26 02:40:02 UTC
[ozone] branch ozone-1.3 updated (ffdbf145f0 -> 3f6d238a87)
This is an automated email from the ASF dual-hosted git repository.
ckj pushed a change to branch ozone-1.3
in repository https://gitbox.apache.org/repos/asf/ozone.git
from ffdbf145f0 HDDS-7284. JVM crash for rocksdb for read/write after close (#3801)
new 1add651b8a HDDS-7253. Fix exception when '/' in key name (#3774)
new b2ff74f7f5 HDDS-7258. Cleanup the allocated but uncommitted blocks (#3778)
new 3f6d238a87 HDDS-7396. Force close non-RATIS containers in ReplicationManager (#3877)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../health/ClosingContainerHandler.java | 5 +-
.../health/TestClosingContainerHandler.java | 103 +++++++++-------
.../apache/hadoop/ozone/om/helpers/OmKeyInfo.java | 76 ++++++++----
.../hadoop/fs/ozone/TestOzoneFileSystem.java | 81 +++++++++++-
.../apache/hadoop/ozone/om/TestKeyManagerImpl.java | 70 ++++++++++-
.../org/apache/hadoop/ozone/om/KeyManagerImpl.java | 68 +++++++++--
.../ozone/om/request/key/OMKeyCommitRequest.java | 19 ++-
.../om/request/key/OMKeyCommitRequestWithFSO.java | 23 +++-
.../hadoop/ozone/om/request/key/OMKeyRequest.java | 24 ++++
.../ozone/om/response/key/OMKeyCommitResponse.java | 6 +
.../om/request/key/TestOMKeyCommitRequest.java | 96 ++++++++++++++-
.../ozone/om/service/TestKeyDeletingService.java | 136 +++++++++++++++++++--
12 files changed, 595 insertions(+), 112 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org
[ozone] 01/03: HDDS-7253. Fix exception when '/' in key name (#3774)
Posted by ck...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ckj pushed a commit to branch ozone-1.3
in repository https://gitbox.apache.org/repos/asf/ozone.git
commit 1add651b8ae3b6085fa6b2695164b46247c82fc4
Author: XiChen <32...@users.noreply.github.com>
AuthorDate: Tue Oct 25 20:09:50 2022 +0800
HDDS-7253. Fix exception when '/' in key name (#3774)
---
.../hadoop/fs/ozone/TestOzoneFileSystem.java | 81 ++++++++++++++++++++--
.../apache/hadoop/ozone/om/TestKeyManagerImpl.java | 70 +++++++++++++++++--
.../org/apache/hadoop/ozone/om/KeyManagerImpl.java | 68 ++++++++++++++----
3 files changed, 199 insertions(+), 20 deletions(-)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystem.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystem.java
index f3e8cf10be..1f688029b6 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystem.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystem.java
@@ -153,6 +153,7 @@ public class TestOzoneFileSystem {
private static OzoneManagerProtocol writeClient;
private static FileSystem fs;
private static OzoneFileSystem o3fs;
+ private static OzoneBucket ozoneBucket;
private static String volumeName;
private static String bucketName;
private static Trash trash;
@@ -179,10 +180,9 @@ public class TestOzoneFileSystem {
writeClient = cluster.getRpcClient().getObjectStore()
.getClientProxy().getOzoneManagerClient();
// create a volume and a bucket to be used by OzoneFileSystem
- OzoneBucket bucket =
- TestDataUtil.createVolumeAndBucket(cluster, bucketLayout);
- volumeName = bucket.getVolumeName();
- bucketName = bucket.getName();
+ ozoneBucket = TestDataUtil.createVolumeAndBucket(cluster, bucketLayout);
+ volumeName = ozoneBucket.getVolumeName();
+ bucketName = ozoneBucket.getName();
String rootPath = String.format("%s://%s.%s/",
OzoneConsts.OZONE_URI_SCHEME, bucketName, volumeName);
@@ -334,6 +334,30 @@ public class TestOzoneFileSystem {
assertTrue("Shouldn't send error if dir exists", status);
}
+ @Test
+ public void testMakeDirsWithAnFakeDirectory() throws Exception {
+ /*
+ * Op 1. commit a key -> "dir1/dir2/key1"
+ * Op 2. create dir -> "dir1/testDir", the dir1 is a fake dir,
+ * "dir1/testDir" can be created normal
+ */
+
+ String fakeGrandpaKey = "dir1";
+ String fakeParentKey = fakeGrandpaKey + "/dir2";
+ String fullKeyName = fakeParentKey + "/key1";
+ TestDataUtil.createKey(ozoneBucket, fullKeyName, "");
+
+ // /dir1/dir2 should not exist
+ assertFalse(fs.exists(new Path(fakeParentKey)));
+
+ // /dir1/dir2/key2 should be created because has a fake parent directory
+ Path subdir = new Path(fakeParentKey, "key2");
+ assertTrue(fs.mkdirs(subdir));
+ // the intermediate directories /dir1 and /dir1/dir2 will be created too
+ assertTrue(fs.exists(new Path(fakeGrandpaKey)));
+ assertTrue(fs.exists(new Path(fakeParentKey)));
+ }
+
@Test
public void testCreateWithInvalidPaths() throws Exception {
// Test for path with ..
@@ -727,6 +751,37 @@ public class TestOzoneFileSystem {
}
}
+ @Test
+ public void testListStatusOnKeyNameContainDelimiter() throws Exception {
+ /*
+ * op1: create a key -> "dir1/dir2/key1"
+ * op2: `ls /` child dir "/dir1/" will be return
+ * op2: `ls /dir1` child dir "/dir1/dir2/" will be return
+ * op3: `ls /dir1/dir2` file "/dir1/dir2/key" will be return
+ *
+ * the "/dir1", "/dir1/dir2/" are fake directory
+ * */
+ String keyName = "dir1/dir2/key1";
+ TestDataUtil.createKey(ozoneBucket, keyName, "");
+ FileStatus[] fileStatuses;
+
+ fileStatuses = fs.listStatus(new Path("/"));
+ assertEquals(1, fileStatuses.length);
+ assertEquals("/dir1", fileStatuses[0].getPath().toUri().getPath());
+ assertTrue(fileStatuses[0].isDirectory());
+
+ fileStatuses = fs.listStatus(new Path("/dir1"));
+ assertEquals(1, fileStatuses.length);
+ assertEquals("/dir1/dir2", fileStatuses[0].getPath().toUri().getPath());
+ assertTrue(fileStatuses[0].isDirectory());
+
+ fileStatuses = fs.listStatus(new Path("/dir1/dir2"));
+ assertEquals(1, fileStatuses.length);
+ assertEquals("/dir1/dir2/key1",
+ fileStatuses[0].getPath().toUri().getPath());
+ assertTrue(fileStatuses[0].isFile());
+ }
+
/**
* Cleanup files and directories.
*
@@ -1273,6 +1328,24 @@ public class TestOzoneFileSystem {
"file1")));
}
+ @Test
+ public void testRenameContainDelimiterFile() throws Exception {
+ String fakeGrandpaKey = "dir1";
+ String fakeParentKey = fakeGrandpaKey + "/dir2";
+ String sourceKeyName = fakeParentKey + "/key1";
+ String targetKeyName = fakeParentKey + "/key2";
+ TestDataUtil.createKey(ozoneBucket, sourceKeyName, "");
+
+ Path sourcePath = new Path(fs.getUri().toString() + "/" + sourceKeyName);
+ Path targetPath = new Path(fs.getUri().toString() + "/" + targetKeyName);
+ assertTrue(fs.rename(sourcePath, targetPath));
+ assertFalse(fs.exists(sourcePath));
+ assertTrue(fs.exists(targetPath));
+ // intermediate directories will not be created
+ assertFalse(fs.exists(new Path(fakeGrandpaKey)));
+ assertFalse(fs.exists(new Path(fakeParentKey)));
+ }
+
/**
* Fails if the (a) parent of dst does not exist or (b) parent is a file.
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestKeyManagerImpl.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestKeyManagerImpl.java
index 492173b71c..e30d27fc74 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestKeyManagerImpl.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestKeyManagerImpl.java
@@ -800,19 +800,20 @@ public class TestKeyManagerImpl {
List<OmKeyLocationInfo> locationList =
keySession.getKeyInfo().getLatestVersionLocations().getLocationList();
Assert.assertEquals(1, locationList.size());
+ long containerID = locationList.get(0).getContainerID();
locationInfoList.add(
new OmKeyLocationInfo.Builder().setPipeline(pipeline)
- .setBlockID(new BlockID(locationList.get(0).getContainerID(),
+ .setBlockID(new BlockID(containerID,
locationList.get(0).getLocalID())).build());
keyArgs.setLocationInfoList(locationInfoList);
writeClient.commitKey(keyArgs, keySession.getId());
- ContainerInfo containerInfo = new ContainerInfo.Builder().setContainerID(1L)
- .setPipelineID(pipeline.getId()).build();
+ ContainerInfo containerInfo = new ContainerInfo.Builder()
+ .setContainerID(containerID).setPipelineID(pipeline.getId()).build();
List<ContainerWithPipeline> containerWithPipelines = Arrays.asList(
new ContainerWithPipeline(containerInfo, pipeline));
when(mockScmContainerClient.getContainerWithPipelineBatch(
- Arrays.asList(1L))).thenReturn(containerWithPipelines);
+ Arrays.asList(containerID))).thenReturn(containerWithPipelines);
OmKeyInfo key = keyManager.lookupKey(keyArgs, null);
Assert.assertEquals(key.getKeyName(), keyName);
@@ -1273,6 +1274,67 @@ public class TestKeyManagerImpl {
}
}
+ @Test
+ public void testGetFileStatus() throws IOException {
+ // create a key
+ String keyName = RandomStringUtils.randomAlphabetic(5);
+ OmKeyArgs keyArgs = createBuilder()
+ .setKeyName(keyName)
+ .setLatestVersionLocation(true)
+ .build();
+ writeClient.createFile(keyArgs, false, false);
+ OpenKeySession keySession = writeClient.createFile(keyArgs, true, true);
+ keyArgs.setLocationInfoList(
+ keySession.getKeyInfo().getLatestVersionLocations().getLocationList());
+ writeClient.commitKey(keyArgs, keySession.getId());
+ OzoneFileStatus ozoneFileStatus = keyManager.getFileStatus(keyArgs);
+ Assert.assertEquals(keyName, ozoneFileStatus.getKeyInfo().getFileName());
+ }
+
+ @Test
+ public void testGetFileStatusWithFakeDir() throws IOException {
+ String parentDir = "dir1";
+ String key = "key1";
+ String fullKeyName = parentDir + OZONE_URI_DELIMITER + key;
+ OzoneFileStatus ozoneFileStatus;
+
+ // create a key "dir1/key1"
+ OmKeyArgs keyArgs = createBuilder().setKeyName(fullKeyName).build();
+ OpenKeySession keySession = writeClient.openKey(keyArgs);
+ keyArgs.setLocationInfoList(
+ keySession.getKeyInfo().getLatestVersionLocations().getLocationList());
+ writeClient.commitKey(keyArgs, keySession.getId());
+
+ // verify
+ String keyArg;
+ keyArg = metadataManager.getOzoneKey(VOLUME_NAME, BUCKET_NAME, parentDir);
+ Assert.assertNull(
+ metadataManager.getKeyTable(getDefaultBucketLayout()).get(keyArg));
+ keyArg = metadataManager.getOzoneKey(VOLUME_NAME, BUCKET_NAME, fullKeyName);
+ Assert.assertNotNull(metadataManager.getKeyTable(getDefaultBucketLayout())
+ .get(keyArg));
+
+ // get a non-existing "dir1", since the key is prefixed "dir1/key1",
+ // a fake "/dir1" will be returned
+ keyArgs = createBuilder().setKeyName(parentDir).build();
+ ozoneFileStatus = keyManager.getFileStatus(keyArgs);
+ Assert.assertEquals(parentDir, ozoneFileStatus.getKeyInfo().getFileName());
+ Assert.assertTrue(ozoneFileStatus.isDirectory());
+
+ // get a non-existing "dir", since the key is not prefixed "dir1/key1",
+ // a `OMException` will be thrown
+ keyArgs = createBuilder().setKeyName("dir").build();
+ OmKeyArgs finalKeyArgs = keyArgs;
+ Assert.assertThrows(OMException.class, () -> keyManager.getFileStatus(
+ finalKeyArgs));
+
+ // get a file "dir1/key1"
+ keyArgs = createBuilder().setKeyName(fullKeyName).build();
+ ozoneFileStatus = keyManager.getFileStatus(keyArgs);
+ Assert.assertEquals(key, ozoneFileStatus.getKeyInfo().getFileName());
+ Assert.assertTrue(ozoneFileStatus.isFile());
+ }
+
@Test
public void testRefreshPipeline() throws Exception {
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java
index dd66c33a1e..483a2e88b0 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java
@@ -1209,6 +1209,8 @@ public class KeyManagerImpl implements KeyManager {
final String keyName = args.getKeyName();
OmKeyInfo fileKeyInfo = null;
+ OmKeyInfo dirKeyInfo = null;
+ OmKeyInfo fakeDirKeyInfo = null;
metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volumeName,
bucketName);
try {
@@ -1221,28 +1223,27 @@ public class KeyManagerImpl implements KeyManager {
// Check if the key is a file.
String fileKeyBytes = metadataManager.getOzoneKey(
volumeName, bucketName, keyName);
- fileKeyInfo = metadataManager
- .getKeyTable(getBucketLayout(metadataManager, volumeName, bucketName))
- .get(fileKeyBytes);
+ BucketLayout layout =
+ getBucketLayout(metadataManager, volumeName, bucketName);
+ fileKeyInfo = metadataManager.getKeyTable(layout).get(fileKeyBytes);
+ String dirKey = OzoneFSUtils.addTrailingSlashIfNeeded(keyName);
// Check if the key is a directory.
if (fileKeyInfo == null) {
- String dirKey = OzoneFSUtils.addTrailingSlashIfNeeded(keyName);
String dirKeyBytes = metadataManager.getOzoneKey(
volumeName, bucketName, dirKey);
- OmKeyInfo dirKeyInfo = metadataManager.getKeyTable(
- getBucketLayout(metadataManager, volumeName, bucketName))
- .get(dirKeyBytes);
- if (dirKeyInfo != null) {
- return new OzoneFileStatus(dirKeyInfo, scmBlockSize, true);
+ dirKeyInfo = metadataManager.getKeyTable(layout).get(dirKeyBytes);
+ if (dirKeyInfo == null) {
+ fakeDirKeyInfo =
+ createFakeDirIfShould(volumeName, bucketName, keyName, layout);
}
}
} finally {
metadataManager.getLock().releaseReadLock(BUCKET_LOCK, volumeName,
bucketName);
-
- // if the key is a file then do refresh pipeline info in OM by asking SCM
if (fileKeyInfo != null) {
+ // if the key is a file
+ // then do refresh pipeline info in OM by asking SCM
if (args.getLatestVersionLocation()) {
slimLocationVersion(fileKeyInfo);
}
@@ -1257,10 +1258,21 @@ public class KeyManagerImpl implements KeyManager {
sortDatanodes(clientAddress, fileKeyInfo);
}
}
- return new OzoneFileStatus(fileKeyInfo, scmBlockSize, false);
}
}
+ if (fileKeyInfo != null) {
+ return new OzoneFileStatus(fileKeyInfo, scmBlockSize, false);
+ }
+
+ if (dirKeyInfo != null) {
+ return new OzoneFileStatus(dirKeyInfo, scmBlockSize, true);
+ }
+
+ if (fakeDirKeyInfo != null) {
+ return new OzoneFileStatus(fakeDirKeyInfo, scmBlockSize, true);
+ }
+
// Key is not found, throws exception
if (LOG.isDebugEnabled()) {
LOG.debug("Unable to get file status for the key: volume: {}, bucket:" +
@@ -1272,6 +1284,37 @@ public class KeyManagerImpl implements KeyManager {
FILE_NOT_FOUND);
}
+ /**
+ * Create a fake directory if the key is a path prefix,
+ * otherwise returns null.
+ * Some keys may contain '/' Ozone will treat '/' as directory separator
+ * such as : key name is 'a/b/c', 'a' and 'b' may not really exist,
+ * but Ozone treats 'a' and 'b' as a directory.
+ * we need create a fake directory 'a' or 'a/b'
+ *
+ * @return OmKeyInfo if the key is a path prefix, otherwise returns null.
+ */
+ private OmKeyInfo createFakeDirIfShould(String volume, String bucket,
+ String keyName, BucketLayout layout) throws IOException {
+ OmKeyInfo fakeDirKeyInfo = null;
+ String dirKey = OzoneFSUtils.addTrailingSlashIfNeeded(keyName);
+ String fileKeyBytes = metadataManager.getOzoneKey(volume, bucket, keyName);
+ Table.KeyValue<String, OmKeyInfo> keyValue =
+ metadataManager.getKeyTable(layout).iterator().seek(fileKeyBytes);
+
+ if (keyValue != null) {
+ Path fullPath = Paths.get(keyValue.getValue().getKeyName());
+ Path subPath = Paths.get(dirKey);
+ OmKeyInfo omKeyInfo = keyValue.getValue();
+ if (fullPath.startsWith(subPath)) {
+ // create fake directory
+ fakeDirKeyInfo = createDirectoryKey(omKeyInfo, dirKey);
+ }
+ }
+
+ return fakeDirKeyInfo;
+ }
+
private OzoneFileStatus getOzoneFileStatusFSO(OmKeyArgs args,
String clientAddress, boolean skipFileNotFoundError) throws IOException {
@@ -1349,6 +1392,7 @@ public class KeyManagerImpl implements KeyManager {
.setVolumeName(keyInfo.getVolumeName())
.setBucketName(keyInfo.getBucketName())
.setKeyName(dir)
+ .setFileName(OzoneFSUtils.getFileName(keyName))
.setOmKeyLocationInfos(Collections.singletonList(
new OmKeyLocationInfoGroup(0, new ArrayList<>())))
.setCreationTime(Time.now())
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org
[ozone] 02/03: HDDS-7258. Cleanup the allocated but uncommitted blocks (#3778)
Posted by ck...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ckj pushed a commit to branch ozone-1.3
in repository https://gitbox.apache.org/repos/asf/ozone.git
commit b2ff74f7f5ed71faca9c7bdcccdb18b5bb27f050
Author: Nibiru <ax...@qq.com>
AuthorDate: Wed Oct 26 01:00:02 2022 +0800
HDDS-7258. Cleanup the allocated but uncommitted blocks (#3778)
---
.../apache/hadoop/ozone/om/helpers/OmKeyInfo.java | 76 ++++++++----
.../ozone/om/request/key/OMKeyCommitRequest.java | 19 ++-
.../om/request/key/OMKeyCommitRequestWithFSO.java | 23 +++-
.../hadoop/ozone/om/request/key/OMKeyRequest.java | 24 ++++
.../ozone/om/response/key/OMKeyCommitResponse.java | 6 +
.../om/request/key/TestOMKeyCommitRequest.java | 96 ++++++++++++++-
.../ozone/om/service/TestKeyDeletingService.java | 136 +++++++++++++++++++--
7 files changed, 333 insertions(+), 47 deletions(-)
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java
index 9b2014dd7b..f8f589af2e 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java
@@ -26,9 +26,9 @@ import java.util.Map;
import java.util.Objects;
import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.fs.FileChecksum;
import org.apache.hadoop.fs.FileEncryptionInfo;
-import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.client.ContainerBlockID;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.client.ReplicationConfig;
@@ -180,25 +180,30 @@ public final class OmKeyInfo extends WithParentObjectId {
/**
* updates the length of the each block in the list given.
* This will be called when the key is being committed to OzoneManager.
+ * Return the uncommitted locationInfo to be deleted.
*
* @param locationInfoList list of locationInfo
+ * @return allocated but uncommitted locationInfos
*/
- public void updateLocationInfoList(List<OmKeyLocationInfo> locationInfoList,
- boolean isMpu) {
- updateLocationInfoList(locationInfoList, isMpu, false);
+ public List<OmKeyLocationInfo> updateLocationInfoList(
+ List<OmKeyLocationInfo> locationInfoList, boolean isMpu) {
+ return updateLocationInfoList(locationInfoList, isMpu, false);
}
/**
* updates the length of the each block in the list given.
* This will be called when the key is being committed to OzoneManager.
+ * Return the uncommitted locationInfo to be deleted.
*
* @param locationInfoList list of locationInfo
* @param isMpu a true represents multi part key, false otherwise
* @param skipBlockIDCheck a true represents that the blockId verification
* check should be skipped, false represents that
* the blockId verification will be required
+ * @return allocated but uncommitted locationInfos
*/
- public void updateLocationInfoList(List<OmKeyLocationInfo> locationInfoList,
+ public List<OmKeyLocationInfo> updateLocationInfoList(
+ List<OmKeyLocationInfo> locationInfoList,
boolean isMpu, boolean skipBlockIDCheck) {
long latestVersion = getLatestVersionLocations().getVersion();
OmKeyLocationInfoGroup keyLocationInfoGroup = getLatestVersionLocations();
@@ -207,51 +212,68 @@ public final class OmKeyInfo extends WithParentObjectId {
// Compare user given block location against allocatedBlockLocations
// present in OmKeyInfo.
+ List<OmKeyLocationInfo> uncommittedBlocks;
List<OmKeyLocationInfo> updatedBlockLocations;
if (skipBlockIDCheck) {
updatedBlockLocations = locationInfoList;
+ uncommittedBlocks = new ArrayList<>();
} else {
- updatedBlockLocations =
+ Pair<List<OmKeyLocationInfo>, List<OmKeyLocationInfo>> verifiedResult =
verifyAndGetKeyLocations(locationInfoList, keyLocationInfoGroup);
+ updatedBlockLocations = verifiedResult.getLeft();
+ uncommittedBlocks = verifiedResult.getRight();
}
- // Updates the latest locationList in the latest version only with
- // given locationInfoList here.
- // TODO : The original allocated list and the updated list here may vary
- // as the containers on the Datanode on which the blocks were pre allocated
- // might get closed. The diff of blocks between these two lists here
- // need to be garbage collected in case the ozone client dies.
+
keyLocationInfoGroup.removeBlocks(latestVersion);
// set each of the locationInfo object to the latest version
updatedBlockLocations.forEach(omKeyLocationInfo -> omKeyLocationInfo
.setCreateVersion(latestVersion));
keyLocationInfoGroup.addAll(latestVersion, updatedBlockLocations);
- }
- private List<OmKeyLocationInfo> verifyAndGetKeyLocations(
- List<OmKeyLocationInfo> locationInfoList,
- OmKeyLocationInfoGroup keyLocationInfoGroup) {
-
- List<OmKeyLocationInfo> allocatedBlockLocations =
- keyLocationInfoGroup.getBlocksLatestVersionOnly();
- List<OmKeyLocationInfo> updatedBlockLocations = new ArrayList<>();
+ return uncommittedBlocks;
+ }
- List<ContainerBlockID> existingBlockIDs = new ArrayList<>();
- for (OmKeyLocationInfo existingLocationInfo : allocatedBlockLocations) {
- BlockID existingBlockID = existingLocationInfo.getBlockID();
- existingBlockIDs.add(existingBlockID.getContainerBlockID());
+ /**
+ * 1. Verify committed KeyLocationInfos
+ * 2. Find out the allocated but uncommitted KeyLocationInfos.
+ *
+ * @param locationInfoList committed KeyLocationInfos
+ * @param keyLocationInfoGroup allocated KeyLocationInfoGroup
+ * @return Pair of updatedOmKeyLocationInfo and uncommittedOmKeyLocationInfo
+ */
+ private Pair<List<OmKeyLocationInfo>, List<OmKeyLocationInfo>>
+ verifyAndGetKeyLocations(
+ List<OmKeyLocationInfo> locationInfoList,
+ OmKeyLocationInfoGroup keyLocationInfoGroup) {
+ // Only check ContainerBlockID here to avoid the mismatch of the pipeline
+ // field and BcsId in the OmKeyLocationInfo, as the OmKeyInfoCodec ignores
+ // the pipeline field by default and bcsId would be updated in Ratis mode.
+ Map<ContainerBlockID, OmKeyLocationInfo> allocatedBlockLocations =
+ new HashMap<>();
+ for (OmKeyLocationInfo existingLocationInfo : keyLocationInfoGroup.
+ getLocationList()) {
+ ContainerBlockID existingBlockID = existingLocationInfo.getBlockID().
+ getContainerBlockID();
+ // The case of overwriting value should never happen
+ allocatedBlockLocations.put(existingBlockID, existingLocationInfo);
}
+ List<OmKeyLocationInfo> updatedBlockLocations = new ArrayList<>();
for (OmKeyLocationInfo modifiedLocationInfo : locationInfoList) {
- BlockID modifiedBlockID = modifiedLocationInfo.getBlockID();
- if (existingBlockIDs.contains(modifiedBlockID.getContainerBlockID())) {
+ ContainerBlockID modifiedContainerBlockId =
+ modifiedLocationInfo.getBlockID().getContainerBlockID();
+ if (allocatedBlockLocations.containsKey(modifiedContainerBlockId)) {
updatedBlockLocations.add(modifiedLocationInfo);
+ allocatedBlockLocations.remove(modifiedContainerBlockId);
} else {
LOG.warn("Unknown BlockLocation:{}, where the blockID of given "
+ "location doesn't match with the stored/allocated block of"
+ " keyName:{}", modifiedLocationInfo, keyName);
}
}
- return updatedBlockLocations;
+ List<OmKeyLocationInfo> uncommittedLocationInfos = new ArrayList<>(
+ allocatedBlockLocations.values());
+ return Pair.of(updatedBlockLocations, uncommittedLocationInfos);
}
/**
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java
index ebff5416f3..bb070b6ba1 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java
@@ -218,8 +218,11 @@ public class OMKeyCommitRequest extends OMKeyRequest {
omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
omKeyInfo.setDataSize(commitKeyArgs.getDataSize());
omKeyInfo.setModificationTime(commitKeyArgs.getModificationTime());
- // Update the block length for each block
- omKeyInfo.updateLocationInfoList(locationInfoList, false);
+ // Update the block length for each block, return the allocated but
+ // uncommitted blocks
+ List<OmKeyLocationInfo> uncommitted = omKeyInfo.updateLocationInfoList(
+ locationInfoList, false);
+
// Set the UpdateID to current transactionLogIndex
omKeyInfo.setUpdateID(trxnLogIndex, ozoneManager.isRatisEnabled());
@@ -239,6 +242,18 @@ public class OMKeyCommitRequest extends OMKeyRequest {
omBucketInfo.incrUsedNamespace(1L);
}
+ // let the uncommitted blocks pretend as key's old version blocks
+ // which will be deleted as RepeatedOmKeyInfo
+ OmKeyInfo pseudoKeyInfo = wrapUncommittedBlocksAsPseudoKey(uncommitted,
+ omKeyInfo);
+ if (pseudoKeyInfo != null) {
+ if (oldKeyVersionsToDelete != null) {
+ oldKeyVersionsToDelete.addOmKeyInfo(pseudoKeyInfo);
+ } else {
+ oldKeyVersionsToDelete = new RepeatedOmKeyInfo(pseudoKeyInfo);
+ }
+ }
+
// Add to cache of open key table and key table.
omMetadataManager.getOpenKeyTable(getBucketLayout()).addCacheEntry(
new CacheKey<>(dbOpenKey),
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequestWithFSO.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequestWithFSO.java
index b9239c9d86..5c7ac450b8 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequestWithFSO.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequestWithFSO.java
@@ -41,6 +41,8 @@ import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyLoca
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.file.Path;
@@ -58,6 +60,9 @@ import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.BUCKET_L
*/
public class OMKeyCommitRequestWithFSO extends OMKeyCommitRequest {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(OMKeyCommitRequestWithFSO.class);
+
public OMKeyCommitRequestWithFSO(OMRequest omRequest,
BucketLayout bucketLayout) {
super(omRequest, bucketLayout);
@@ -144,10 +149,8 @@ public class OMKeyCommitRequestWithFSO extends OMKeyCommitRequest {
omKeyInfo.setModificationTime(commitKeyArgs.getModificationTime());
- // Update the block length for each block
- List<OmKeyLocationInfo> allocatedLocationInfoList =
- omKeyInfo.getLatestVersionLocations().getLocationList();
- omKeyInfo.updateLocationInfoList(locationInfoList, false);
+ List<OmKeyLocationInfo> uncommitted = omKeyInfo.updateLocationInfoList(
+ locationInfoList, false);
// Set the UpdateID to current transactionLogIndex
omKeyInfo.setUpdateID(trxnLogIndex, ozoneManager.isRatisEnabled());
@@ -177,6 +180,18 @@ public class OMKeyCommitRequestWithFSO extends OMKeyCommitRequest {
omBucketInfo.incrUsedNamespace(1L);
}
+ // let the uncommitted blocks pretend as key's old version blocks
+ // which will be deleted as RepeatedOmKeyInfo
+ OmKeyInfo pseudoKeyInfo = wrapUncommittedBlocksAsPseudoKey(uncommitted,
+ omKeyInfo);
+ if (pseudoKeyInfo != null) {
+ if (oldKeyVersionsToDelete != null) {
+ oldKeyVersionsToDelete.addOmKeyInfo(pseudoKeyInfo);
+ } else {
+ oldKeyVersionsToDelete = new RepeatedOmKeyInfo(pseudoKeyInfo);
+ }
+ }
+
// Add to cache of open key table and key table.
OMFileRequest.addOpenFileTableCacheEntry(omMetadataManager, dbFileKey,
null, fileName, trxnLogIndex);
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java
index 904b5c0005..8c79e16dcd 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java
@@ -816,4 +816,28 @@ public abstract class OMKeyRequest extends OMClientRequest {
return ozoneManager.getOzoneLockProvider()
.createLockStrategy(getBucketLayout());
}
+
+ /**
+ * Wrap the uncommitted blocks as pseudoKeyInfo.
+ *
+ * @param uncommitted Uncommitted OmKeyLocationInfo
+ * @param omKeyInfo Args for key block
+ * @return pseudoKeyInfo
+ */
+ protected OmKeyInfo wrapUncommittedBlocksAsPseudoKey(
+ List<OmKeyLocationInfo> uncommitted, OmKeyInfo omKeyInfo) {
+ if (uncommitted.isEmpty()) {
+ return null;
+ }
+ LOG.info("Detect allocated but uncommitted blocks {} in key {}.",
+ uncommitted, omKeyInfo.getKeyName());
+ OmKeyInfo pseudoKeyInfo = omKeyInfo.copyObject();
+ // TODO dataSize of pseudoKey is not real here
+ List<OmKeyLocationInfoGroup> uncommittedGroups = new ArrayList<>();
+ // version not matters in the current logic of keyDeletingService,
+ // all versions of blocks will be deleted.
+ uncommittedGroups.add(new OmKeyLocationInfoGroup(0, uncommitted));
+ pseudoKeyInfo.setKeyLocationVersions(uncommittedGroups);
+ return pseudoKeyInfo;
+ }
}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyCommitResponse.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyCommitResponse.java
index 513e10cba4..911a61cce6 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyCommitResponse.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyCommitResponse.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.ozone.om.response.key;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.helpers.BucketLayout;
import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
@@ -104,6 +105,11 @@ public class OMKeyCommitResponse extends OmKeyResponse {
return ozoneKeyName;
}
+ @VisibleForTesting
+ public RepeatedOmKeyInfo getKeysToDelete() {
+ return keysToDelete;
+ }
+
protected void updateDeletedTable(OMMetadataManager omMetadataManager,
BatchOperation batchOperation) throws IOException {
if (this.keysToDelete != null) {
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMKeyCommitRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMKeyCommitRequest.java
index bdf16fef42..10552e380d 100644
--- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMKeyCommitRequest.java
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMKeyCommitRequest.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.ozone.om.helpers.BucketLayout;
import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
import org.apache.hadoop.ozone.om.helpers.OzoneFSUtils;
import org.apache.hadoop.ozone.om.request.OMRequestTestUtils;
+import org.apache.hadoop.ozone.om.response.key.OMKeyCommitResponse;
import org.apache.hadoop.util.Time;
import org.jetbrains.annotations.NotNull;
import org.junit.Assert;
@@ -190,6 +191,93 @@ public class TestOMKeyCommitRequest extends TestOMKeyRequest {
omKeyInfo.getLatestVersionLocations().getLocationList());
}
+ @Test
+ public void testValidateAndUpdateCacheWithUncommittedBlocks()
+ throws Exception {
+
+ // allocated block list
+ List<KeyLocation> allocatedKeyLocationList = getKeyLocation(5);
+
+ List<OmKeyLocationInfo> allocatedBlockList = allocatedKeyLocationList
+ .stream().map(OmKeyLocationInfo::getFromProtobuf)
+ .collect(Collectors.toList());
+
+ // committed block list, with three blocks different with the allocated
+ List<KeyLocation> committedKeyLocationList = getKeyLocation(3);
+
+ OMRequest modifiedOmRequest = doPreExecute(createCommitKeyRequest(
+ committedKeyLocationList));
+
+ OMKeyCommitRequest omKeyCommitRequest =
+ getOmKeyCommitRequest(modifiedOmRequest);
+
+ OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName,
+ omMetadataManager, omKeyCommitRequest.getBucketLayout());
+
+ String ozoneKey = addKeyToOpenKeyTable(allocatedBlockList);
+
+ // Key should not be there in key table, as validateAndUpdateCache is
+ // still not called.
+ OmKeyInfo omKeyInfo =
+ omMetadataManager.getKeyTable(omKeyCommitRequest.getBucketLayout())
+ .get(ozoneKey);
+
+ Assert.assertNull(omKeyInfo);
+
+ OMClientResponse omClientResponse =
+ omKeyCommitRequest.validateAndUpdateCache(ozoneManager,
+ 100L, ozoneManagerDoubleBufferHelper);
+
+ Assert.assertEquals(OzoneManagerProtocolProtos.Status.OK,
+ omClientResponse.getOMResponse().getStatus());
+
+ List<OmKeyInfo> toDeleteKeyList = ((OMKeyCommitResponse) omClientResponse).
+ getKeysToDelete().cloneOmKeyInfoList();
+
+ // This is the first time to commit key, only the allocated but uncommitted
+ // blocks should be deleted.
+ Assert.assertEquals(1, toDeleteKeyList.size());
+ Assert.assertEquals(2, toDeleteKeyList.get(0).
+ getKeyLocationVersions().get(0).getLocationList().size());
+
+ // Entry should be deleted from openKey Table.
+ omKeyInfo =
+ omMetadataManager.getOpenKeyTable(omKeyCommitRequest.getBucketLayout())
+ .get(ozoneKey);
+ Assert.assertNull(omKeyInfo);
+
+ // Now entry should be created in key Table.
+ omKeyInfo =
+ omMetadataManager.getKeyTable(omKeyCommitRequest.getBucketLayout())
+ .get(ozoneKey);
+
+ Assert.assertNotNull(omKeyInfo);
+
+ // DB keyInfo format
+ verifyKeyName(omKeyInfo);
+
+ // Check modification time
+ CommitKeyRequest commitKeyRequest = modifiedOmRequest.getCommitKeyRequest();
+ Assert.assertEquals(commitKeyRequest.getKeyArgs().getModificationTime(),
+ omKeyInfo.getModificationTime());
+
+ // Check block location.
+ List<OmKeyLocationInfo> locationInfoListFromCommitKeyRequest =
+ commitKeyRequest.getKeyArgs()
+ .getKeyLocationsList().stream()
+ .map(OmKeyLocationInfo::getFromProtobuf)
+ .collect(Collectors.toList());
+
+ List<OmKeyLocationInfo> intersection = new ArrayList<>(allocatedBlockList);
+ intersection.retainAll(locationInfoListFromCommitKeyRequest);
+
+ // Key table should have three blocks.
+ Assert.assertEquals(intersection,
+ omKeyInfo.getLatestVersionLocations().getLocationList());
+ Assert.assertEquals(3, intersection.size());
+
+ }
+
@Test
public void testValidateAndUpdateCacheWithSubDirs() throws Exception {
parentDir = "dir1/dir2/dir3/";
@@ -466,15 +554,19 @@ public class TestOMKeyCommitRequest extends TestOMKeyRequest {
modifiedKeyArgs.getFactor());
}
+ private OMRequest createCommitKeyRequest() {
+ return createCommitKeyRequest(getKeyLocation(5));
+ }
+
/**
* Create OMRequest which encapsulates CommitKeyRequest.
*/
- private OMRequest createCommitKeyRequest() {
+ private OMRequest createCommitKeyRequest(List<KeyLocation> keyLocations) {
KeyArgs keyArgs =
KeyArgs.newBuilder().setDataSize(dataSize).setVolumeName(volumeName)
.setKeyName(keyName).setBucketName(bucketName)
.setType(replicationType).setFactor(replicationFactor)
- .addAllKeyLocations(getKeyLocation(5)).build();
+ .addAllKeyLocations(keyLocations).build();
CommitKeyRequest commitKeyRequest =
CommitKeyRequest.newBuilder().setKeyArgs(keyArgs)
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyDeletingService.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyDeletingService.java
index a24a72ac80..3bc35c8b3b 100644
--- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyDeletingService.java
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyDeletingService.java
@@ -22,14 +22,21 @@ package org.apache.hadoop.ozone.om.service;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
import org.apache.hadoop.ozone.om.KeyManager;
import org.apache.hadoop.ozone.om.OmTestManagers;
import org.apache.hadoop.ozone.om.OzoneManager;
import org.apache.hadoop.ozone.om.ScmBlockLocationTestingClient;
+import org.apache.hadoop.ozone.common.BlockGroup;
+import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
+import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
import org.apache.hadoop.ozone.om.request.OMRequestTestUtils;
import org.apache.ratis.util.ExitUtils;
import org.junit.BeforeClass;
@@ -133,8 +140,8 @@ public class TestKeyDeletingService {
() -> keyDeletingService.getDeletedKeyCount().get() >= keyCount,
1000, 10000);
Assert.assertTrue(keyDeletingService.getRunCount().get() > 1);
- Assert.assertEquals(
- keyManager.getPendingDeletionKeys(Integer.MAX_VALUE).size(), 0);
+ Assert.assertEquals(0,
+ keyManager.getPendingDeletionKeys(Integer.MAX_VALUE).size());
}
@Test(timeout = 40000)
@@ -176,9 +183,9 @@ public class TestKeyDeletingService {
() -> keyDeletingService.getRunCount().get() >= 5,
100, 1000);
// Since SCM calls are failing, deletedKeyCount should be zero.
- Assert.assertEquals(keyDeletingService.getDeletedKeyCount().get(), 0);
- Assert.assertEquals(
- keyManager.getPendingDeletionKeys(Integer.MAX_VALUE).size(), keyCount);
+ Assert.assertEquals(0, keyDeletingService.getDeletedKeyCount().get());
+ Assert.assertEquals(keyCount,
+ keyManager.getPendingDeletionKeys(Integer.MAX_VALUE).size());
}
@Test(timeout = 30000)
@@ -200,15 +207,86 @@ public class TestKeyDeletingService {
KeyDeletingService keyDeletingService =
(KeyDeletingService) keyManager.getDeletingService();
- // Since empty keys are directly deleted from db there should be no
- // pending deletion keys. Also deletedKeyCount should be zero.
- Assert.assertEquals(
- keyManager.getPendingDeletionKeys(Integer.MAX_VALUE).size(), 0);
+ // the pre-allocated blocks are not committed, hence they will be deleted.
+ Assert.assertEquals(100,
+ keyManager.getPendingDeletionKeys(Integer.MAX_VALUE).size());
// Make sure that we have run the background thread 2 times or more
GenericTestUtils.waitFor(
() -> keyDeletingService.getRunCount().get() >= 2,
100, 1000);
- Assert.assertEquals(keyDeletingService.getDeletedKeyCount().get(), 0);
+ // the blockClient is set to fail the deletion of key blocks, hence no keys
+ // will be deleted
+ Assert.assertEquals(0, keyDeletingService.getDeletedKeyCount().get());
+ }
+
+ @Test(timeout = 30000)
+ public void checkDeletionForPartiallyCommitKey()
+ throws IOException, TimeoutException, InterruptedException,
+ AuthenticationException {
+ OzoneConfiguration conf = createConfAndInitValues();
+ ScmBlockLocationProtocol blockClient =
+ //failCallsFrequency = 1 , means all calls fail.
+ new ScmBlockLocationTestingClient(null, null, 1);
+ OmTestManagers omTestManagers
+ = new OmTestManagers(conf, blockClient, null);
+ KeyManager keyManager = omTestManagers.getKeyManager();
+ writeClient = omTestManagers.getWriteClient();
+ om = omTestManagers.getOzoneManager();
+
+ String volumeName = String.format("volume%s",
+ RandomStringUtils.randomAlphanumeric(5));
+ String bucketName = String.format("bucket%s",
+ RandomStringUtils.randomAlphanumeric(5));
+ String keyName = String.format("key%s",
+ RandomStringUtils.randomAlphanumeric(5));
+
+ // Create Volume and Bucket
+ createVolumeAndBucket(keyManager, volumeName, bucketName, false);
+
+ OmKeyArgs keyArg = createAndCommitKey(keyManager, volumeName, bucketName,
+ keyName, 3, 1);
+
+ // Only the uncommitted block should be pending to be deleted.
+ GenericTestUtils.waitFor(
+ () -> {
+ try {
+ return keyManager.getPendingDeletionKeys(Integer.MAX_VALUE)
+ .stream()
+ .map(BlockGroup::getBlockIDList)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList()).size() == 1;
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ return false;
+ },
+ 500, 3000);
+
+ // Delete the key
+ writeClient.deleteKey(keyArg);
+
+ KeyDeletingService keyDeletingService =
+ (KeyDeletingService) keyManager.getDeletingService();
+
+ // All blocks should be pending to be deleted.
+ GenericTestUtils.waitFor(
+ () -> {
+ try {
+ return keyManager.getPendingDeletionKeys(Integer.MAX_VALUE)
+ .stream()
+ .map(BlockGroup::getBlockIDList)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList()).size() == 3;
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ return false;
+ },
+ 500, 3000);
+
+ // the blockClient is set to fail the deletion of key blocks, hence no keys
+ // will be deleted
+ Assert.assertEquals(0, keyDeletingService.getDeletedKeyCount().get());
}
@Test(timeout = 30000)
@@ -299,6 +377,15 @@ public class TestKeyDeletingService {
private OmKeyArgs createAndCommitKey(KeyManager keyManager, String volumeName,
String bucketName, String keyName, int numBlocks) throws IOException {
+ return createAndCommitKey(keyManager, volumeName, bucketName, keyName,
+ numBlocks, 0);
+ }
+
+ private OmKeyArgs createAndCommitKey(KeyManager keyManager, String volumeName,
+ String bucketName, String keyName, int numBlocks, int numUncommitted)
+ throws IOException {
+ // Even if no key size is appointed, there will be at least one
+ // block pre-allocated when key is created
OmKeyArgs keyArg =
new OmKeyArgs.Builder()
.setVolumeName(volumeName)
@@ -311,10 +398,35 @@ public class TestKeyDeletingService {
.build();
//Open and Commit the Key in the Key Manager.
OpenKeySession session = writeClient.openKey(keyArg);
- for (int i = 0; i < numBlocks; i++) {
- keyArg.addLocationInfo(writeClient.allocateBlock(keyArg, session.getId(),
+
+ // add pre-allocated blocks into args and avoid creating excessive block
+ OmKeyLocationInfoGroup keyLocationVersions = session.getKeyInfo().
+ getLatestVersionLocations();
+ assert keyLocationVersions != null;
+ List<OmKeyLocationInfo> latestBlocks = keyLocationVersions.
+ getBlocksLatestVersionOnly();
+ int preAllocatedSize = latestBlocks.size();
+ for (OmKeyLocationInfo block : latestBlocks) {
+ keyArg.addLocationInfo(block);
+ }
+
+ // allocate blocks until the blocks num equal to numBlocks
+ LinkedList<OmKeyLocationInfo> allocated = new LinkedList<>();
+ for (int i = 0; i < numBlocks - preAllocatedSize; i++) {
+ allocated.add(writeClient.allocateBlock(keyArg, session.getId(),
new ExcludeList()));
}
+
+ // remove the blocks not to be committed
+ for (int i = 0; i < numUncommitted; i++) {
+ allocated.removeFirst();
+ }
+
+ // add the blocks to be committed
+ for (OmKeyLocationInfo block: allocated) {
+ keyArg.addLocationInfo(block);
+ }
+
writeClient.commitKey(keyArg, session.getId());
return keyArg;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org
[ozone] 03/03: HDDS-7396. Force close non-RATIS containers in ReplicationManager (#3877)
Posted by ck...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ckj pushed a commit to branch ozone-1.3
in repository https://gitbox.apache.org/repos/asf/ozone.git
commit 3f6d238a874b2f36c78be1752abf1ab23cfbb8cd
Author: Kaijie Chen <ck...@apache.org>
AuthorDate: Wed Oct 26 09:55:50 2022 +0800
HDDS-7396. Force close non-RATIS containers in ReplicationManager (#3877)
---
.../health/ClosingContainerHandler.java | 5 +-
.../health/TestClosingContainerHandler.java | 103 ++++++++++++---------
2 files changed, 63 insertions(+), 45 deletions(-)
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/ClosingContainerHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/ClosingContainerHandler.java
index 9259d084c0..103f7d6646 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/ClosingContainerHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/health/ClosingContainerHandler.java
@@ -51,10 +51,13 @@ public class ClosingContainerHandler extends AbstractCheck {
return false;
}
+ boolean forceClose = request.getContainerInfo().getReplicationConfig()
+ .getReplicationType() != HddsProtos.ReplicationType.RATIS;
+
for (ContainerReplica replica : request.getContainerReplicas()) {
if (replica.getState() != ContainerReplicaProto.State.UNHEALTHY) {
replicationManager.sendCloseContainerReplicaCommand(
- containerInfo, replica.getDatanodeDetails(), false);
+ containerInfo, replica.getDatanodeDetails(), forceClose);
}
}
return true;
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/health/TestClosingContainerHandler.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/health/TestClosingContainerHandler.java
index 2999884e1d..06c2c7fa63 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/health/TestClosingContainerHandler.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/health/TestClosingContainerHandler.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hdds.scm.container.replication.health;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.client.ReplicationConfig;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
@@ -32,13 +33,20 @@ import org.apache.hadoop.hdds.scm.container.replication.ReplicationTestUtil;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import java.util.Collections;
+import java.util.HashSet;
import java.util.Set;
+import java.util.stream.Stream;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.CLOSED;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.CLOSING;
+import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType.EC;
+import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType.RATIS;
/**
* Tests for {@link ClosingContainerHandler}.
@@ -46,18 +54,21 @@ import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.CL
public class TestClosingContainerHandler {
private ReplicationManager replicationManager;
private ClosingContainerHandler closingContainerHandler;
- private ECReplicationConfig ecReplicationConfig;
- private RatisReplicationConfig ratisReplicationConfig;
+ private static final ECReplicationConfig EC_REPLICATION_CONFIG =
+ new ECReplicationConfig(3, 2);
+ private static final RatisReplicationConfig RATIS_REPLICATION_CONFIG =
+ RatisReplicationConfig.getInstance(HddsProtos.ReplicationFactor.THREE);
@BeforeEach
public void setup() {
- ecReplicationConfig = new ECReplicationConfig(3, 2);
- ratisReplicationConfig = RatisReplicationConfig.getInstance(
- HddsProtos.ReplicationFactor.THREE);
replicationManager = Mockito.mock(ReplicationManager.class);
closingContainerHandler = new ClosingContainerHandler(replicationManager);
}
+ private static Stream<ReplicationConfig> replicationConfigs() {
+ return Stream.of(RATIS_REPLICATION_CONFIG, EC_REPLICATION_CONFIG);
+ }
+
/**
* If a container is not closing, it should not be handled by
* ClosingContainerHandler. It should return false so the request can be
@@ -66,7 +77,7 @@ public class TestClosingContainerHandler {
@Test
public void testNonClosingContainerReturnsFalse() {
ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(
- ecReplicationConfig, 1, CLOSED);
+ EC_REPLICATION_CONFIG, 1, CLOSED);
Set<ContainerReplica> containerReplicas = ReplicationTestUtil
.createReplicas(containerInfo.containerID(),
ContainerReplicaProto.State.CLOSING, 1, 2, 3, 4, 5);
@@ -84,7 +95,7 @@ public class TestClosingContainerHandler {
@Test
public void testNonClosingRatisContainerReturnsFalse() {
ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(
- ratisReplicationConfig, 1, CLOSED);
+ RATIS_REPLICATION_CONFIG, 1, CLOSED);
Set<ContainerReplica> containerReplicas = ReplicationTestUtil
.createReplicas(containerInfo.containerID(),
ContainerReplicaProto.State.CLOSING, 0, 0, 0);
@@ -107,7 +118,7 @@ public class TestClosingContainerHandler {
@Test
public void testUnhealthyReplicaIsNotClosed() {
ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(
- ecReplicationConfig, 1, CLOSING);
+ EC_REPLICATION_CONFIG, 1, CLOSING);
Set<ContainerReplica> containerReplicas = ReplicationTestUtil
.createReplicas(containerInfo.containerID(),
ContainerReplicaProto.State.UNHEALTHY, 1, 2, 3, 4);
@@ -130,7 +141,7 @@ public class TestClosingContainerHandler {
@Test
public void testUnhealthyRatisReplicaIsNotClosed() {
ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(
- ratisReplicationConfig, 1, CLOSING);
+ RATIS_REPLICATION_CONFIG, 1, CLOSING);
Set<ContainerReplica> containerReplicas = ReplicationTestUtil
.createReplicas(containerInfo.containerID(),
ContainerReplicaProto.State.UNHEALTHY, 0, 0);
@@ -153,51 +164,55 @@ public class TestClosingContainerHandler {
/**
* Close commands should be sent for Open or Closing replicas.
*/
- @Test
- public void testOpenOrClosingReplicasAreClosed() {
+ @ParameterizedTest
+ @MethodSource("replicationConfigs")
+ public void testOpenOrClosingReplicasAreClosed(ReplicationConfig repConfig) {
ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(
- ecReplicationConfig, 1, CLOSING);
- Set<ContainerReplica> containerReplicas = ReplicationTestUtil
- .createReplicas(containerInfo.containerID(),
- ContainerReplicaProto.State.CLOSING, 1, 2);
- containerReplicas.add(ReplicationTestUtil.createContainerReplica(
- containerInfo.containerID(), 3,
- HddsProtos.NodeOperationalState.IN_SERVICE,
- ContainerReplicaProto.State.OPEN));
+ repConfig, 1, CLOSING);
+
+ final int replicas = repConfig.getRequiredNodes();
+ final int closing = replicas / 2;
+ final boolean force = repConfig.getReplicationType() != RATIS;
+
+ Set<ContainerReplica> containerReplicas = new HashSet<>();
+
+ // Add CLOSING container replicas.
+ // For EC, replica index will be in [1, closing].
+ for (int i = 1; i <= closing; i++) {
+ containerReplicas.add(ReplicationTestUtil.createContainerReplica(
+ containerInfo.containerID(),
+ repConfig.getReplicationType() == EC ? i : 0,
+ HddsProtos.NodeOperationalState.IN_SERVICE,
+ ContainerReplicaProto.State.CLOSING));
+ }
+
+ // Add OPEN container replicas.
+ // For EC, replica index will be in [closing + 1, replicas].
+ for (int i = closing + 1; i <= replicas; i++) {
+ containerReplicas.add(ReplicationTestUtil.createContainerReplica(
+ containerInfo.containerID(),
+ repConfig.getReplicationType() == EC ? i : 0,
+ HddsProtos.NodeOperationalState.IN_SERVICE,
+ ContainerReplicaProto.State.OPEN));
+ }
ContainerCheckRequest request = new ContainerCheckRequest.Builder()
- .setPendingOps(Collections.EMPTY_LIST)
+ .setPendingOps(Collections.emptyList())
.setReport(new ReplicationManagerReport())
.setContainerInfo(containerInfo)
.setContainerReplicas(containerReplicas)
.build();
- assertAndVerify(request, true, 3);
- }
-
- @Test
- public void testOpenOrClosingRatisReplicasAreClosed() {
- ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(
- ratisReplicationConfig, 1, CLOSING);
- Set<ContainerReplica> containerReplicas = ReplicationTestUtil
- .createReplicas(containerInfo.containerID(),
- ContainerReplicaProto.State.CLOSING, 0, 0);
- containerReplicas.add(ReplicationTestUtil.createContainerReplica(
- containerInfo.containerID(), 0,
- HddsProtos.NodeOperationalState.IN_SERVICE,
- ContainerReplicaProto.State.OPEN));
-
- ContainerCheckRequest request = new ContainerCheckRequest.Builder()
- .setPendingOps(Collections.EMPTY_LIST)
- .setReport(new ReplicationManagerReport())
- .setContainerInfo(containerInfo)
- .setContainerReplicas(containerReplicas)
- .build();
-
- assertAndVerify(request, true, 3);
+ ArgumentCaptor<Boolean> forceCaptor =
+ ArgumentCaptor.forClass(Boolean.class);
+ Assertions.assertTrue(closingContainerHandler.handle(request));
+ Mockito.verify(replicationManager, Mockito.times(replicas))
+ .sendCloseContainerReplicaCommand(Mockito.any(ContainerInfo.class),
+ Mockito.any(DatanodeDetails.class), forceCaptor.capture());
+ forceCaptor.getAllValues()
+ .forEach(f -> Assertions.assertEquals(force, f));
}
-
private void assertAndVerify(ContainerCheckRequest request,
boolean assertion, int times) {
Assertions.assertEquals(assertion, closingContainerHandler.handle(request));
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org