This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-7593
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/HDDS-7593 by this push:
new b532f81e00 HDDS-10077. Add hsync metadata to hsync'ed keys in
OpenKeyTable as well (#6046)
b532f81e00 is described below
commit b532f81e00c1df1b23562ddcfff32671949df453
Author: Siyao Meng <[email protected]>
AuthorDate: Mon Jan 29 19:47:17 2024 -0800
HDDS-10077. Add hsync metadata to hsync'ed keys in OpenKeyTable as well
(#6046)
---
.../java/org/apache/hadoop/fs/ozone/TestHSync.java | 107 ++++++++++++++++++++-
.../apache/hadoop/fs/ozone/TestLeaseRecovery.java | 12 ++-
.../hadoop/ozone/om/OmMetadataManagerImpl.java | 34 +------
.../om/request/file/OMRecoverLeaseRequest.java | 4 +-
.../ozone/om/request/key/OMKeyCommitRequest.java | 39 +++++---
.../om/request/key/OMKeyCommitRequestWithFSO.java | 38 +++++---
.../ozone/om/request/util/OmKeyHSyncUtil.java | 56 +++++++++++
.../ozone/om/response/key/OMKeyCommitResponse.java | 14 ++-
.../response/key/OMKeyCommitResponseWithFSO.java | 10 +-
.../om/request/key/TestOMKeyCommitRequest.java | 12 +--
.../om/response/key/TestOMKeyCommitResponse.java | 8 +-
.../key/TestOMKeyCommitResponseWithFSO.java | 4 +-
12 files changed, 261 insertions(+), 77 deletions(-)
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestHSync.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestHSync.java
index c7a5c23164..8d7439604e 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestHSync.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestHSync.java
@@ -26,6 +26,7 @@ import java.security.PrivilegedExceptionAction;
import java.util.HashMap;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.lang3.RandomStringUtils;
@@ -66,6 +67,7 @@ import org.apache.hadoop.ozone.om.OzoneManager;
import org.apache.hadoop.ozone.om.exceptions.OMException;
import org.apache.hadoop.ozone.om.helpers.BucketLayout;
+import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
import org.apache.hadoop.ozone.om.request.key.OMKeyCommitRequest;
import org.apache.hadoop.ozone.om.request.key.OMKeyCommitRequestWithFSO;
@@ -75,13 +77,17 @@ import org.apache.hadoop.util.Time;
import org.apache.ozone.test.GenericTestUtils;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.MethodOrderer.OrderAnnotation;
+import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.event.Level;
import static java.nio.charset.StandardCharsets.UTF_8;
+import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_INTERVAL;
import static org.apache.hadoop.ozone.OzoneConsts.OZONE_OFS_URI_SCHEME;
import static org.apache.hadoop.ozone.OzoneConsts.OZONE_ROOT;
import static org.apache.hadoop.ozone.OzoneConsts.OZONE_URI_DELIMITER;
@@ -89,6 +95,7 @@ import static
org.apache.hadoop.ozone.OzoneConsts.OZONE_URI_SCHEME;
import static
org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_DEFAULT_BUCKET_LAYOUT;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
import static
org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_RATIS_ENABLE_KEY;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.fail;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -101,6 +108,7 @@ import static org.mockito.Mockito.when;
* Test HSync.
*/
@Timeout(value = 300)
+@TestMethodOrder(OrderAnnotation.class)
public class TestHSync {
private static final Logger LOG =
LoggerFactory.getLogger(TestHSync.class);
@@ -110,6 +118,7 @@ public class TestHSync {
private static final OzoneConfiguration CONF = new OzoneConfiguration();
private static OzoneClient client;
+ private static final BucketLayout BUCKET_LAYOUT =
BucketLayout.FILE_SYSTEM_OPTIMIZED;
@BeforeAll
public static void init() throws Exception {
@@ -117,11 +126,13 @@ public class TestHSync {
final int flushSize = 2 * chunkSize;
final int maxFlushSize = 2 * flushSize;
final int blockSize = 2 * maxFlushSize;
- final BucketLayout layout = BucketLayout.FILE_SYSTEM_OPTIMIZED;
+ final BucketLayout layout = BUCKET_LAYOUT;
CONF.setBoolean(OZONE_OM_RATIS_ENABLE_KEY, false);
CONF.set(OZONE_DEFAULT_BUCKET_LAYOUT, layout.name());
CONF.setBoolean(OzoneConfigKeys.OZONE_FS_HSYNC_ENABLED, true);
+ // Reduce KeyDeletingService interval
+ CONF.setTimeDuration(OZONE_BLOCK_DELETING_SERVICE_INTERVAL, 100,
TimeUnit.MILLISECONDS);
cluster = MiniOzoneCluster.newBuilder(CONF)
.setNumDatanodes(5)
.setTotalPipelineNumLimit(10)
@@ -154,6 +165,83 @@ public class TestHSync {
}
}
+ @Test
+ // Making this the first test to be run to avoid db key composition headaches
+ @Order(1)
+ public void testKeyMetadata() throws Exception {
+ // Tests key metadata behavior upon create(), hsync() and close():
+ // 1. When a key is create()'d, neither OpenKeyTable nor KeyTable entry
shall have hsync metadata.
+ // 2. When the key is hsync()'ed, both OpenKeyTable and KeyTable shall
have hsync metadata.
+ // 3. When the key is hsync()'ed again, both OpenKeyTable and KeyTable
shall have hsync metadata.
+ // 4. When the key is close()'d, KeyTable entry shall not have hsync
metadata. Key shall not exist in OpenKeyTable.
+
+ final String rootPath = String.format("%s://%s/",
+ OZONE_OFS_URI_SCHEME, CONF.get(OZONE_OM_ADDRESS_KEY));
+ CONF.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, rootPath);
+
+ final String dir = OZONE_ROOT + bucket.getVolumeName()
+ + OZONE_URI_DELIMITER + bucket.getName();
+ final String keyName = "file-test-key-metadata";
+ final Path file = new Path(dir, keyName);
+
+ OMMetadataManager omMetadataManager =
+ cluster.getOzoneManager().getMetadataManager();
+
+ // Expect empty OpenKeyTable and KeyTable before key creation
+ Table<String, OmKeyInfo> openKeyTable =
omMetadataManager.getOpenKeyTable(BUCKET_LAYOUT);
+ assertTrue(openKeyTable.isEmpty());
+ Table<String, OmKeyInfo> keyTable =
omMetadataManager.getKeyTable(BUCKET_LAYOUT);
+ assertTrue(keyTable.isEmpty());
+
+ try (FileSystem fs = FileSystem.get(CONF)) {
+ try (FSDataOutputStream os = fs.create(file, true)) {
+ // Wait for double buffer flush to avoid flakiness because RDB
iterator bypasses table cache
+ cluster.getOzoneManager().awaitDoubleBufferFlush();
+ // OpenKeyTable key should NOT have HSYNC_CLIENT_ID
+ OmKeyInfo keyInfo = getFirstKeyInTable(keyName, openKeyTable);
+
assertFalse(keyInfo.getMetadata().containsKey(OzoneConsts.HSYNC_CLIENT_ID));
+ // KeyTable should still be empty
+ assertTrue(keyTable.isEmpty());
+
+ os.hsync();
+ cluster.getOzoneManager().awaitDoubleBufferFlush();
+ // OpenKeyTable key should have HSYNC_CLIENT_ID now
+ keyInfo = getFirstKeyInTable(keyName, openKeyTable);
+
assertTrue(keyInfo.getMetadata().containsKey(OzoneConsts.HSYNC_CLIENT_ID));
+ // KeyTable key should be there and have HSYNC_CLIENT_ID
+ keyInfo = getFirstKeyInTable(keyName, keyTable);
+
assertTrue(keyInfo.getMetadata().containsKey(OzoneConsts.HSYNC_CLIENT_ID));
+
+ // hsync again, metadata should not change
+ os.hsync();
+ cluster.getOzoneManager().awaitDoubleBufferFlush();
+ keyInfo = getFirstKeyInTable(keyName, openKeyTable);
+
assertTrue(keyInfo.getMetadata().containsKey(OzoneConsts.HSYNC_CLIENT_ID));
+ keyInfo = getFirstKeyInTable(keyName, keyTable);
+
assertTrue(keyInfo.getMetadata().containsKey(OzoneConsts.HSYNC_CLIENT_ID));
+ }
+ // key is closed, OpenKeyTable should be empty
+ cluster.getOzoneManager().awaitDoubleBufferFlush();
+ assertTrue(openKeyTable.isEmpty());
+ // KeyTable should have the key. But the key shouldn't have metadata
HSYNC_CLIENT_ID anymore
+ OmKeyInfo keyInfo = getFirstKeyInTable(keyName, keyTable);
+
assertFalse(keyInfo.getMetadata().containsKey(OzoneConsts.HSYNC_CLIENT_ID));
+
+ // Clean up
+ assertTrue(fs.delete(file, false));
+ // Wait for KeyDeletingService to finish to avoid interfering other tests
+ Table<String, RepeatedOmKeyInfo> deletedTable =
omMetadataManager.getDeletedTable();
+ GenericTestUtils.waitFor(
+ () -> {
+ try {
+ return deletedTable.isEmpty();
+ } catch (IOException e) {
+ return false;
+ }
+ }, 250, 10000);
+ }
+ }
+
@Test
public void testKeyHSyncThenClose() throws Exception {
// Check that deletedTable should not have keys with the same block as in
@@ -597,6 +685,23 @@ public class TestHSync {
}
}
+ /**
+ * Helper method to check and get the first key in the OpenKeyTable.
+ * @param keyName expect key name to contain this string
+ * @param openKeyTable Table<String, OmKeyInfo>
+ * @return OmKeyInfo
+ */
+ private OmKeyInfo getFirstKeyInTable(String keyName, Table<String,
OmKeyInfo> openKeyTable) throws IOException {
+ try (TableIterator<String, ? extends Table.KeyValue<String, OmKeyInfo>> it
= openKeyTable.iterator()) {
+ assertTrue(it.hasNext());
+ Table.KeyValue<String, OmKeyInfo> kv = it.next();
+ String dbOpenKey = kv.getKey();
+ assertNotNull(dbOpenKey);
+ assertTrue(dbOpenKey.contains(keyName));
+ return kv.getValue();
+ }
+ }
+
private void testEncryptedStreamCapabilities(boolean isEC) throws
IOException,
GeneralSecurityException {
KeyOutputStream kos;
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestLeaseRecovery.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestLeaseRecovery.java
index a40c7f5275..93775f4013 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestLeaseRecovery.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestLeaseRecovery.java
@@ -92,10 +92,14 @@ public class TestLeaseRecovery {
* is no longer open in OM. This is currently expected (see HDDS-9358).
*/
public static void closeIgnoringKeyNotFound(OutputStream stream) {
+ closeIgnoringOMException(stream, OMException.ResultCodes.KEY_NOT_FOUND);
+ }
+
+ public static void closeIgnoringOMException(OutputStream stream,
OMException.ResultCodes expectedResultCode) {
try {
stream.close();
} catch (IOException e) {
- assertEquals(OMException.ResultCodes.KEY_NOT_FOUND,
((OMException)e).getResult());
+ assertEquals(expectedResultCode, ((OMException)e).getResult());
}
}
@@ -327,7 +331,11 @@ public class TestLeaseRecovery {
// Since all DNs are out, then the length in OM keyInfo will be used as
the final file length
assertEquals(dataSize, fileStatus.getLen());
} finally {
- closeIgnoringKeyNotFound(stream);
+ if (!forceRecovery) {
+ closeIgnoringOMException(stream,
OMException.ResultCodes.KEY_UNDER_LEASE_RECOVERY);
+ } else {
+ closeIgnoringKeyNotFound(stream);
+ }
KeyValueHandler.setInjector(null);
}
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
index 86ba834bc9..cd0382f599 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
@@ -104,7 +104,6 @@ import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import org.apache.commons.lang3.StringUtils;
import static org.apache.hadoop.ozone.OzoneConsts.DB_TRANSIENT_MARKER;
-import static org.apache.hadoop.ozone.OzoneConsts.HSYNC_CLIENT_ID;
import static org.apache.hadoop.ozone.OzoneConsts.OM_DB_NAME;
import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
import static
org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_FS_SNAPSHOT_MAX_LIMIT;
@@ -1207,10 +1206,8 @@ public class OmMetadataManagerImpl implements
OMMetadataManager,
// listKeys do. But that complicates the iteration logic by quite a bit.
// And if we do that, we need to refactor listKeys as well to dedup.
- final Table<String, OmKeyInfo> okTable, kTable;
+ final Table<String, OmKeyInfo> okTable;
okTable = getOpenKeyTable(bucketLayout);
- // keyTable required to check key hsync metadata. TODO: HDDS-10077
- kTable = getKeyTable(bucketLayout);
// No lock required since table iterator creates a "snapshot"
try (TableIterator<String, ? extends KeyValue<String, OmKeyInfo>>
@@ -1228,13 +1225,7 @@ public class OmMetadataManagerImpl implements
OMMetadataManager,
String dbKey = kv.getKey();
long clientID = OMMetadataManager.getClientIDFromOpenKeyDBKey(dbKey);
OmKeyInfo omKeyInfo = kv.getValue();
-
- // Trim client ID to get the keyTable dbKey
- int lastSlashIdx = dbKey.lastIndexOf(OM_KEY_PREFIX);
- String ktDbKey = dbKey.substring(0, lastSlashIdx);
- // Check whether the key has been hsync'ed by checking keyTable
- checkAndUpdateKeyHsyncStatus(omKeyInfo, ktDbKey, kTable);
-
+ // Note with HDDS-10077, there is no need to check KeyTable for
hsync metadata
openKeySessionList.add(
new OpenKeySession(clientID, omKeyInfo,
omKeyInfo.getLatestVersionLocations().getVersion()));
@@ -1261,23 +1252,6 @@ public class OmMetadataManagerImpl implements
OMMetadataManager,
openKeySessionList);
}
- /**
- * Check and update OmKeyInfo from OpenKeyTable with hsync status in
KeyTable.
- */
- private void checkAndUpdateKeyHsyncStatus(OmKeyInfo omKeyInfo,
- String dbKey,
- Table<String, OmKeyInfo> kTable)
- throws IOException {
- OmKeyInfo ktOmKeyInfo = kTable.get(dbKey);
- if (ktOmKeyInfo != null) {
- // The same key in OpenKeyTable also exists in KeyTable, indicating
- // the key has been hsync'ed
- String hsyncClientId = ktOmKeyInfo.getMetadata().get(HSYNC_CLIENT_ID);
- // Append HSYNC_CLIENT_ID to OmKeyInfo to be returned to the client
- omKeyInfo.getMetadata().put(HSYNC_CLIENT_ID, hsyncClientId);
- }
- }
-
@Override
public ListKeysResult listKeys(String volumeName, String bucketName,
String startKey, String keyPrefix, int
maxKeys)
@@ -1878,8 +1852,7 @@ public class OmMetadataManagerImpl implements
OMMetadataManager,
final String clientIdString
= dbOpenKeyName.substring(lastPrefix + 1);
- final OmKeyInfo info = kt.get(dbKeyName);
- final boolean isHsync = java.util.Optional.ofNullable(info)
+ final boolean isHsync = java.util.Optional.of(openKeyInfo)
.map(WithMetadata::getMetadata)
.map(meta -> meta.get(OzoneConsts.HSYNC_CLIENT_ID))
.filter(id -> id.equals(clientIdString))
@@ -1892,6 +1865,7 @@ public class OmMetadataManagerImpl implements
OMMetadataManager,
} else if (isHsync && openKeyInfo.getModificationTime() <=
expiredLeaseTimestamp &&
!openKeyInfo.getMetadata().containsKey(OzoneConsts.LEASE_RECOVERY)) {
// add hsync'ed keys
+ final OmKeyInfo info = kt.get(dbKeyName);
final KeyArgs.Builder keyArgs = KeyArgs.newBuilder()
.setVolumeName(info.getVolumeName())
.setBucketName(info.getBucketName())
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMRecoverLeaseRequest.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMRecoverLeaseRequest.java
index 6d997ed272..798fed7dcc 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMRecoverLeaseRequest.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMRecoverLeaseRequest.java
@@ -22,8 +22,6 @@ import com.google.common.base.Preconditions;
import
org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
import org.apache.hadoop.hdds.security.token.OzoneBlockTokenSecretManager;
-import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
-import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.audit.OMAction;
import org.apache.hadoop.ozone.om.OMMetadataManager;
@@ -238,7 +236,7 @@ public class OMRecoverLeaseRequest extends OMKeyRequest {
openKeyInfo.setModificationTime(Time.now());
// add to cache.
omMetadataManager.getOpenKeyTable(getBucketLayout()).addCacheEntry(
- new CacheKey<>(dbOpenFileKey), CacheValue.get(transactionLogIndex,
openKeyInfo));
+ dbOpenFileKey, openKeyInfo, transactionLogIndex);
}
// override key name with normalizedKeyPath
keyInfo.setKeyName(keyName);
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 ab6357a01d..154e456414 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
@@ -42,6 +42,7 @@ import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
import org.apache.hadoop.ozone.om.helpers.WithMetadata;
import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
+import org.apache.hadoop.ozone.om.request.util.OmKeyHSyncUtil;
import org.apache.hadoop.ozone.om.request.validation.RequestFeatureValidator;
import org.apache.hadoop.ozone.om.request.validation.RequestProcessingPhase;
import org.apache.hadoop.ozone.om.request.validation.ValidationCondition;
@@ -81,8 +82,7 @@ import static
org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.BUCKET_L
public class OMKeyCommitRequest extends OMKeyRequest {
@VisibleForTesting
- public static final Logger LOG =
- LoggerFactory.getLogger(OMKeyCommitRequest.class);
+ public static final Logger LOG =
LoggerFactory.getLogger(OMKeyCommitRequest.class);
public OMKeyCommitRequest(OMRequest omRequest, BucketLayout bucketLayout) {
super(omRequest, bucketLayout);
@@ -237,7 +237,7 @@ public class OMKeyCommitRequest extends OMKeyRequest {
final String clientIdString = String.valueOf(writerClientId);
if (null != keyToDelete) {
- isPreviousCommitHsync = java.util.Optional.ofNullable(keyToDelete)
+ isPreviousCommitHsync = java.util.Optional.of(keyToDelete)
.map(WithMetadata::getMetadata)
.map(meta -> meta.get(OzoneConsts.HSYNC_CLIENT_ID))
.filter(id -> id.equals(clientIdString))
@@ -260,17 +260,24 @@ public class OMKeyCommitRequest extends OMKeyRequest {
" metadata while recovery flag is not set in request",
KEY_UNDER_LEASE_RECOVERY);
}
}
- omKeyInfo.getMetadata().putAll(KeyValueUtil.getFromProtobuf(
- commitKeyArgs.getMetadataList()));
+
+ omKeyInfo.setModificationTime(commitKeyArgs.getModificationTime());
+
+ // non-null indicates it is necessary to update the open key
+ OmKeyInfo newOpenKeyInfo = null;
if (isHSync) {
- omKeyInfo.getMetadata().put(OzoneConsts.HSYNC_CLIENT_ID,
clientIdString);
- } else if (isRecovery) {
- omKeyInfo.getMetadata().remove(OzoneConsts.HSYNC_CLIENT_ID);
- omKeyInfo.getMetadata().remove(OzoneConsts.LEASE_RECOVERY);
+ if (!OmKeyHSyncUtil.isHSyncedPreviously(omKeyInfo, clientIdString,
dbOpenKey)) {
+ // Update open key as well if it is the first hsync of this key
+ omKeyInfo.getMetadata().put(OzoneConsts.HSYNC_CLIENT_ID,
clientIdString);
+ newOpenKeyInfo = omKeyInfo.copyObject();
+ }
}
+
+ omKeyInfo.getMetadata().putAll(KeyValueUtil.getFromProtobuf(
+ commitKeyArgs.getMetadataList()));
omKeyInfo.setDataSize(commitKeyArgs.getDataSize());
- omKeyInfo.setModificationTime(commitKeyArgs.getModificationTime());
+
// Update the block length for each block, return the allocated but
// uncommitted blocks
List<OmKeyLocationInfo> uncommitted =
@@ -337,6 +344,16 @@ public class OMKeyCommitRequest extends OMKeyRequest {
// So that this key can't be committed again.
omMetadataManager.getOpenKeyTable(getBucketLayout()).addCacheEntry(
dbOpenKey, trxnLogIndex);
+
+ // Prevent hsync metadata from getting committed to the final key
+ omKeyInfo.getMetadata().remove(OzoneConsts.HSYNC_CLIENT_ID);
+ if (isRecovery) {
+ omKeyInfo.getMetadata().remove(OzoneConsts.LEASE_RECOVERY);
+ }
+ } else if (newOpenKeyInfo != null) {
+ // isHSync is true and newOpenKeyInfo is set, update OpenKeyTable
+ omMetadataManager.getOpenKeyTable(getBucketLayout()).addCacheEntry(
+ dbOpenKey, newOpenKeyInfo, trxnLogIndex);
}
omMetadataManager.getKeyTable(getBucketLayout()).addCacheEntry(
@@ -346,7 +363,7 @@ public class OMKeyCommitRequest extends OMKeyRequest {
omClientResponse = new OMKeyCommitResponse(omResponse.build(),
omKeyInfo, dbOzoneKey, dbOpenKey, omBucketInfo.copyObject(),
- oldKeyVersionsToDeleteMap, isHSync);
+ oldKeyVersionsToDeleteMap, isHSync, newOpenKeyInfo);
result = Result.SUCCESS;
} catch (IOException | InvalidPathException ex) {
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 42bee999d8..f6f8f8b9cb 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
@@ -22,6 +22,7 @@ import java.nio.file.InvalidPathException;
import java.util.HashMap;
import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.ozone.om.request.util.OmKeyHSyncUtil;
import org.apache.ratis.server.protocol.TermIndex;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.audit.AuditLogger;
@@ -171,17 +172,23 @@ public class OMKeyCommitRequestWithFSO extends
OMKeyCommitRequest {
}
}
- omKeyInfo.getMetadata().putAll(KeyValueUtil.getFromProtobuf(
- commitKeyArgs.getMetadataList()));
+ omKeyInfo.setModificationTime(commitKeyArgs.getModificationTime());
+
+ final String clientIdString = String.valueOf(writerClientId);
+ // non-null indicates it is necessary to update the open key
+ OmKeyInfo newOpenKeyInfo = null;
+
if (isHSync) {
- omKeyInfo.getMetadata().put(OzoneConsts.HSYNC_CLIENT_ID,
String.valueOf(writerClientId));
- } else if (isRecovery) {
- omKeyInfo.getMetadata().remove(OzoneConsts.HSYNC_CLIENT_ID);
- omKeyInfo.getMetadata().remove(OzoneConsts.LEASE_RECOVERY);
+ if (!OmKeyHSyncUtil.isHSyncedPreviously(omKeyInfo, clientIdString,
dbOpenFileKey)) {
+ // Update open key as well if it is the first hsync of this key
+ omKeyInfo.getMetadata().put(OzoneConsts.HSYNC_CLIENT_ID,
clientIdString);
+ newOpenKeyInfo = omKeyInfo.copyObject();
+ }
}
+ omKeyInfo.getMetadata().putAll(KeyValueUtil.getFromProtobuf(
+ commitKeyArgs.getMetadataList()));
omKeyInfo.setDataSize(commitKeyArgs.getDataSize());
- omKeyInfo.setModificationTime(commitKeyArgs.getModificationTime());
List<OmKeyLocationInfo> uncommitted =
omKeyInfo.updateLocationInfoList(locationInfoList, false);
@@ -196,8 +203,7 @@ public class OMKeyCommitRequestWithFSO extends
OMKeyCommitRequest {
boolean isPreviousCommitHsync = false;
Map<String, RepeatedOmKeyInfo> oldKeyVersionsToDeleteMap = null;
if (null != keyToDelete) {
- final String clientIdString = String.valueOf(writerClientId);
- isPreviousCommitHsync = java.util.Optional.ofNullable(keyToDelete)
+ isPreviousCommitHsync = java.util.Optional.of(keyToDelete)
.map(WithMetadata::getMetadata)
.map(meta -> meta.get(OzoneConsts.HSYNC_CLIENT_ID))
.filter(id -> id.equals(clientIdString))
@@ -267,6 +273,16 @@ public class OMKeyCommitRequestWithFSO extends
OMKeyCommitRequest {
// So that this key can't be committed again.
OMFileRequest.addOpenFileTableCacheEntry(omMetadataManager,
dbOpenFileKey, null, fileName, trxnLogIndex);
+
+ // Prevent hsync metadata from getting committed to the final key
+ omKeyInfo.getMetadata().remove(OzoneConsts.HSYNC_CLIENT_ID);
+ if (isRecovery) {
+ omKeyInfo.getMetadata().remove(OzoneConsts.LEASE_RECOVERY);
+ }
+ } else if (newOpenKeyInfo != null) {
+ // isHSync is true and newOpenKeyInfo is set, update OpenKeyTable
+ OMFileRequest.addOpenFileTableCacheEntry(omMetadataManager,
+ dbOpenFileKey, newOpenKeyInfo, fileName, trxnLogIndex);
}
OMFileRequest.addFileTableCacheEntry(omMetadataManager, dbFileKey,
@@ -275,8 +291,8 @@ public class OMKeyCommitRequestWithFSO extends
OMKeyCommitRequest {
omBucketInfo.incrUsedBytes(correctedSpace);
omClientResponse = new OMKeyCommitResponseWithFSO(omResponse.build(),
- omKeyInfo, dbFileKey, dbOpenFileKey, omBucketInfo.copyObject(),
- oldKeyVersionsToDeleteMap, volumeId, isHSync);
+ omKeyInfo, dbFileKey, dbOpenFileKey, omBucketInfo.copyObject(),
+ oldKeyVersionsToDeleteMap, volumeId, isHSync, newOpenKeyInfo);
result = Result.SUCCESS;
} catch (IOException | InvalidPathException ex) {
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/util/OmKeyHSyncUtil.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/util/OmKeyHSyncUtil.java
new file mode 100644
index 0000000000..e32b2e2177
--- /dev/null
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/util/OmKeyHSyncUtil.java
@@ -0,0 +1,56 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.ozone.om.request.util;
+
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Helper methods related to OM key HSync.
+ */
+public final class OmKeyHSyncUtil {
+
+ public static final Logger LOG =
LoggerFactory.getLogger(OmKeyHSyncUtil.class);
+
+ private OmKeyHSyncUtil() {
+ }
+
+ /**
+ * Returns true if the key has been hsync'ed before (has metadata
HSYNC_CLIENT_ID).
+ * @param omKeyInfo OmKeyInfo
+ * @param clientIdString Client ID String
+ * @param dbOpenKey dbOpenKey
+ */
+ public static boolean isHSyncedPreviously(OmKeyInfo omKeyInfo, String
clientIdString, String dbOpenKey) {
+ // Check whether the key has been hsync'ed before
+ final String previousHsyncClientId =
omKeyInfo.getMetadata().get(OzoneConsts.HSYNC_CLIENT_ID);
+ if (previousHsyncClientId != null) {
+ if (clientIdString.equals(previousHsyncClientId)) {
+ // Same client ID, no need to update OpenKeyTable. One less DB write
+ return true;
+ } else {
+ // Sanity check. Should never enter
+ LOG.warn("Client ID '{}' currently hsync'ing key does not match
previous hsync client ID '{}'. dbOpenKey='{}'",
+ clientIdString, previousHsyncClientId, dbOpenKey);
+ }
+ }
+ return false;
+ }
+}
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 c4f90958c7..0de6d27eb5 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
@@ -49,15 +49,17 @@ public class OMKeyCommitResponse extends OmKeyResponse {
private String openKeyName;
private OmBucketInfo omBucketInfo;
private Map<String, RepeatedOmKeyInfo> keyToDeleteMap;
-
private boolean isHSync;
+ private OmKeyInfo newOpenKeyInfo;
+ @SuppressWarnings("checkstyle:ParameterNumber")
public OMKeyCommitResponse(
@Nonnull OMResponse omResponse,
@Nonnull OmKeyInfo omKeyInfo, String ozoneKeyName, String openKeyName,
@Nonnull OmBucketInfo omBucketInfo,
Map<String, RepeatedOmKeyInfo> keyToDeleteMap,
- boolean isHSync) {
+ boolean isHSync,
+ OmKeyInfo newOpenKeyInfo) {
super(omResponse, omBucketInfo.getBucketLayout());
this.omKeyInfo = omKeyInfo;
this.ozoneKeyName = ozoneKeyName;
@@ -65,6 +67,7 @@ public class OMKeyCommitResponse extends OmKeyResponse {
this.omBucketInfo = omBucketInfo;
this.keyToDeleteMap = keyToDeleteMap;
this.isHSync = isHSync;
+ this.newOpenKeyInfo = newOpenKeyInfo;
}
/**
@@ -85,6 +88,9 @@ public class OMKeyCommitResponse extends OmKeyResponse {
if (!isHSync()) {
omMetadataManager.getOpenKeyTable(getBucketLayout())
.deleteWithBatch(batchOperation, openKeyName);
+ } else if (newOpenKeyInfo != null) {
+ omMetadataManager.getOpenKeyTable(getBucketLayout()).putWithBatch(
+ batchOperation, openKeyName, newOpenKeyInfo);
}
omMetadataManager.getKeyTable(getBucketLayout())
@@ -133,4 +139,8 @@ public class OMKeyCommitResponse extends OmKeyResponse {
protected boolean isHSync() {
return isHSync;
}
+
+ public OmKeyInfo getNewOpenKeyInfo() {
+ return newOpenKeyInfo;
+ }
}
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyCommitResponseWithFSO.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyCommitResponseWithFSO.java
index 6073632e55..c12c3a295d 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyCommitResponseWithFSO.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyCommitResponseWithFSO.java
@@ -47,16 +47,17 @@ public class OMKeyCommitResponseWithFSO extends
OMKeyCommitResponse {
private long volumeId;
- @SuppressWarnings("parameternumber")
+ @SuppressWarnings("checkstyle:ParameterNumber")
public OMKeyCommitResponseWithFSO(
@Nonnull OMResponse omResponse,
@Nonnull OmKeyInfo omKeyInfo,
String ozoneKeyName, String openKeyName,
@Nonnull OmBucketInfo omBucketInfo,
Map<String, RepeatedOmKeyInfo> deleteKeyMap, long volumeId,
- boolean isHSync) {
+ boolean isHSync,
+ OmKeyInfo newOpenKeyInfo) {
super(omResponse, omKeyInfo, ozoneKeyName, openKeyName,
- omBucketInfo, deleteKeyMap, isHSync);
+ omBucketInfo, deleteKeyMap, isHSync, newOpenKeyInfo);
this.volumeId = volumeId;
}
@@ -78,6 +79,9 @@ public class OMKeyCommitResponseWithFSO extends
OMKeyCommitResponse {
if (!this.isHSync()) {
omMetadataManager.getOpenKeyTable(getBucketLayout())
.deleteWithBatch(batchOperation, getOpenKeyName());
+ } else if (getNewOpenKeyInfo() != null) {
+ omMetadataManager.getOpenKeyTable(getBucketLayout()).putWithBatch(
+ batchOperation, getOpenKeyName(), getNewOpenKeyInfo());
}
OMFileRequest.addToFileTable(omMetadataManager, batchOperation,
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 f71ddbf9b8..abc5fcbb49 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
@@ -45,14 +45,10 @@ import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
import org.apache.hadoop.ozone.om.response.OMClientResponse;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
- .CommitKeyRequest;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
- .KeyArgs;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
- .KeyLocation;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos
- .OMRequest;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.CommitKeyRequest;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyArgs;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyLocation;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/key/TestOMKeyCommitResponse.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/key/TestOMKeyCommitResponse.java
index 78b2dcec7c..c26a07c97e 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/key/TestOMKeyCommitResponse.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/key/TestOMKeyCommitResponse.java
@@ -65,7 +65,7 @@ public class TestOMKeyCommitResponse extends
TestOMKeyResponse {
String ozoneKey = getOzoneKey();
OMKeyCommitResponse omKeyCommitResponse = getOmKeyCommitResponse(
- omKeyInfo, omResponse, openKey, ozoneKey, keysToDelete, false);
+ omKeyInfo, omResponse, openKey, ozoneKey, keysToDelete, false,
null);
omKeyCommitResponse.addToDBBatch(omMetadataManager, batchOperation);
@@ -94,7 +94,7 @@ public class TestOMKeyCommitResponse extends
TestOMKeyResponse {
String ozoneKey = getOzoneKey();
OMKeyCommitResponse omKeyCommitResponse = getOmKeyCommitResponse(
- omKeyInfo, omResponse, openKey, ozoneKey, null, false);
+ omKeyInfo, omResponse, openKey, ozoneKey, null, false, null);
// As during commit Key, entry will be already there in openKeyTable.
// Adding it here.
@@ -148,7 +148,7 @@ public class TestOMKeyCommitResponse extends
TestOMKeyResponse {
@NotNull
protected OMKeyCommitResponse getOmKeyCommitResponse(OmKeyInfo omKeyInfo,
OzoneManagerProtocolProtos.OMResponse omResponse, String openKey,
- String ozoneKey, RepeatedOmKeyInfo deleteKeys, Boolean isHSync)
+ String ozoneKey, RepeatedOmKeyInfo deleteKeys, Boolean isHSync,
OmKeyInfo newOpenKeyInfo)
throws IOException {
assertNotNull(omBucketInfo);
Map<String, RepeatedOmKeyInfo> deleteKeyMap = new HashMap<>();
@@ -158,6 +158,6 @@ public class TestOMKeyCommitResponse extends
TestOMKeyResponse {
new RepeatedOmKeyInfo(e)));
}
return new OMKeyCommitResponse(omResponse, omKeyInfo, ozoneKey, openKey,
- omBucketInfo, deleteKeyMap, isHSync);
+ omBucketInfo, deleteKeyMap, isHSync, newOpenKeyInfo);
}
}
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/key/TestOMKeyCommitResponseWithFSO.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/key/TestOMKeyCommitResponseWithFSO.java
index b320667803..f5838ddc0f 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/key/TestOMKeyCommitResponseWithFSO.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/key/TestOMKeyCommitResponseWithFSO.java
@@ -42,7 +42,7 @@ public class TestOMKeyCommitResponseWithFSO extends
TestOMKeyCommitResponse {
@Override
protected OMKeyCommitResponse getOmKeyCommitResponse(OmKeyInfo omKeyInfo,
OzoneManagerProtocolProtos.OMResponse omResponse, String openKey,
- String ozoneKey, RepeatedOmKeyInfo deleteKeys, Boolean isHSync)
+ String ozoneKey, RepeatedOmKeyInfo deleteKeys, Boolean isHSync,
OmKeyInfo newOpenKeyInfo)
throws IOException {
assertNotNull(omBucketInfo);
long volumeId = omMetadataManager.getVolumeId(omKeyInfo.getVolumeName());
@@ -55,7 +55,7 @@ public class TestOMKeyCommitResponseWithFSO extends
TestOMKeyCommitResponse {
new RepeatedOmKeyInfo(e)));
}
return new OMKeyCommitResponseWithFSO(omResponse, omKeyInfo, ozoneKey,
- openKey, omBucketInfo, deleteKeyMap, volumeId, isHSync);
+ openKey, omBucketInfo, deleteKeyMap, volumeId, isHSync,
newOpenKeyInfo);
}
@NotNull
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
