This is an automated email from the ASF dual-hosted git repository.
adoroszlai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 17f5be015f5 HDDS-16033. Fix intermittent failures in
TestKeyLifecycleService (#11145)
17f5be015f5 is described below
commit 17f5be015f59848858a3707ad072798fdec991c0
Author: Chi-Hsuan Huang <[email protected]>
AuthorDate: Sat Aug 29 18:47:53 2026 +0800
HDDS-16033. Fix intermittent failures in TestKeyLifecycleService (#11145)
---
.../apache/hadoop/ozone/om/FaultInjectorImpl.java | 12 +
.../ozone/om/service/TestKeyLifecycleService.java | 368 ++++++++++++++-------
2 files changed, 252 insertions(+), 128 deletions(-)
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/FaultInjectorImpl.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/FaultInjectorImpl.java
index 85296b89cbf..b1c0b76b0bd 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/FaultInjectorImpl.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/FaultInjectorImpl.java
@@ -20,6 +20,7 @@
import com.google.common.annotations.VisibleForTesting;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.utils.FaultInjector;
import org.assertj.core.api.Fail;
@@ -66,6 +67,17 @@ public void resume() throws IOException {
wait.countDown();
}
+ /** Wait until someone is paused here, without letting them through. */
+ public void awaitPaused(long timeoutMillis) throws InterruptedException {
+ Assertions.assertTrue(ready.await(timeoutMillis, TimeUnit.MILLISECONDS),
+ "Timed out waiting for the injector to be reached");
+ }
+
+ /** Unlike {@link #resume()}, does not wait for a pause first, so a
discarded injector cannot park a thread. */
+ public void release() {
+ wait.countDown();
+ }
+
@Override
public void reset() throws IOException {
init();
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyLifecycleService.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyLifecycleService.java
index 5f41fb37d75..048cafd5bc0 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyLifecycleService.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyLifecycleService.java
@@ -77,9 +77,12 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -96,6 +99,7 @@
import org.apache.hadoop.hdds.server.ServerUtils;
import org.apache.hadoop.hdds.utils.BackgroundTask;
import org.apache.hadoop.hdds.utils.BackgroundTaskQueue;
+import org.apache.hadoop.hdds.utils.FaultInjector;
import org.apache.hadoop.hdds.utils.db.DBConfigFromFile;
import org.apache.hadoop.hdds.utils.db.RocksDatabaseException;
import org.apache.hadoop.hdds.utils.db.Table;
@@ -141,7 +145,6 @@
import org.apache.hadoop.ozone.om.request.key.OMKeysDeleteRequest;
import org.apache.hadoop.ozone.om.request.util.OMMultipartUploadUtils;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
-import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.LifecycleConfiguration;
import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
import org.apache.hadoop.ozone.security.acl.OzoneObj;
import org.apache.hadoop.ozone.security.acl.OzoneObjInfo;
@@ -191,6 +194,9 @@ class TestKeyLifecycleService extends OzoneTestBase {
private static final AtomicInteger OBJECT_ID_COUNTER = new AtomicInteger();
private static final int KEY_COUNT = 2;
private static final int EXPIRE_SECONDS = 2;
+ // Date-based expiration compares the key's modificationTime against the
date, so tests that rename or
+ // update a key after it was scanned need a date that stays in the future
for the whole test.
+ private static final int LONG_EXPIRE_SECONDS = 600;
private static final int SERVICE_INTERVAL = 300;
private static final int WAIT_CHECK_INTERVAL = 50;
@@ -205,6 +211,8 @@ class TestKeyLifecycleService extends OzoneTestBase {
private ScmBlockLocationTestingClient scmBlockTestingClient;
private KeyLifecycleServiceMetrics metrics;
private long bucketObjectID;
+ private final List<FaultInjectorImpl> installedInjectors = new ArrayList<>();
+ private final Set<String> createdLifecyclePolicies = new HashSet<>();
@Parameter(0)
private long stateSaveInternal;
@@ -273,12 +281,8 @@ void setup(@TempDir File testDir) throws Exception {
}
@AfterEach
- void resume() {
- keyLifecycleService.setOzoneTrash(null);
- keyLifecycleService.setMoveToTrashEnabled(true);
- KeyLifecycleService.setInjectors(null);
-
keyLifecycleService.setListMaxSize(conf.getInt(OZONE_KEY_LIFECYCLE_SERVICE_DELETE_BATCH_SIZE,
- OZONE_KEY_LIFECYCLE_SERVICE_DELETE_BATCH_SIZE_DEFAULT));
+ void resume() throws Exception {
+ cleanUpService();
}
@AfterAll
@@ -400,6 +404,7 @@ void testPeriodicStateSave(BucketLayout bucketLayout)
throws Exception {
assertEquals(testKeyCount, keyList.size());
GenericTestUtils.waitFor(() -> getKeyCount(bucketLayout) -
initialKeyCount == testKeyCount,
WAIT_CHECK_INTERVAL, 1000);
+ awaitKeyCacheDrained(bucketLayout, volumeName, bucketName);
// Inject spy to LifecycleScanStateTable to count put operations
Field tableField =
OmMetadataManagerImpl.class.getDeclaredField("lifecycleScanStateTable");
@@ -469,6 +474,7 @@ void testBucketScanResume(BucketLayout bucketLayout,
boolean createPrefix) throw
assertEquals(testKeyCount, keyList.size());
GenericTestUtils.waitFor(() -> getKeyCount(bucketLayout) -
initialKeyCount == testKeyCount,
WAIT_CHECK_INTERVAL, 1000);
+ awaitKeyCacheDrained(bucketLayout, volumeName, bucketName);
// determine db keys
List<String> dbKeys = new ArrayList<>();
@@ -521,18 +527,16 @@ void testBucketScanResume(BucketLayout bucketLayout,
boolean createPrefix) throw
assertNotNull(state);
// resume the service
- keyLifecycleService.resume();
+ runSingleLifecycleScan(volumeName, bucketName);
- // wait for it to process
// it should skip the first 3 keys (index 0, 1, 2) since we set
lastScannedDbKey as index 2.
// So it deletes only the last 2 keys (index 3 and 4).
int expectedDeleted = 2;
GenericTestUtils.waitFor(() ->
- (getDeletedKeyCount() - initialDeletedKeyCount) >= expectedDeleted,
WAIT_CHECK_INTERVAL, 10000);
+ (getDeletedKeyCount() - initialDeletedKeyCount) == expectedDeleted,
WAIT_CHECK_INTERVAL, 10000);
// confirm it hasn't deleted all keys
assertEquals(testKeyCount - expectedDeleted, getKeyCount(bucketLayout) -
initialKeyCount);
- deleteLifecyclePolicy(volumeName, bucketName);
}
@Test
@@ -584,7 +588,8 @@ void testAbortedScanDoesNotMarkScanComplete(BucketLayout
bucketLayout, boolean c
int testKeyCount = 3;
keyLifecycleService.setListMaxSize(1);
- KeyLifecycleService.setInjectors(Arrays.asList(new FaultInjectorImpl()));
+ FaultInjectorImpl taskStart = new FaultInjectorImpl();
+ installInjectors(taskStart);
List<OmKeyArgs> keyList =
createKeys(volumeName, bucketName, bucketLayout, testKeyCount, 1,
keyPrefix, null);
@@ -600,13 +605,14 @@ void testAbortedScanDoesNotMarkScanComplete(BucketLayout
bucketLayout, boolean c
}
String bucketKey = metadataManager.getBucketKey(volumeName, bucketName);
- GenericTestUtils.waitFor(() ->
keyLifecycleService.status().getRunningBucketsList().contains(bucketKey),
- WAIT_CHECK_INTERVAL, 10000);
+ // The in-flight list is filled when the task is scheduled, so wait on
the injector instead:
+ // suspending before the task actually starts makes it skip its run and
never pause here.
+ taskStart.awaitPaused(10000);
GenericTestUtils.LogCapturer logCapturer =
GenericTestUtils.LogCapturer.captureLogs(
LoggerFactory.getLogger(KeyLifecycleService.class));
keyLifecycleService.suspend();
- KeyLifecycleService.getInjector(0).resume();
+ taskStart.release();
GenericTestUtils.waitFor(() ->
keyLifecycleService.status().getRunningBucketsList().isEmpty(),
WAIT_CHECK_INTERVAL, 10000);
@@ -640,6 +646,7 @@ void testBucketScanWithScanEndTime(BucketLayout
bucketLayout, boolean createPref
assertEquals(testKeyCount, keyList.size());
GenericTestUtils.waitFor(() -> getKeyCount(bucketLayout) -
initialKeyCount == testKeyCount,
WAIT_CHECK_INTERVAL, 1000);
+ awaitKeyCacheDrained(bucketLayout, volumeName, bucketName);
// determine db keys
List<String> dbKeys = new ArrayList<>();
@@ -847,6 +854,8 @@ void testNestedFSODirectoryScanResume(BucketLayout
bucketLayout, int maxSize) th
assertEquals(testKeyCount, keyList.size());
GenericTestUtils.waitFor(
() -> getKeyCount(bucketLayout) - initialKeyCount == testKeyCount,
WAIT_CHECK_INTERVAL, 1000);
+ awaitKeyCacheDrained(bucketLayout, volumeName, bucketName);
+ awaitDirCacheDrained(volumeName, bucketName);
// Create Lifecycle configuration
ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
@@ -854,54 +863,67 @@ void testNestedFSODirectoryScanResume(BucketLayout
bucketLayout, int maxSize) th
createLifecyclePolicy(volumeName, bucketName, bucketLayout, prefix,
null, date.toString(), true);
// Inject to cause resume case for FSO with nested directory
+ FaultInjectorImpl firstTaskStart = new FaultInjectorImpl();
+ FaultInjectorImpl firstDelete = new FaultInjectorImpl();
FaultInjectorImpl lastFaultInjector = new FaultInjectorImpl();
lastFaultInjector.setException(new IOException("Injected exception for
testing"));
- KeyLifecycleService.setInjectors(
- Arrays.asList(new FaultInjectorImpl(), new FaultInjectorImpl(),
lastFaultInjector));
+ installInjectors(firstTaskStart, firstDelete, lastFaultInjector);
// Resume the service
keyLifecycleService.resume();
- KeyLifecycleService.getInjector(0).resume();
- KeyLifecycleService.getInjector(1).resume();
+ // Returns once the aborted task is past its start, so it keeps using
the injectors below
+ firstTaskStart.awaitPaused(10000);
+ firstTaskStart.release();
+ // Hold the follow-up task at its start, otherwise it overwrites the
scan state asserted below
+ FaultInjectorImpl nextTaskStart = new FaultInjectorImpl();
+ installInjectors(nextTaskStart, firstDelete, lastFaultInjector);
+ firstDelete.awaitPaused(10000);
+ firstDelete.release();
+ // BackgroundService waits for the previous batch, so the follow-up task
reaching its start means
+ // the first task is done
+ nextTaskStart.awaitPaused(10000);
- // wait for scanState to be updated
String bucketKey = metadataManager.getBucketKey(volumeName, bucketName);
- GenericTestUtils.waitFor(() -> {
- try {
- OmLifecycleScanState scanState =
metadataManager.getLifecycleScanStateTable().get(bucketKey);
- if (bucketLayout == FILE_SYSTEM_OPTIMIZED) {
- return scanState != null && scanState.getLastScannedDir() != null
&& scanState.getLastScannedKey() != null;
- } else {
- return scanState != null && scanState.getLastScannedKey() != null;
- }
- } catch (IOException e) {
- return false;
- }
- }, WAIT_CHECK_INTERVAL, 10000);
-
- OmLifecycleScanState scanState =
metadataManager.getLifecycleScanStateTable().get(bucketKey);
- if (stateSaveInternal != -1) {
- if (maxSize == 2) {
- if (bucketLayout == FILE_SYSTEM_OPTIMIZED) {
- assertEquals("dir3/dir6/dir9", scanState.getLastScannedDir());
- assertTrue(scanState.getLastScannedKey().endsWith("key8"));
- } else {
- assertTrue(scanState.getLastScannedKey().endsWith("key1"));
- }
- } else if (maxSize == 3) {
- if (bucketLayout == FILE_SYSTEM_OPTIMIZED) {
- assertEquals("dir3/dir6/dir8", scanState.getLastScannedDir());
- assertTrue(scanState.getLastScannedKey().endsWith("key6"));
- } else {
- assertTrue(scanState.getLastScannedKey().endsWith("key2"));
+ try {
+ GenericTestUtils.waitFor(() -> {
+ try {
+ OmLifecycleScanState state =
metadataManager.getLifecycleScanStateTable().get(bucketKey);
+ if (bucketLayout == FILE_SYSTEM_OPTIMIZED) {
+ return state != null && state.getLastScannedDir() != null &&
state.getLastScannedKey() != null;
+ } else {
+ return state != null && state.getLastScannedKey() != null;
+ }
+ } catch (IOException e) {
+ return false;
}
- } else {
- if (bucketLayout == FILE_SYSTEM_OPTIMIZED) {
- assertEquals("dir1/dir5", scanState.getLastScannedDir());
- assertTrue(scanState.getLastScannedKey().endsWith("key3"));
+ }, WAIT_CHECK_INTERVAL, 10000);
+
+ OmLifecycleScanState scanState =
metadataManager.getLifecycleScanStateTable().get(bucketKey);
+ if (stateSaveInternal != -1) {
+ if (maxSize == 2) {
+ if (bucketLayout == FILE_SYSTEM_OPTIMIZED) {
+ assertEquals("dir3/dir6/dir9", scanState.getLastScannedDir());
+ assertTrue(scanState.getLastScannedKey().endsWith("key8"));
+ } else {
+ assertTrue(scanState.getLastScannedKey().endsWith("key1"));
+ }
+ } else if (maxSize == 3) {
+ if (bucketLayout == FILE_SYSTEM_OPTIMIZED) {
+ assertEquals("dir3/dir6/dir8", scanState.getLastScannedDir());
+ assertTrue(scanState.getLastScannedKey().endsWith("key6"));
+ } else {
+ assertTrue(scanState.getLastScannedKey().endsWith("key2"));
+ }
} else {
- assertTrue(scanState.getLastScannedKey().endsWith("key7"));
+ if (bucketLayout == FILE_SYSTEM_OPTIMIZED) {
+ assertEquals("dir1/dir5", scanState.getLastScannedDir());
+ assertTrue(scanState.getLastScannedKey().endsWith("key3"));
+ } else {
+ assertTrue(scanState.getLastScannedKey().endsWith("key7"));
+ }
}
}
+ } finally {
+ nextTaskStart.release();
}
GenericTestUtils.waitFor(() ->
@@ -1025,9 +1047,8 @@ void testDirectorySkippedAfterResume(String
lastScannedDir, String[] skippedDir,
GenericTestUtils.LogCapturer logCapturer =
GenericTestUtils.LogCapturer.captureLogs(
LoggerFactory.getLogger(KeyLifecycleService.class));
// Resume the service
- keyLifecycleService.resume();
+ runSingleLifecycleScan(volumeName, bucketName);
- // Wait for it to process
GenericTestUtils.waitFor(() ->
(getDeletedKeyCount() - initialDeletedKeyCount) == expectedDeleted,
WAIT_CHECK_INTERVAL, 10000);
@@ -1038,8 +1059,6 @@ void testDirectorySkippedAfterResume(String
lastScannedDir, String[] skippedDir,
d -> assertTrue(logCapturer.getOutput().contains("Skip " + d)));
logCapturer.clearOutput();
}
-
- deleteLifecyclePolicy(volumeName, bucketName);
}
@ParameterizedTest
@@ -1066,6 +1085,7 @@ void testLastScannedKeySeek(boolean keyBelongToDir)
throws Exception {
assertEquals(4, keyList.size());
GenericTestUtils.waitFor(() ->
getKeyCount(BucketLayout.FILE_SYSTEM_OPTIMIZED) - initialKeyCount == 4,
WAIT_CHECK_INTERVAL, 1000);
+ awaitKeyCacheDrained(BucketLayout.FILE_SYSTEM_OPTIMIZED, volumeName,
bucketName);
long bucketId =
metadataManager.getBucketTable().get(metadataManager.getBucketKey(volumeName,
bucketName)).getObjectID();
@@ -1110,22 +1130,24 @@ void testLastScannedKeySeek(boolean keyBelongToDir)
throws Exception {
GenericTestUtils.LogCapturer logCapturer =
GenericTestUtils.LogCapturer.captureLogs(
LoggerFactory.getLogger(KeyLifecycleService.class));
- keyLifecycleService.resume();
+ runSingleLifecycleScan(volumeName, bucketName);
- // dir1 can be fully evaluated depending on whether lastScannedKey
belong to it (no seek) or not
- // dir2 should be skipped dir2 > dir1
- int expectedDeleted = 2;
+ // dir2 is skipped since dir2 > dir1. dir1 is fully evaluated only when
lastScannedKey does not
+ // belong to it, otherwise the scan seeks past its last key and nothing
is left to expire.
+ int expectedDeleted = keyBelongToDir ? 0 : 2;
GenericTestUtils.waitFor(() ->
- (getDeletedKeyCount() - initialDeletedKeyCount) >= expectedDeleted,
WAIT_CHECK_INTERVAL, 5000);
+ (getDeletedKeyCount() - initialDeletedKeyCount) == expectedDeleted,
WAIT_CHECK_INTERVAL, 5000);
assertEquals(keyList.size() - expectedDeleted,
getKeyCount(BucketLayout.FILE_SYSTEM_OPTIMIZED) - initialKeyCount);
+ // The task drops the bucket from the in-flight list before it updates
the metrics
GenericTestUtils.waitFor(() ->
expectedDeleted == metrics.getNumKeyIterated().value() -
keyIterated, WAIT_CHECK_INTERVAL, 5000);
+ // dir2 sorts after the resumed dir1, so it is skipped in both cases
+ assertTrue(logCapturer.getOutput().contains("Skip dir2"));
if (keyBelongToDir) {
assertTrue(logCapturer.getOutput().contains("Seek to key"));
} else {
assertFalse(logCapturer.getOutput().contains("Seek to key"));
}
- deleteLifecyclePolicy(volumeName, bucketName);
}
@ParameterizedTest
@@ -1151,6 +1173,7 @@ void testBucketRootScannedDirResume(BucketLayout layout)
throws Exception {
assertEquals(3, keyList.size());
GenericTestUtils.waitFor(() -> getKeyCount(layout) - initialKeyCount ==
3,
WAIT_CHECK_INTERVAL, 1000);
+ awaitKeyCacheDrained(layout, volumeName, bucketName);
long bucketId =
metadataManager.getBucketTable().get(metadataManager.getBucketKey(volumeName,
bucketName)).getObjectID();
@@ -1184,7 +1207,7 @@ void testBucketRootScannedDirResume(BucketLayout layout)
throws Exception {
metadataManager.getLifecycleScanStateTable().addCacheEntry(new
CacheKey<>(bucketKey),
CacheValue.get(1L, scanState));
- keyLifecycleService.resume();
+ runSingleLifecycleScan(volumeName, bucketName);
// It should seek to key2. key1 is skipped, key2 is skipped too.
// So key1 and key2 are skipped, key3 is deleted.
@@ -1195,7 +1218,6 @@ void testBucketRootScannedDirResume(BucketLayout layout)
throws Exception {
assertEquals(2, getKeyCount(layout) - initialKeyCount);
GenericTestUtils.waitFor(() ->
expectedDeleted == metrics.getNumKeyIterated().value() -
keyIterated, WAIT_CHECK_INTERVAL, 5000);
- deleteLifecyclePolicy(volumeName, bucketName);
}
@ParameterizedTest
@@ -1720,8 +1742,7 @@ void testMatchedDirectoryNotDeleted(String keyPrefix,
String rulePrefix, String
KeyInfoWithVolumeContext keyInfo = getDirectory(volumeName, bucketName,
dirName);
assertFalse(keyInfo.getKeyInfo().isFile());
- KeyLifecycleService.setInjectors(
- Arrays.asList(new FaultInjectorImpl(), new FaultInjectorImpl()));
+ installInjectors(new FaultInjectorImpl(), new FaultInjectorImpl());
// create Lifecycle configuration
ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
@@ -1751,24 +1772,21 @@ void testMatchedDirectoryNotDeleted(String keyPrefix,
String rulePrefix, String
public Stream<Arguments> parameters4() {
return Stream.of(
- arguments("dir1/dir2/dir3//", "dir1/dir2/dir3/", 3, true, false),
- arguments("dir1/dir2/dir3//", "dir1/dir2/dir3/", 3, false, true),
- arguments("dir1/dir2//", "dir1/dir2/", 2, true, false),
- arguments("dir1/dir2//", "dir1/dir2/", 2, false, true),
- arguments("dir1//", "dir1/", 1, true, false),
- arguments("dir1//", "dir1/", 1, false, true)
+ arguments("dir1/dir2/dir3//", "dir1/dir2/dir3/", true, false),
+ arguments("dir1/dir2/dir3//", "dir1/dir2/dir3/", false, true),
+ arguments("dir1/dir2//", "dir1/dir2/", true, false),
+ arguments("dir1/dir2//", "dir1/dir2/", false, true),
+ arguments("dir1//", "dir1/", true, false),
+ arguments("dir1//", "dir1/", false, true)
);
}
@ParameterizedTest
@MethodSource("parameters4")
- void testPrefixDirectoryNotExpired(String dirName, String prefix, int
dirDepth, boolean createPrefix,
+ void testPrefixDirectoryNotExpired(String dirName, String prefix, boolean
createPrefix,
boolean createFilterPrefix) throws IOException, TimeoutException,
InterruptedException {
final String volumeName = getTestName();
final String bucketName = uniqueObjectName("bucket");
- long initialDeletedDirCount = getDeletedDirectoryCount();
- long initialDirCount = getDirCount();
- long initialNumDeletedDir = metrics.getNumDirDeleted().value();
// Create the directory
createVolumeAndBucket(volumeName, bucketName, FILE_SYSTEM_OPTIMIZED,
@@ -1778,10 +1796,6 @@ void testPrefixDirectoryNotExpired(String dirName,
String prefix, int dirDepth,
assertFalse(keyInfo.getKeyInfo().isFile());
Thread.sleep(SERVICE_INTERVAL);
- GenericTestUtils.waitFor(() -> dirDepth == getDirCount() -
initialDirCount,
- WAIT_CHECK_INTERVAL, 5000);
- assertEquals(0, getDeletedDirectoryCount() - initialDeletedDirCount);
-
GenericTestUtils.LogCapturer log =
GenericTestUtils.LogCapturer.captureLogs(
LoggerFactory.getLogger(KeyLifecycleService.class));
@@ -1800,8 +1814,8 @@ void testPrefixDirectoryNotExpired(String dirName, String
prefix, int dirDepth,
GenericTestUtils.waitFor(() -> log.getOutput().contains("Prefix
directory " + prefix + " doesn't get expired"),
WAIT_CHECK_INTERVAL, 10000);
- assertEquals(dirDepth, getDirCount() - initialDirCount);
- assertEquals(0, metrics.getNumDirDeleted().value() -
initialNumDeletedDir);
+ // getKeyInfo resolves the whole path, so this fails if any level of the
directory was expired
+ assertFalse(getDirectory(volumeName, bucketName,
dirName).getKeyInfo().isFile());
deleteLifecyclePolicy(volumeName, bucketName);
}
@@ -1934,6 +1948,7 @@ void testConsolidatedPrefixDirectoryNotExpired(String
dir1, String dir2, String
.build());
keyList.add(createAndCommitKey(volumeName, bucketName, dir4 + "key4",
1, null));
}
+ awaitKeyCacheDrained(FILE_SYSTEM_OPTIMIZED, volumeName, bucketName);
createLifecyclePolicy(volumeName, bucketName, FILE_SYSTEM_OPTIMIZED,
ruleList);
try {
@@ -2087,12 +2102,11 @@ void testKeyUpdatedShouldNotGetDeleted(BucketLayout
bucketLayout)
List<OmKeyArgs> keyList =
createKeys(volumeName, bucketName, bucketLayout, KEY_COUNT, 1,
keyPrefix, null);
- KeyLifecycleService.setInjectors(
- Arrays.asList(new FaultInjectorImpl(), new FaultInjectorImpl()));
+ installInjectors(new FaultInjectorImpl(), new FaultInjectorImpl());
// create Lifecycle configuration
ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
- ZonedDateTime date = now.plusSeconds(EXPIRE_SECONDS);
+ ZonedDateTime date = now.plusSeconds(LONG_EXPIRE_SECONDS);
createLifecyclePolicy(volumeName, bucketName, bucketLayout, rulePrefix,
null, date.toString(), true);
Thread.sleep(SERVICE_INTERVAL);
KeyLifecycleService.getInjector(0).resume();
@@ -2121,10 +2135,8 @@ void testKeyUpdatedShouldNotGetDeleted(BucketLayout
bucketLayout)
key.getBucketName() + "/" + key.getKeyName() + " whose updateID not
match or null";
GenericTestUtils.waitFor(() ->
requestLog.getOutput().contains(expectedString), WAIT_CHECK_INTERVAL, 10000);
- // rename will change object's modificationTime. But since expiration
action is an absolute timestamp, so
- // the renamed key will expire in next evaluation task
- GenericTestUtils.waitFor(() -> log.getOutput().contains("1 expired keys
and 0 expired dirs found"),
- WAIT_CHECK_INTERVAL, 10000);
+ // How many candidates a task reports depends on whether the in-flight
delete has landed when it
+ // starts, so assert on the outcome instead.
GenericTestUtils.waitFor(() -> getKeyCount(bucketLayout) -
initialKeyCount == 0, WAIT_CHECK_INTERVAL, 10000);
assertEquals(KEY_COUNT, getDeletedKeyCount() - initialDeletedKeyCount);
deleteLifecyclePolicy(volumeName, bucketName);
@@ -2565,8 +2577,7 @@ void testMultipleDirectoriesMatched(String keyPrefix1,
String keyPrefix2, String
KeyInfoWithVolumeContext keyInfo = getDirectory(volumeName, bucketName,
dirName);
assertFalse(keyInfo.getKeyInfo().isFile());
- KeyLifecycleService.setInjectors(
- Arrays.asList(new FaultInjectorImpl(), new FaultInjectorImpl()));
+ installInjectors(new FaultInjectorImpl(), new FaultInjectorImpl());
// create Lifecycle configuration
ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
@@ -2622,8 +2633,7 @@ void testGetLifecycleServiceStatus() throws Exception {
// Create and inject for test
createKeys(volumeName, bucketName, FILE_SYSTEM_OPTIMIZED, KEY_COUNT, 1,
prefix, null);
ZonedDateTime date =
ZonedDateTime.now(ZoneOffset.UTC).plusSeconds(EXPIRE_SECONDS);
- KeyLifecycleService.setInjectors(
- Arrays.asList(new FaultInjectorImpl(), new FaultInjectorImpl()));
+ installInjectors(new FaultInjectorImpl(), new FaultInjectorImpl());
createLifecyclePolicy(volumeName, bucketName, FILE_SYSTEM_OPTIMIZED, "",
null, date.toString(), true);
Thread.sleep(SERVICE_INTERVAL + 100);
@@ -3139,7 +3149,8 @@ void setup(@TempDir File testDir) throws Exception {
}
@AfterEach
- void resume() {
+ void resume() throws Exception {
+ cleanUpService();
}
@AfterAll
@@ -3183,7 +3194,7 @@ void testBucketDeleted(BucketLayout bucketLayout) throws
IOException, Interrupte
ZonedDateTime date = now.plusSeconds(EXPIRE_SECONDS);
FaultInjectorImpl injector = new FaultInjectorImpl();
- KeyLifecycleService.setInjectors(Arrays.asList(injector));
+ installInjectors(injector);
createLifecyclePolicy(volumeName, bucketName, bucketLayout, "", null,
date.toString(), true);
Thread.sleep(1000);
@@ -3232,12 +3243,11 @@ void testKeyDeletedOrRenamed(BucketLayout bucketLayout,
boolean deleted)
List<OmKeyArgs> keyList =
createKeys(volumeName, bucketName, bucketLayout, KEY_COUNT, 1,
keyPrefix, null);
- KeyLifecycleService.setInjectors(
- Arrays.asList(new FaultInjectorImpl(), new FaultInjectorImpl()));
+ installInjectors(new FaultInjectorImpl(), new FaultInjectorImpl());
// create Lifecycle configuration
ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
- ZonedDateTime date = now.plusSeconds(EXPIRE_SECONDS);
+ ZonedDateTime date = now.plusSeconds(LONG_EXPIRE_SECONDS);
createLifecyclePolicy(volumeName, bucketName, bucketLayout, rulePrefix,
null, date.toString(), true);
Thread.sleep(SERVICE_INTERVAL);
KeyLifecycleService.getInjector(0).resume();
@@ -3260,11 +3270,8 @@ void testKeyDeletedOrRenamed(BucketLayout bucketLayout,
boolean deleted)
String expectedString = "Received a request to delete a Key does not
exist /" + key.getVolumeName() + "/" +
key.getBucketName() + "/" + key.getKeyName();
GenericTestUtils.waitFor(() ->
requestLog.getOutput().contains(expectedString), WAIT_CHECK_INTERVAL, 10000);
- if (!deleted) {
- // Since expiration action is an absolute timestamp, so the renamed
key will expire in next evaluation task
- GenericTestUtils.waitFor(() -> log.getOutput().contains("1 expired
keys and 0 expired dirs found"),
- SERVICE_INTERVAL, 10000);
- }
+ // How many candidates a task reports depends on whether the in-flight
delete has landed when it
+ // starts, so assert on the outcome instead.
GenericTestUtils.waitFor(() -> getKeyCount(bucketLayout) -
initialKeyCount == 0, WAIT_CHECK_INTERVAL, 10000);
assertEquals(KEY_COUNT, getDeletedKeyCount() - initialDeletedKeyCount);
deleteLifecyclePolicy(volumeName, bucketName);
@@ -3384,13 +3391,7 @@ private void createLifecyclePolicy(String volume, String
bucket, BucketLayout la
}
throw e;
}
- String key = "/" + volume + "/" + bucket;
- LifecycleConfiguration lcProto = lcc.getProtobuf();
- OmLifecycleConfiguration canonicalLcc =
OmLifecycleConfiguration.getFromProtobuf(lcProto);
- canonicalLcc.valid();
- metadataManager.getLifecycleConfigurationTable().put(key, lcc);
- metadataManager.getLifecycleConfigurationTable().addCacheEntry(
- new CacheKey<>(key), CacheValue.get(1L, canonicalLcc));
+ putLifecyclePolicy(volume, bucket, lcc, true);
}
private void createLifecyclePolicy(String volume, String bucket,
BucketLayout layout, List<OmLCRule> ruleList)
@@ -3410,20 +3411,132 @@ private void createLifecyclePolicy(String volume,
String bucket, BucketLayout la
}
throw e;
}
- String key = "/" + volume + "/" + bucket;
- LifecycleConfiguration lcProto = lcc.getProtobuf();
- OmLifecycleConfiguration canonicalLcc =
OmLifecycleConfiguration.getFromProtobuf(lcProto);
+ putLifecyclePolicy(volume, bucket, lcc, false);
+ }
+
+ private void putLifecyclePolicy(String volume, String bucket,
OmLifecycleConfiguration lcc, boolean validate)
+ throws IOException {
+ String key = metadataManager.getBucketKey(volume, bucket);
+ OmLifecycleConfiguration canonicalLcc =
OmLifecycleConfiguration.getFromProtobuf(lcc.getProtobuf());
+ if (validate) {
+ canonicalLcc.valid();
+ }
metadataManager.getLifecycleConfigurationTable().put(key, lcc);
metadataManager.getLifecycleConfigurationTable().addCacheEntry(
new CacheKey<>(key), CacheValue.get(1L, canonicalLcc));
+ createdLifecyclePolicies.add(key);
}
private void deleteLifecyclePolicy(String volume, String bucket)
throws IOException {
- String key = "/" + volume + "/" + bucket;
- metadataManager.getLifecycleConfigurationTable().delete(key);
+ deleteLifecyclePolicy(metadataManager.getBucketKey(volume, bucket));
+ }
+
+ private void deleteLifecyclePolicy(String policyKey) throws IOException {
+ metadataManager.getLifecycleConfigurationTable().delete(policyKey);
metadataManager.getLifecycleConfigurationTable().addCacheEntry(
- new CacheKey<>(key), CacheValue.get(1L));
+ new CacheKey<>(policyKey), CacheValue.get(1L));
+ createdLifecyclePolicies.remove(policyKey);
+ }
+
+ // A scan reads keys still in the cache from an unordered map and skips them
in the sorted table
+ // iterator without counting them, so lastScannedKey and numKeyIterated are
unstable until they drain.
+ private void awaitKeyCacheDrained(BucketLayout layout, String volume, String
bucket)
+ throws TimeoutException, InterruptedException {
+ Table<String, OmKeyInfo> keyTable = metadataManager.getKeyTable(layout);
+ GenericTestUtils.waitFor(() -> {
+ Iterator<Map.Entry<CacheKey<String>, CacheValue<OmKeyInfo>>> cacheIter =
keyTable.cacheIterator();
+ while (cacheIter.hasNext()) {
+ OmKeyInfo key = cacheIter.next().getValue().getCacheValue();
+ if (key != null && volume.equals(key.getVolumeName()) &&
bucket.equals(key.getBucketName())) {
+ return false;
+ }
+ }
+ return true;
+ }, WAIT_CHECK_INTERVAL, 10000);
+ }
+
+ // getSubDirectory evaluates directories still in the cache and adds them
again from the table iterator,
+ // so a scan sees them twice until they drain.
+ private void awaitDirCacheDrained(String volume, String bucket)
+ throws IOException, TimeoutException, InterruptedException {
+ long volumeId =
metadataManager.getVolumeTable().get(metadataManager.getVolumeKey(volume)).getObjectID();
+ long bucketId =
metadataManager.getBucketTable().get(metadataManager.getBucketKey(volume,
bucket)).getObjectID();
+ String bucketPrefix = OM_KEY_PREFIX + volumeId + OM_KEY_PREFIX + bucketId
+ OM_KEY_PREFIX;
+ Table<String, OmDirectoryInfo> dirTable =
metadataManager.getDirectoryTable();
+ GenericTestUtils.waitFor(() -> {
+ Iterator<Map.Entry<CacheKey<String>, CacheValue<OmDirectoryInfo>>>
cacheIter = dirTable.cacheIterator();
+ while (cacheIter.hasNext()) {
+ if (cacheIter.next().getKey().getCacheKey().startsWith(bucketPrefix)) {
+ return false;
+ }
+ }
+ return true;
+ }, WAIT_CHECK_INTERVAL, 10000);
+ }
+
+ private void awaitMultipartCacheDrained(String dbKey) throws
TimeoutException, InterruptedException {
+ Table<String, OmMultipartKeyInfo> mpuTable =
metadataManager.getMultipartInfoTable();
+ GenericTestUtils.waitFor(() -> {
+ Iterator<Map.Entry<CacheKey<String>, CacheValue<OmMultipartKeyInfo>>>
cacheIter = mpuTable.cacheIterator();
+ while (cacheIter.hasNext()) {
+ if (dbKey.equals(cacheIter.next().getKey().getCacheKey())) {
+ return false;
+ }
+ }
+ return true;
+ }, WAIT_CHECK_INTERVAL, 10000);
+ }
+
+ // Tracks every injector installed, not just the current set, so a replaced
one is still released
+ private void installInjectors(FaultInjectorImpl... injectors) {
+ installedInjectors.addAll(Arrays.asList(injectors));
+ KeyLifecycleService.setInjectors(new
ArrayList<FaultInjector>(Arrays.asList(injectors)));
+ }
+
+ // A task parked in an injector blocks every following cycle, and a policy
left behind keeps its bucket
+ // scanned, so both have to be cleared for the next test to start from a
quiet service.
+ private void cleanUpService() throws Exception {
+ try {
+ // Suspending first closes the window where a cycle has already read the
policies but has not
+ // registered its task yet, which would let that task run into the next
test.
+ keyLifecycleService.suspend();
+ installedInjectors.forEach(FaultInjectorImpl::release);
+ GenericTestUtils.waitFor(() ->
keyLifecycleService.status().getRunningBucketsList().isEmpty(),
+ WAIT_CHECK_INTERVAL, 30000);
+ } finally {
+ try {
+ for (String policyKey : new ArrayList<>(createdLifecyclePolicies)) {
+ deleteLifecyclePolicy(policyKey);
+ }
+ } finally {
+ installedInjectors.clear();
+ KeyLifecycleService.setInjectors(null);
+ keyLifecycleService.setOzoneTrash(null);
+ keyLifecycleService.setMoveToTrashEnabled(true);
+
keyLifecycleService.setListMaxSize(conf.getInt(OZONE_KEY_LIFECYCLE_SERVICE_DELETE_BATCH_SIZE,
+ OZONE_KEY_LIFECYCLE_SERVICE_DELETE_BATCH_SIZE_DEFAULT));
+ keyLifecycleService.resume();
+ }
+ }
+ }
+
+ // Runs exactly one scan: without dropping the policy first, the next scan
starts SERVICE_INTERVAL
+ // later and deletes the objects the caller expects to remain.
+ private void runSingleLifecycleScan(String volume, String bucket)
+ throws IOException, TimeoutException, InterruptedException {
+ FaultInjectorImpl taskStart = new FaultInjectorImpl();
+ FaultInjectorImpl firstDelete = new FaultInjectorImpl();
+ // The third injector is only read for an injected exception, but the scan
expects it to exist
+ installInjectors(taskStart, firstDelete, new FaultInjectorImpl());
+ keyLifecycleService.resume();
+ // The in-flight list is filled at schedule time, so wait for the task to
reach the injector
+ taskStart.awaitPaused(10000);
+ deleteLifecyclePolicy(volume, bucket);
+ taskStart.release();
+ firstDelete.release();
+ GenericTestUtils.waitFor(() ->
keyLifecycleService.status().getRunningBucketsList().isEmpty(),
+ WAIT_CHECK_INTERVAL, 10000);
}
private void createVolumeAndBucket(String volumeName,
@@ -3555,16 +3668,6 @@ private long getDeletedDirectoryCount() {
}
}
- private long getDirCount() {
- final Table<String, OmDirectoryInfo> table =
metadataManager.getDirectoryTable();
- try {
- return metadataManager.countRowsInTable(table);
- } catch (IOException e) {
- fail("Failed to count directories " + e.getMessage());
- return -1;
- }
- }
-
private long getKeyCount(BucketLayout layout) {
final Table<String, OmKeyInfo> table = metadataManager.getKeyTable(layout);
try {
@@ -3597,8 +3700,12 @@ private long getMultipartUploadCount(String volumeName,
String bucketName) {
}
private void updateMultipartUploadCreationTime(String volumeName, String
bucketName,
- String keyName, String uploadId, long newCreationTime) throws
IOException {
+ String keyName, String uploadId, long newCreationTime)
+ throws IOException, TimeoutException, InterruptedException {
String dbKey = metadataManager.getMultipartKey(volumeName, bucketName,
keyName, uploadId);
+ // The create is flushed to the DB asynchronously, and that flush would
overwrite the new creation
+ // time, leaving the upload looking fresh. Wait for the entry to leave the
cache first.
+ awaitMultipartCacheDrained(dbKey);
OmMultipartKeyInfo existingInfo =
metadataManager.getMultipartInfoTable().get(dbKey);
if (existingInfo == null) {
fail("Multipart upload not found: " + dbKey);
@@ -3750,6 +3857,11 @@ void setup(@TempDir File testDir) throws Exception {
directoryDeletingService.suspend();
}
+ @AfterEach
+ void resume() throws Exception {
+ cleanUpService();
+ }
+
@AfterAll
void cleanup() {
if (om != null) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
