This is an automated email from the ASF dual-hosted git repository.
ChenSammi 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 3fa99ad4718 HDDS-16227. Aborted LifecycleActionTask should not mark
scan state completed (#11069)
3fa99ad4718 is described below
commit 3fa99ad471879166afbc02a2bc56a703fc33c2ed
Author: Sammi Chen <[email protected]>
AuthorDate: Thu Aug 20 18:22:27 2026 +0800
HDDS-16227. Aborted LifecycleActionTask should not mark scan state
completed (#11069)
---
.../ozone/om/service/KeyLifecycleService.java | 27 +++++-----
.../ozone/om/service/TestKeyLifecycleService.java | 62 ++++++++++++++++++++--
2 files changed, 74 insertions(+), 15 deletions(-)
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleService.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleService.java
index 09289826594..9ff0929c3a3 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleService.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleService.java
@@ -311,6 +311,7 @@ public final class LifecycleActionTask implements
BackgroundTask {
private long lastStateSaveTime = Time.monotonicNow();
private long lastStateSaveKeyCount = 0;
+ private boolean scanAborted;
private boolean shouldSaveState() {
if ((Time.monotonicNow() - lastStateSaveTime) > stateSaveIntervalMs ||
@@ -436,9 +437,10 @@ public BackgroundTaskResult call() {
evaluateBucket(bucket, keyTable, expirationRules,
expiredKeyList, scanStateBuilder);
}
+ boolean scanFinished = !scanAborted;
if (expiredKeyList.isEmpty() && expiredDirList.isEmpty()) {
LOG.info("No expired keys/dirs found/remained for bucket {}",
bucketKey);
- sendSaveScanStateRequest(scanStateBuilder, true);
+ sendSaveScanStateRequest(scanStateBuilder, scanFinished);
} else {
LOG.info("{} expired keys and {} expired dirs found and remained
for bucket {}",
expiredKeyList.size(), expiredDirList.size(), bucketKey);
@@ -447,11 +449,11 @@ public BackgroundTaskResult call() {
// OBS bucket doesn't support trash.
if (bucket.getBucketLayout() == OBJECT_STORE) {
sendDeleteKeysRequestAndClearList(bucket.getVolumeName(),
bucket.getBucketName(),
- bucket.getOwner(), expiredKeyList, false,
scanStateBuilder, true);
+ bucket.getOwner(), expiredKeyList, false,
scanStateBuilder, scanFinished);
} else {
// handle keys first, then directories
- handleAndClearFullList(bucket, expiredKeyList, false,
scanStateBuilder, true);
- handleAndClearFullList(bucket, expiredDirList, true,
scanStateBuilder, true);
+ handleAndClearFullList(bucket, expiredKeyList, false,
scanStateBuilder, scanFinished);
+ handleAndClearFullList(bucket, expiredDirList, true,
scanStateBuilder, scanFinished);
}
}
}
@@ -677,10 +679,9 @@ private void evaluateKeyAndDirTable(OmBucketInfo bucket,
long volumeObjId, Table
HashSet<Long> deletedDirSet = new HashSet<>();
while (!stack.isEmpty()) {
if (!shouldRun()) {
- LOG.info("LifecycleActionTask for bucket {} stopping. " +
- "Service enabled: {}, suspended: {}, leader ready: {}",
- bucketName, isServiceEnabled.get(), suspended.get(),
- getOzoneManager() != null ? getOzoneManager().isLeaderReady() :
"N/A");
+ scanAborted = true;
+ LOG.info("KeyLifecycleService is suspended, disabled, or leader not
ready. Stopping task for bucket {}.",
+ bucketName);
return;
}
@@ -1028,8 +1029,9 @@ private void evaluateBucket(OmBucketInfo bucketInfo,
while (keyTblItr.hasNext()) {
if (!shouldRun()) {
- LOG.info("KeyLifecycleService is suspended or disabled. " +
- "Stopping LifecycleActionTask for bucket {}.", bucketName);
+ scanAborted = true;
+ LOG.info("KeyLifecycleService is suspended, disabled, or leader
not ready. Stopping task for bucket {}.",
+ bucketName);
return;
}
if (shouldSaveState()) {
@@ -1093,8 +1095,9 @@ private void processMultipartUploads(OmBucketInfo
bucketInfo, List<OmLCRule> rul
omMetadataManager.getMultipartInfoTable().iterator(bucketPrefix)) {
while (mpuIterator.hasNext()) {
if (!shouldRun()) {
- LOG.info("KeyLifecycleService is suspended or disabled. " +
- "Stopping multipart upload processing for bucket {}.",
bucketName);
+ scanAborted = true;
+ LOG.info("KeyLifecycleService is suspended, disabled, or leader
not ready. Stopping task for bucket {}.",
+ bucketName);
return;
}
Table.KeyValue<String, OmMultipartKeyInfo> entry =
mpuIterator.next();
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 3f2a93c20c5..ab725ba4800 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
@@ -32,6 +32,7 @@
import static org.apache.hadoop.ozone.OzoneConsts.ETAG;
import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
import static
org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_DELETE_BATCH_SIZE;
+import static
org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_DELETE_BATCH_SIZE_DEFAULT;
import static
org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_ENABLED;
import static
org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_INTERVAL;
import static
org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_STATE_SAVE_INTERVAL_MS;
@@ -275,6 +276,8 @@ 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));
}
@AfterAll
@@ -569,6 +572,52 @@ void testInFlightClearedWhenTaskSkipsRun() throws
Exception {
}
}
+ @ParameterizedTest
+ @MethodSource("parameters1")
+ void testAbortedScanDoesNotMarkScanComplete(BucketLayout bucketLayout,
boolean createPrefix)
+ throws Exception {
+ final String volumeName = getTestName();
+ final String bucketName = uniqueObjectName("bucket");
+ String keyPrefix = "key";
+ String rulePrefix = bucketLayout == FILE_SYSTEM_OPTIMIZED ? "" : "key";
+ int testKeyCount = 3;
+
+ keyLifecycleService.setListMaxSize(1);
+ KeyLifecycleService.setInjectors(Arrays.asList(new FaultInjectorImpl()));
+
+ List<OmKeyArgs> keyList =
+ createKeys(volumeName, bucketName, bucketLayout, testKeyCount, 1,
keyPrefix, null);
+ assertEquals(testKeyCount, keyList.size());
+
+ ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
+ ZonedDateTime date = now.plusSeconds(EXPIRE_SECONDS);
+ if (createPrefix) {
+ createLifecyclePolicy(volumeName, bucketName, bucketLayout,
rulePrefix, null, date.toString(), true);
+ } else {
+ OmLCFilter.Builder filter = getOmLCFilterBuilder(rulePrefix, null,
null);
+ createLifecyclePolicy(volumeName, bucketName, bucketLayout, null,
filter.build(), date.toString(), true);
+ }
+
+ String bucketKey = metadataManager.getBucketKey(volumeName, bucketName);
+ GenericTestUtils.waitFor(() ->
keyLifecycleService.status().getRunningBucketsList().contains(bucketKey),
+ WAIT_CHECK_INTERVAL, 10000);
+
+ GenericTestUtils.LogCapturer logCapturer =
GenericTestUtils.LogCapturer.captureLogs(
+ LoggerFactory.getLogger(KeyLifecycleService.class));
+ keyLifecycleService.suspend();
+ KeyLifecycleService.getInjector(0).resume();
+ GenericTestUtils.waitFor(() ->
keyLifecycleService.status().getRunningBucketsList().isEmpty(),
+ WAIT_CHECK_INTERVAL, 10000);
+
+ OmLifecycleScanState scanState =
metadataManager.getLifecycleScanStateTable().get(bucketKey);
+ assertNotNull(scanState);
+ assertNull(scanState.getScanEndTime(),
+ "Aborted scan must not persist scanEndTime so a new leader can
resume");
+ assertTrue(logCapturer.getOutput().contains("KeyLifecycleService is
suspended, disabled, or leader not ready"));
+ keyLifecycleService.resume();
+ deleteLifecyclePolicy(volumeName, bucketName);
+ }
+
@ParameterizedTest
@MethodSource("parameters1")
void testBucketScanWithScanEndTime(BucketLayout bucketLayout, boolean
createPrefix) throws IOException,
@@ -1727,7 +1776,9 @@ void testPrefixDirectoryNotExpired(String dirName, String
prefix, int dirDepth,
KeyInfoWithVolumeContext keyInfo = getDirectory(volumeName, bucketName,
dirName);
assertFalse(keyInfo.getKeyInfo().isFile());
Thread.sleep(SERVICE_INTERVAL);
- assertEquals(dirDepth, getDirCount() - initialDirCount);
+
+ GenericTestUtils.waitFor(() -> dirDepth == getDirCount() -
initialDirCount,
+ WAIT_CHECK_INTERVAL, 5000);
assertEquals(0, getDeletedDirectoryCount() - initialDeletedDirCount);
GenericTestUtils.LogCapturer log =
@@ -3435,9 +3486,14 @@ private long getDeletedDirectoryCount() {
}
}
- private long getDirCount() throws IOException {
+ private long getDirCount() {
final Table<String, OmDirectoryInfo> table =
metadataManager.getDirectoryTable();
- return metadataManager.countRowsInTable(table);
+ try {
+ return metadataManager.countRowsInTable(table);
+ } catch (IOException e) {
+ fail("Failed to count directories " + e.getMessage());
+ return -1;
+ }
}
private long getKeyCount(BucketLayout layout) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
