You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by su...@apache.org on 2025/01/19 06:28:00 UTC
(ozone) branch master updated: HDDS-11680. Enhance Recon Metrics For Improved Observability (#7517)
This is an automated email from the ASF dual-hosted git repository.
sumitagrawal 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 f90e625eea HDDS-11680. Enhance Recon Metrics For Improved Observability (#7517)
f90e625eea is described below
commit f90e625eeae6eaf7e521c631f6e2e0f2c117ebe8
Author: Abhishek Pal <43...@users.noreply.github.com>
AuthorDate: Sun Jan 19 11:57:54 2025 +0530
HDDS-11680. Enhance Recon Metrics For Improved Observability (#7517)
---
.../main/smoketest/recon/recon-taskstatus.robot | 125 ++++++++++
.../hadoop/ozone/TestOzoneConfigurationFields.java | 1 +
.../ozone/recon/TestReconWithOzoneManager.java | 4 +-
.../recon/schema/ReconTaskSchemaDefinition.java | 2 +
.../hadoop/ozone/recon/ReconControllerModule.java | 2 +
.../hadoop/ozone/recon/api/TaskStatusService.java | 5 +-
.../ozone/recon/fsck/ContainerHealthTask.java | 23 +-
.../hadoop/ozone/recon/scm/PipelineSyncTask.java | 22 +-
.../ozone/recon/scm/ReconDeadNodeHandler.java | 4 +-
.../hadoop/ozone/recon/scm/ReconScmTask.java | 44 ++--
.../ozone/recon/scm/ReconStaleNodeHandler.java | 2 +-
.../scm/ReconStorageContainerManagerFacade.java | 34 +--
.../spi/impl/OzoneManagerServiceProviderImpl.java | 115 +++++----
.../ozone/recon/tasks/ContainerKeyMapperTask.java | 6 +-
.../ozone/recon/tasks/ContainerSizeCountTask.java | 61 ++---
.../ozone/recon/tasks/FileSizeCountTask.java | 4 +-
.../hadoop/ozone/recon/tasks/NSSummaryTask.java | 3 +
.../ozone/recon/tasks/OMDBUpdatesHandler.java | 9 +
.../ozone/recon/tasks/OMUpdateEventBatch.java | 10 +-
.../ozone/recon/tasks/OmTableInsightTask.java | 4 +-
.../ozone/recon/tasks/ReconTaskController.java | 14 +-
.../ozone/recon/tasks/ReconTaskControllerImpl.java | 263 ++++++++++++---------
.../ozone/recon/tasks/types/NamedCallableTask.java | 49 ++++
.../recon/tasks/types/TaskExecutionException.java | 35 +++
.../ozone/recon/tasks/types/package-info.java | 22 ++
.../tasks/updater/ReconTaskStatusUpdater.java | 123 ++++++++++
.../updater/ReconTaskStatusUpdaterManager.java | 69 ++++++
.../ozone/recon/tasks/updater/package-info.java | 22 ++
.../ozone/recon/upgrade/ReconLayoutFeature.java | 3 +-
.../upgrade/ReconTaskStatusTableUpgradeAction.java | 104 ++++++++
.../hadoop/ozone/recon/api/TestEndpoints.java | 2 +-
.../ozone/recon/api/TestTaskStatusService.java | 21 +-
.../ozone/recon/api/TestTriggerDBSyncEndpoint.java | 16 +-
.../ozone/recon/fsck/TestContainerHealthTask.java | 34 ++-
.../TestReconInternalSchemaDefinition.java | 6 +-
.../persistence/TestReconWithDifferentSqlDBs.java | 2 +-
.../recon/persistence/TestSqlSchemaSetup.java | 10 +-
.../impl/TestOzoneManagerServiceProviderImpl.java | 115 +++++----
.../recon/tasks/TestContainerKeyMapperTask.java | 6 +-
.../recon/tasks/TestContainerSizeCountTask.java | 22 +-
.../ozone/recon/tasks/TestFileSizeCountTask.java | 8 +-
.../ozone/recon/tasks/TestNSSummaryTask.java | 2 +-
.../recon/tasks/TestNSSummaryTaskWithFSO.java | 2 +-
.../recon/tasks/TestNSSummaryTaskWithLegacy.java | 2 +-
.../TestNSSummaryTaskWithLegacyOBSLayout.java | 2 +-
.../recon/tasks/TestNSSummaryTaskWithOBS.java | 2 +-
.../ozone/recon/tasks/TestOmTableInsightTask.java | 18 +-
.../recon/tasks/TestReconTaskControllerImpl.java | 53 ++++-
.../recon/tasks/TestReconTaskStatusUpdater.java | 101 ++++++++
49 files changed, 1200 insertions(+), 408 deletions(-)
diff --git a/hadoop-ozone/dist/src/main/smoketest/recon/recon-taskstatus.robot b/hadoop-ozone/dist/src/main/smoketest/recon/recon-taskstatus.robot
new file mode 100644
index 0000000000..3b5b011523
--- /dev/null
+++ b/hadoop-ozone/dist/src/main/smoketest/recon/recon-taskstatus.robot
@@ -0,0 +1,125 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+*** Settings ***
+Documentation Test to validate the recon task status API works properly
+Library OperatingSystem
+Library String
+Library BuiltIn
+Library Collections
+Resource ../ozone-lib/freon.robot
+Resource ../commonlib.robot
+Test Timeout 5 minutes
+
+*** Variables ***
+${BASE_URL} http://recon:9888
+${TASK_STATUS_ENDPOINT} ${BASE_URL}/api/v1/task/status
+${TRIGGER_SYNC_ENDPOINT} ${BASE_URL}/api/v1/triggerdbsync/om
+${TASK_NAME_1} ContainerHealthTask
+${TASK_NAME_2} OmDeltaRequest
+${BUCKET} testbucket
+${VOLUME} testvolume
+${KEYPATH} ${VOLUME}/${BUCKET}/testkey
+
+*** Keywords ***
+
+Kinit as ozone admin
+ Run Keyword if '${SECURITY_ENABLED}' == 'true' Kinit test user testuser testuser.keytab
+
+Sync OM Data
+ Log To Console Sending CURL request to ${TRIGGER_SYNC_ENDPOINT}
+ ${result} = Execute curl --negotiate -u : -LSs ${TRIGGER_SYNC_ENDPOINT}
+ [return] ${result}
+
+Fetch Task Status
+ Log To Console Sending CURL request to ${TASK_STATUS_ENDPOINT}
+ ${result} = Execute curl -H "Accepts: application/json" --negotiate -u : -LSs ${TASK_STATUS_ENDPOINT}
+ ${parsed_response} = Evaluate json.loads('''${result}''')
+ ${tasks} = Evaluate [task for task in ${parsed_response}]
+ [return] ${tasks}
+
+*** Test Cases ***
+
+Prepopulate Data and Trigger OM DB Sync
+ [Documentation] Use Freon to prepopulate the OM DB with data and trigger OM DB sync.
+
+ Kinit as ozone admin
+ Freon DFSG n=100 path=${KEYPATH} size=100
+
+ ${result} = Sync OM Data
+ Should contain ${result} true # Sync should return true if successful
+
+Validate Task Status After Sync
+ [Documentation] Validate that task status is updated after triggering the OM DB sync.
+
+ ${tasks} = Fetch Task Status
+ Should Not Be Empty ${tasks}
+
+ FOR ${task} IN @{tasks}
+ Dictionary Should Contain Key ${task} taskName
+ Dictionary Should Contain Key ${task} lastUpdatedSeqNumber
+ Dictionary Should Contain Key ${task} lastUpdatedTimestamp
+ Dictionary Should Contain Key ${task} isCurrentTaskRunning
+ Dictionary Should Contain Key ${task} lastTaskRunStatus
+ END
+
+Validate Stats for Specific Task
+ [Documentation] Validate response for a specific task after OM DB sync.
+
+ ${tasks} = Fetch Task Status
+
+ ${task_list} = Evaluate [task for task in ${tasks} if task["taskName"] == "${TASK_NAME_1}"]
+ ${list_length} = Get Length ${task_list}
+ Should Be Equal As Integers ${list_length} 1
+
+ ${task} = Get From List ${task_list} 0
+
+ # Validate table fields
+ Should Be True ${task["lastUpdatedTimestamp"]}!=${None}
+ Should Be True ${task["lastUpdatedSeqNumber"]}!=${None}
+ Should Be True ${task["isCurrentTaskRunning"]}!=${None}
+ Should Be True ${task["lastTaskRunStatus"]}!=${None}
+
+Validate All Tasks Updated After Sync
+ [Documentation] Ensure all tasks have been updated after an OM DB sync operation.
+
+ ${tasks} = Fetch Task Status
+ Should Not Be Empty ${tasks}
+
+ FOR ${task} IN @{tasks}
+ Should Be True ${task["lastUpdatedTimestamp"]}!=${None}
+ Should Be True ${task["lastUpdatedSeqNumber"]}!=${None}
+ END
+
+Validate Sequence number is updated after sync
+ Log To Console Triggering OM DB sync for updates
+ Sync OM Data
+ ${tasks} = Fetch Task Status
+ Should Not Be Empty ${tasks}
+
+ ${om_delta_task_list} = Evaluate [task for task in ${tasks} if task["taskName"] == "OmDeltaRequest"]
+ ${list_length} = Get Length ${om_delta_task_list}
+ Should Be Equal As Integers ${list_length} 1
+
+ ${om_delta_task} = Get From List ${om_delta_task_list} 0
+ ${om_delta_task_seq_num} = Evaluate int(${om_delta_task["lastUpdatedSeqNumber"]})
+ ${om_task_names} = Evaluate ["NSSummaryTask", "ContainerKeyMapperTask", "FileSizeCountTask", "OmTableInsightTask"]
+ ${om_tasks} = Evaluate [task for task in ${tasks} if task["taskName"] in ${om_task_names}]
+
+ FOR ${task} IN @{om_tasks}
+ IF ${task["isCurrentTaskRunning"]} == 0
+ Should Be Equal As Integers ${task["lastUpdatedSeqNumber"]} ${om_delta_task_seq_num}
+ END
+ END
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOzoneConfigurationFields.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOzoneConfigurationFields.java
index 3b650f1bf5..809c19c972 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOzoneConfigurationFields.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOzoneConfigurationFields.java
@@ -138,3 +138,4 @@ public class TestOzoneConfigurationFields extends TestConfigurationFieldsBase {
));
}
}
+
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManager.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManager.java
index f51d12a7c5..daf506bc7c 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManager.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManager.java
@@ -28,6 +28,7 @@ import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.RECON_OM_DELTA
import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.RECON_OM_DELTA_UPDATE_LOOP_LIMIT;
import static org.apache.hadoop.ozone.recon.spi.impl.OzoneManagerServiceProviderImpl.OmSnapshotTaskName.OmDeltaRequest;
import static org.apache.hadoop.ozone.recon.spi.impl.OzoneManagerServiceProviderImpl.OmSnapshotTaskName.OmSnapshotRequest;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.slf4j.event.Level.INFO;
@@ -255,8 +256,7 @@ public class TestReconWithOzoneManager {
"lastUpdatedTimestamp");
// verify only Delta updates were added to recon after restart.
- assertEquals(beforeRestartSnapShotTimeStamp,
- afterRestartSnapShotTimeStamp);
+ assertThat(afterRestartSnapShotTimeStamp).isGreaterThanOrEqualTo(beforeRestartSnapShotTimeStamp);
//verify sequence number after Delta Updates
assertEquals(omLatestSeqNumber, reconLatestSeqNumber);
diff --git a/hadoop-ozone/recon-codegen/src/main/java/org/hadoop/ozone/recon/schema/ReconTaskSchemaDefinition.java b/hadoop-ozone/recon-codegen/src/main/java/org/hadoop/ozone/recon/schema/ReconTaskSchemaDefinition.java
index dfa76eac4f..ffa2ec3dbb 100644
--- a/hadoop-ozone/recon-codegen/src/main/java/org/hadoop/ozone/recon/schema/ReconTaskSchemaDefinition.java
+++ b/hadoop-ozone/recon-codegen/src/main/java/org/hadoop/ozone/recon/schema/ReconTaskSchemaDefinition.java
@@ -64,6 +64,8 @@ public class ReconTaskSchemaDefinition implements ReconSchemaDefinition {
.column("task_name", SQLDataType.VARCHAR(766).nullable(false))
.column("last_updated_timestamp", SQLDataType.BIGINT)
.column("last_updated_seq_number", SQLDataType.BIGINT)
+ .column("last_task_run_status", SQLDataType.INTEGER)
+ .column("is_current_task_running", SQLDataType.INTEGER)
.constraint(DSL.constraint("pk_task_name")
.primaryKey("task_name"))
.execute();
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconControllerModule.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconControllerModule.java
index 39f41395bc..1a974dd480 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconControllerModule.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconControllerModule.java
@@ -55,6 +55,7 @@ import org.apache.hadoop.ozone.recon.tasks.ReconOmTask;
import org.apache.hadoop.ozone.recon.tasks.ReconTaskController;
import org.apache.hadoop.ozone.recon.tasks.ReconTaskControllerImpl;
import org.apache.hadoop.ozone.recon.tasks.OmTableInsightTask;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.ratis.protocol.ClientId;
import org.hadoop.ozone.recon.codegen.ReconSqlDbConfig;
@@ -110,6 +111,7 @@ public class ReconControllerModule extends AbstractModule {
install(new ReconOmTaskBindingModule());
install(new ReconDaoBindingModule());
+ bind(ReconTaskStatusUpdaterManager.class).in(Singleton.class);
bind(ReconTaskController.class)
.to(ReconTaskControllerImpl.class).in(Singleton.class);
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/TaskStatusService.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/TaskStatusService.java
index f60fc73737..f2f47cfca7 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/TaskStatusService.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/TaskStatusService.java
@@ -40,13 +40,12 @@ public class TaskStatusService {
private ReconTaskStatusDao reconTaskStatusDao;
/**
- * Return the list of Recon Tasks and the last successful timestamp and
- * sequence number.
+ * Return the list of Recon Tasks and their related stats from RECON_TASK_STATUS table.
* @return {@link Response}
*/
@GET
@Path("status")
- public Response getTaskTimes() {
+ public Response getTaskStats() {
List<ReconTaskStatus> resultSet = reconTaskStatusDao.findAll();
return Response.ok(resultSet).build();
}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/fsck/ContainerHealthTask.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/fsck/ContainerHealthTask.java
index 11af6eaff5..3893123f3a 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/fsck/ContainerHealthTask.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/fsck/ContainerHealthTask.java
@@ -45,9 +45,10 @@ import org.apache.hadoop.ozone.recon.scm.ReconScmTask;
import org.apache.hadoop.ozone.recon.spi.ReconContainerMetadataManager;
import org.apache.hadoop.ozone.recon.spi.StorageContainerServiceProvider;
import org.apache.hadoop.ozone.recon.tasks.ReconTaskConfig;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.apache.hadoop.util.Time;
import org.hadoop.ozone.recon.schema.ContainerSchemaDefinition.UnHealthyContainerStates;
-import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
import org.hadoop.ozone.recon.schema.tables.pojos.UnhealthyContainers;
import org.hadoop.ozone.recon.schema.tables.records.UnhealthyContainersRecord;
import org.jooq.Cursor;
@@ -86,17 +87,18 @@ public class ContainerHealthTask extends ReconScmTask {
private final OzoneConfiguration conf;
+ private final ReconTaskStatusUpdater taskStatusUpdater;
+
@SuppressWarnings("checkstyle:ParameterNumber")
public ContainerHealthTask(
ContainerManager containerManager,
StorageContainerServiceProvider scmClient,
- ReconTaskStatusDao reconTaskStatusDao,
ContainerHealthSchemaManager containerHealthSchemaManager,
PlacementPolicy placementPolicy,
ReconTaskConfig reconTaskConfig,
ReconContainerMetadataManager reconContainerMetadataManager,
- OzoneConfiguration conf) {
- super(reconTaskStatusDao);
+ OzoneConfiguration conf, ReconTaskStatusUpdaterManager taskStatusUpdaterManager) {
+ super(taskStatusUpdaterManager);
this.scmClient = scmClient;
this.containerHealthSchemaManager = containerHealthSchemaManager;
this.reconContainerMetadataManager = reconContainerMetadataManager;
@@ -104,13 +106,14 @@ public class ContainerHealthTask extends ReconScmTask {
this.containerManager = containerManager;
this.conf = conf;
interval = reconTaskConfig.getMissingContainerTaskInterval().toMillis();
+ this.taskStatusUpdater = getTaskStatusUpdater();
}
@Override
public void run() {
try {
while (canRun()) {
- triggerContainerHealthCheck();
+ initializeAndRunTask();
Thread.sleep(interval);
}
} catch (Throwable t) {
@@ -118,10 +121,13 @@ public class ContainerHealthTask extends ReconScmTask {
if (t instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
+ taskStatusUpdater.setLastTaskRunStatus(-1);
+ taskStatusUpdater.recordRunCompletion();
}
}
- public void triggerContainerHealthCheck() {
+ @Override
+ protected void runTask() throws Exception {
lock.writeLock().lock();
// Map contains all UNHEALTHY STATES as keys and value is another map
// with 3 keys (CONTAINER_COUNT, TOTAL_KEYS, TOTAL_USED_BYTES) and value
@@ -144,7 +150,11 @@ public class ContainerHealthTask extends ReconScmTask {
" process {} existing database records.",
Time.monotonicNow() - start, existingCount);
+ start = Time.monotonicNow();
checkAndProcessContainers(unhealthyContainerStateStatsMap, currentTime);
+ LOG.debug("Container Health Task thread took {} milliseconds to process containers",
+ Time.monotonicNow() - start);
+ taskStatusUpdater.setLastTaskRunStatus(0);
processedContainers.clear();
} finally {
lock.writeLock().unlock();
@@ -165,7 +175,6 @@ public class ContainerHealthTask extends ReconScmTask {
.filter(c -> !processedContainers.contains(c))
.forEach(c -> processContainer(c, currentTime,
unhealthyContainerStateStatsMap));
- recordSingleRunCompletion();
LOG.debug("Container Health task thread took {} milliseconds for" +
" processing {} containers.", Time.monotonicNow() - start,
containers.size());
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/PipelineSyncTask.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/PipelineSyncTask.java
index ae3a3531cb..a2acebff63 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/PipelineSyncTask.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/PipelineSyncTask.java
@@ -22,7 +22,6 @@ import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.DEAD;
import java.io.IOException;
import java.util.List;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
@@ -34,8 +33,9 @@ import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.ozone.recon.spi.StorageContainerServiceProvider;
import org.apache.hadoop.ozone.recon.tasks.ReconTaskConfig;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.apache.hadoop.util.Time;
-import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -55,24 +55,26 @@ public class PipelineSyncTask extends ReconScmTask {
private ReadWriteLock lock = new ReentrantReadWriteLock(true);
private final long interval;
+ private final ReconTaskStatusUpdater taskStatusUpdater;
public PipelineSyncTask(ReconPipelineManager pipelineManager,
ReconNodeManager nodeManager,
StorageContainerServiceProvider scmClient,
- ReconTaskStatusDao reconTaskStatusDao,
- ReconTaskConfig reconTaskConfig) {
- super(reconTaskStatusDao);
+ ReconTaskConfig reconTaskConfig,
+ ReconTaskStatusUpdaterManager taskStatusUpdaterManager) {
+ super(taskStatusUpdaterManager);
this.scmClient = scmClient;
this.reconPipelineManager = pipelineManager;
this.nodeManager = nodeManager;
this.interval = reconTaskConfig.getPipelineSyncTaskInterval().toMillis();
+ this.taskStatusUpdater = getTaskStatusUpdater();
}
@Override
public void run() {
try {
while (canRun()) {
- triggerPipelineSyncTask();
+ initializeAndRunTask();
Thread.sleep(interval);
}
} catch (Throwable t) {
@@ -80,11 +82,13 @@ public class PipelineSyncTask extends ReconScmTask {
if (t instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
+ taskStatusUpdater.setLastTaskRunStatus(-1);
+ taskStatusUpdater.recordRunCompletion();
}
}
- public void triggerPipelineSyncTask()
- throws IOException, TimeoutException, NodeNotFoundException {
+ @Override
+ protected void runTask() throws IOException, NodeNotFoundException {
lock.writeLock().lock();
try {
long start = Time.monotonicNow();
@@ -93,7 +97,7 @@ public class PipelineSyncTask extends ReconScmTask {
syncOperationalStateOnDeadNodes();
LOG.debug("Pipeline sync Thread took {} milliseconds.",
Time.monotonicNow() - start);
- recordSingleRunCompletion();
+ taskStatusUpdater.setLastTaskRunStatus(0);
} finally {
lock.writeLock().unlock();
}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconDeadNodeHandler.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconDeadNodeHandler.java
index 828942c8e5..a5d5552621 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconDeadNodeHandler.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconDeadNodeHandler.java
@@ -83,8 +83,8 @@ public class ReconDeadNodeHandler extends DeadNodeHandler {
LOG.warn("Node {} has reached DEAD state, but SCM does not have " +
"information about it.", datanodeDetails);
}
- containerHealthTask.triggerContainerHealthCheck();
- pipelineSyncTask.triggerPipelineSyncTask();
+ containerHealthTask.initializeAndRunTask();
+ pipelineSyncTask.initializeAndRunTask();
} catch (Exception ioEx) {
LOG.error("Error trying to verify Node operational state from SCM.",
ioEx);
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconScmTask.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconScmTask.java
index 7c070c7d2b..7127d73038 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconScmTask.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconScmTask.java
@@ -18,8 +18,8 @@
package org.apache.hadoop.ozone.recon.scm;
-import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
-import org.hadoop.ozone.recon.schema.tables.pojos.ReconTaskStatus;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -30,28 +30,20 @@ public abstract class ReconScmTask {
private static final Logger LOG = LoggerFactory.getLogger(ReconScmTask.class);
private Thread taskThread;
- private ReconTaskStatusDao reconTaskStatusDao;
private volatile boolean running;
+ private final ReconTaskStatusUpdater taskStatusUpdater;
- protected ReconScmTask(ReconTaskStatusDao reconTaskStatusDao) {
- this.reconTaskStatusDao = reconTaskStatusDao;
- }
-
- private void register() {
- String taskName = getTaskName();
- if (!reconTaskStatusDao.existsById(taskName)) {
- ReconTaskStatus reconTaskStatusRecord = new ReconTaskStatus(
- taskName, 0L, 0L);
- reconTaskStatusDao.insert(reconTaskStatusRecord);
- LOG.info("Registered {} task ", taskName);
- }
+ protected ReconScmTask(
+ ReconTaskStatusUpdaterManager taskStatusUpdaterManager
+ ) {
+ // In case the task is not already present in the DB, table is updated with initial values for task
+ this.taskStatusUpdater = taskStatusUpdaterManager.getTaskStatusUpdater(getTaskName());
}
/**
* Start underlying start thread.
*/
public synchronized void start() {
- register();
if (!isRunning()) {
LOG.info("Starting {} Thread.", getTaskName());
running = true;
@@ -87,11 +79,6 @@ public abstract class ReconScmTask {
return true;
}
- protected void recordSingleRunCompletion() {
- reconTaskStatusDao.update(new ReconTaskStatus(getTaskName(),
- System.currentTimeMillis(), 0L));
- }
-
protected boolean canRun() {
return running;
}
@@ -100,5 +87,20 @@ public abstract class ReconScmTask {
return getClass().getSimpleName();
}
+ public ReconTaskStatusUpdater getTaskStatusUpdater() {
+ return this.taskStatusUpdater;
+ }
+
protected abstract void run();
+
+ protected void initializeAndRunTask() throws Exception {
+ taskStatusUpdater.recordRunStart();
+ runTask();
+ taskStatusUpdater.recordRunCompletion();
+ }
+
+ /**
+ * Override this method for the actual processing logic in child tasks.
+ */
+ protected abstract void runTask() throws Exception;
}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconStaleNodeHandler.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconStaleNodeHandler.java
index 998f063924..8a3ffb90c4 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconStaleNodeHandler.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconStaleNodeHandler.java
@@ -49,7 +49,7 @@ public class ReconStaleNodeHandler extends StaleNodeHandler {
final EventPublisher publisher) {
super.onMessage(datanodeDetails, publisher);
try {
- pipelineSyncTask.triggerPipelineSyncTask();
+ pipelineSyncTask.initializeAndRunTask();
} catch (Exception exp) {
LOG.error("Error trying to trigger pipeline sync task..",
exp);
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconStorageContainerManagerFacade.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconStorageContainerManagerFacade.java
index eff68848a2..2c9ac7a8c2 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconStorageContainerManagerFacade.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconStorageContainerManagerFacade.java
@@ -124,10 +124,10 @@ import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.Containe
import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.ContainerReportFromDatanode;
import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.IncrementalContainerReportFromDatanode;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.apache.ratis.util.ExitUtils;
import org.hadoop.ozone.recon.schema.UtilizationSchemaDefinition;
import org.hadoop.ozone.recon.schema.tables.daos.ContainerCountBySizeDao;
-import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -184,7 +184,6 @@ public class ReconStorageContainerManagerFacade
@SuppressWarnings({"checkstyle:ParameterNumber", "checkstyle:MethodLength"})
public ReconStorageContainerManagerFacade(OzoneConfiguration conf,
StorageContainerServiceProvider scmServiceProvider,
- ReconTaskStatusDao reconTaskStatusDao,
ContainerCountBySizeDao containerCountBySizeDao,
UtilizationSchemaDefinition utilizationSchemaDefinition,
ContainerHealthSchemaManager containerHealthSchemaManager,
@@ -192,7 +191,9 @@ public class ReconStorageContainerManagerFacade
ReconUtils reconUtils,
ReconSafeModeManager safeModeManager,
ReconContext reconContext,
- DataSource dataSource) throws IOException {
+ DataSource dataSource,
+ ReconTaskStatusUpdaterManager taskStatusUpdaterManager)
+ throws IOException {
reconNodeDetails = reconUtils.getReconNodeDetails(conf);
this.threadNamePrefix = reconNodeDetails.threadNamePrefix();
this.eventQueue = new EventQueue(threadNamePrefix);
@@ -270,24 +271,15 @@ public class ReconStorageContainerManagerFacade
new PipelineActionHandler(pipelineManager, scmContext, conf);
ReconTaskConfig reconTaskConfig = conf.getObject(ReconTaskConfig.class);
- PipelineSyncTask pipelineSyncTask = new PipelineSyncTask(
- pipelineManager,
- nodeManager,
- scmServiceProvider,
- reconTaskStatusDao,
- reconTaskConfig);
- containerHealthTask = new ContainerHealthTask(
- containerManager, scmServiceProvider, reconTaskStatusDao,
- containerHealthSchemaManager, containerPlacementPolicy, reconTaskConfig,
- reconContainerMetadataManager, conf);
-
- this.containerSizeCountTask = new ContainerSizeCountTask(
- containerManager,
- scmServiceProvider,
- reconTaskStatusDao,
- reconTaskConfig,
- containerCountBySizeDao,
- utilizationSchemaDefinition);
+ PipelineSyncTask pipelineSyncTask = new PipelineSyncTask(pipelineManager, nodeManager,
+ scmServiceProvider, reconTaskConfig, taskStatusUpdaterManager);
+
+ containerHealthTask = new ContainerHealthTask(containerManager, scmServiceProvider,
+ containerHealthSchemaManager, containerPlacementPolicy,
+ reconTaskConfig, reconContainerMetadataManager, conf, taskStatusUpdaterManager);
+
+ this.containerSizeCountTask = new ContainerSizeCountTask(containerManager, scmServiceProvider,
+ reconTaskConfig, containerCountBySizeDao, utilizationSchemaDefinition, taskStatusUpdaterManager);
this.dataSource = dataSource;
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java
index d5b7b1cfc9..5a49e55b06 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java
@@ -69,6 +69,8 @@ import org.apache.hadoop.ozone.recon.spi.OzoneManagerServiceProvider;
import org.apache.hadoop.ozone.recon.tasks.OMDBUpdatesHandler;
import org.apache.hadoop.ozone.recon.tasks.OMUpdateEventBatch;
import org.apache.hadoop.ozone.recon.tasks.ReconTaskController;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.util.Time;
@@ -97,8 +99,6 @@ import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.RECON_OM_DELTA
import static org.apache.hadoop.ozone.recon.ReconUtils.convertNumericToSymbolic;
import static org.apache.ratis.proto.RaftProtos.RaftPeerRole.LEADER;
-import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
-import org.hadoop.ozone.recon.schema.tables.pojos.ReconTaskStatus;
import org.rocksdb.RocksDBException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -124,7 +124,6 @@ public class OzoneManagerServiceProviderImpl
private ReconOMMetadataManager omMetadataManager;
private ReconTaskController reconTaskController;
- private ReconTaskStatusDao reconTaskStatusDao;
private ReconUtils reconUtils;
private OzoneManagerSyncMetrics metrics;
@@ -135,6 +134,7 @@ public class OzoneManagerServiceProviderImpl
private final String threadNamePrefix;
private ThreadFactory threadFactory;
private ReconContext reconContext;
+ private ReconTaskStatusUpdaterManager taskStatusUpdaterManager;
/**
* OM Snapshot related task names.
@@ -145,13 +145,15 @@ public class OzoneManagerServiceProviderImpl
}
@Inject
+ @SuppressWarnings("checkstyle:ParameterNumber")
public OzoneManagerServiceProviderImpl(
OzoneConfiguration configuration,
ReconOMMetadataManager omMetadataManager,
ReconTaskController reconTaskController,
ReconUtils reconUtils,
OzoneManagerProtocol ozoneManagerClient,
- ReconContext reconContext) {
+ ReconContext reconContext,
+ ReconTaskStatusUpdaterManager taskStatusUpdaterManager) {
int connectionTimeout = (int) configuration.getTimeDuration(
OZONE_RECON_OM_CONNECTION_TIMEOUT,
@@ -212,7 +214,6 @@ public class OzoneManagerServiceProviderImpl
this.reconUtils = reconUtils;
this.omMetadataManager = omMetadataManager;
this.reconTaskController = reconTaskController;
- this.reconTaskStatusDao = reconTaskController.getReconTaskStatusDao();
this.ozoneManagerClient = ozoneManagerClient;
this.configuration = configuration;
this.metrics = OzoneManagerSyncMetrics.create();
@@ -225,28 +226,7 @@ public class OzoneManagerServiceProviderImpl
new ThreadFactoryBuilder().setNameFormat(threadNamePrefix + "SyncOM-%d")
.build();
this.reconContext = reconContext;
- }
-
- public void registerOMDBTasks() {
- ReconTaskStatus reconTaskStatusRecord = new ReconTaskStatus(
- OmSnapshotTaskName.OmDeltaRequest.name(),
- System.currentTimeMillis(), getCurrentOMDBSequenceNumber());
- if (!reconTaskStatusDao.existsById(
- OmSnapshotTaskName.OmDeltaRequest.name())) {
- reconTaskStatusDao.insert(reconTaskStatusRecord);
- LOG.info("Registered {} task ",
- OmSnapshotTaskName.OmDeltaRequest.name());
- }
-
- reconTaskStatusRecord = new ReconTaskStatus(
- OmSnapshotTaskName.OmSnapshotRequest.name(),
- System.currentTimeMillis(), getCurrentOMDBSequenceNumber());
- if (!reconTaskStatusDao.existsById(
- OmSnapshotTaskName.OmSnapshotRequest.name())) {
- reconTaskStatusDao.insert(reconTaskStatusRecord);
- LOG.info("Registered {} task ",
- OmSnapshotTaskName.OmSnapshotRequest.name());
- }
+ this.taskStatusUpdaterManager = taskStatusUpdaterManager;
}
@Override
@@ -258,7 +238,6 @@ public class OzoneManagerServiceProviderImpl
public void start() {
LOG.info("Starting Ozone Manager Service Provider.");
scheduler = Executors.newScheduledThreadPool(1, threadFactory);
- registerOMDBTasks();
try {
omMetadataManager.start(configuration);
} catch (IOException ioEx) {
@@ -304,7 +283,7 @@ public class OzoneManagerServiceProviderImpl
LOG.info("Last known sequence number before sync: {}", getCurrentOMDBSequenceNumber());
boolean isSuccess = syncDataFromOM();
if (!isSuccess) {
- LOG.debug("OM DB sync is already running.");
+ LOG.debug("OM DB sync is already running, or encountered an error while trying to sync data.");
}
LOG.info("Sequence number after sync: {}", getCurrentOMDBSequenceNumber());
} catch (Throwable t) {
@@ -495,6 +474,8 @@ public class OzoneManagerServiceProviderImpl
inLoopStartSequenceNumber = inLoopLatestSequenceNumber;
loopCount++;
}
+
+ omdbUpdatesHandler.setLatestSequenceNumber(getCurrentOMDBSequenceNumber());
LOG.info("Delta updates received from OM : {} loops, {} records", loopCount,
getCurrentOMDBSequenceNumber() - fromSequenceNumber
);
@@ -551,12 +532,21 @@ public class OzoneManagerServiceProviderImpl
}
/**
- * Based on current state of Recon's OM DB, we either get delta updates or
- * full snapshot from Ozone Manager.
+ * This method performs the syncing of data from OM.
+ * <ul>
+ * <li>Initially it will fetch a snapshot of OM DB.</li>
+ * <li>If we already have data synced it will try to fetch delta updates.</li>
+ * <li>If the sync is completed successfully it will trigger other OM tasks to process events</li>
+ * <li>If there is any exception while trying to fetch delta updates, it will fall back to full snapshot update</li>
+ * <li>If there is any exception in full snapshot update it will do nothing, and return true.</li>
+ * <li>In case of an interrupt signal (irrespective of delta or snapshot sync),
+ * it will catch and mark the task as interrupted, and return false i.e. sync failed status.</li>
+ * </ul>
* @return true or false if sync operation between Recon and OM was successful or failed.
*/
@VisibleForTesting
public boolean syncDataFromOM() {
+ ReconTaskStatusUpdater reconTaskUpdater;
if (isSyncDataFromOMRunning.compareAndSet(false, true)) {
try {
LOG.info("Syncing data from Ozone Manager.");
@@ -567,25 +557,43 @@ public class OzoneManagerServiceProviderImpl
if (currentSequenceNumber <= 0) {
fullSnapshot = true;
} else {
+ reconTaskUpdater = taskStatusUpdaterManager.getTaskStatusUpdater(
+ OmSnapshotTaskName.OmDeltaRequest.name());
+
try (OMDBUpdatesHandler omdbUpdatesHandler =
new OMDBUpdatesHandler(omMetadataManager)) {
LOG.info("Obtaining delta updates from Ozone Manager");
- // Get updates from OM and apply to local Recon OM DB.
+
+ // If interrupt was previously signalled,
+ // we should check for it before starting delta update sync.
+ if (Thread.currentThread().isInterrupted()) {
+ throw new InterruptedException("Thread interrupted during delta update.");
+ }
+
+ // Get updates from OM and apply to local Recon OM DB and update task status in table
+ reconTaskUpdater.recordRunStart();
getAndApplyDeltaUpdatesFromOM(currentSequenceNumber,
omdbUpdatesHandler);
- // Update timestamp of successful delta updates query.
- ReconTaskStatus reconTaskStatusRecord = new ReconTaskStatus(
- OmSnapshotTaskName.OmDeltaRequest.name(),
- System.currentTimeMillis(), getCurrentOMDBSequenceNumber());
- reconTaskStatusDao.update(reconTaskStatusRecord);
+ reconTaskUpdater.setLastTaskRunStatus(0);
+ reconTaskUpdater.setLastUpdatedSeqNumber(getCurrentOMDBSequenceNumber());
+ reconTaskUpdater.recordRunCompletion();
// Pass on DB update events to tasks that are listening.
reconTaskController.consumeOMEvents(new OMUpdateEventBatch(
- omdbUpdatesHandler.getEvents()), omMetadataManager);
+ omdbUpdatesHandler.getEvents(), omdbUpdatesHandler.getLatestSequenceNumber()), omMetadataManager);
} catch (InterruptedException intEx) {
+ LOG.error("OM DB Delta update sync thread was interrupted.");
+ // We are updating the table even if it didn't run i.e. got interrupted beforehand
+ // to indicate that a task was supposed to run, but it didn't.
+ reconTaskUpdater.setLastTaskRunStatus(-1);
+ reconTaskUpdater.recordRunCompletion();
Thread.currentThread().interrupt();
+ // Since thread is interrupted, we do not fall back to snapshot sync. Return with sync failed status.
+ return false;
} catch (Exception e) {
metrics.incrNumDeltaRequestsFailed();
+ reconTaskUpdater.setLastTaskRunStatus(-1);
+ reconTaskUpdater.recordRunCompletion();
LOG.warn("Unable to get and apply delta updates from OM.",
e.getMessage());
fullSnapshot = true;
@@ -593,19 +601,26 @@ public class OzoneManagerServiceProviderImpl
}
if (fullSnapshot) {
+ reconTaskUpdater = taskStatusUpdaterManager.getTaskStatusUpdater(
+ OmSnapshotTaskName.OmSnapshotRequest.name());
try {
metrics.incrNumSnapshotRequests();
LOG.info("Obtaining full snapshot from Ozone Manager");
+
+ // Similarly if the interrupt was signalled in between,
+ // we should check before starting snapshot sync.
+ if (Thread.currentThread().isInterrupted()) {
+ throw new InterruptedException("Thread interrupted during snapshot sync.");
+ }
+
// Update local Recon OM DB to new snapshot.
+ reconTaskUpdater.recordRunStart();
boolean success = updateReconOmDBWithNewSnapshot();
// Update timestamp of successful delta updates query.
if (success) {
- ReconTaskStatus reconTaskStatusRecord =
- new ReconTaskStatus(
- OmSnapshotTaskName.OmSnapshotRequest.name(),
- System.currentTimeMillis(),
- getCurrentOMDBSequenceNumber());
- reconTaskStatusDao.update(reconTaskStatusRecord);
+ reconTaskUpdater.setLastUpdatedSeqNumber(getCurrentOMDBSequenceNumber());
+ reconTaskUpdater.setLastTaskRunStatus(0);
+ reconTaskUpdater.recordRunCompletion();
// Reinitialize tasks that are listening.
LOG.info("Calling reprocess on Recon tasks.");
@@ -616,14 +631,23 @@ public class OzoneManagerServiceProviderImpl
reconContext.getErrors().remove(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
} else {
metrics.incrNumSnapshotRequestsFailed();
+ reconTaskUpdater.setLastTaskRunStatus(-1);
+ reconTaskUpdater.recordRunCompletion();
// Update health status in ReconContext
reconContext.updateHealthStatus(new AtomicBoolean(false));
reconContext.updateErrors(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
}
} catch (InterruptedException intEx) {
+ LOG.error("OM DB Snapshot update sync thread was interrupted.");
+ reconTaskUpdater.setLastTaskRunStatus(-1);
+ reconTaskUpdater.recordRunCompletion();
Thread.currentThread().interrupt();
+ // Mark sync status as failed.
+ return false;
} catch (Exception e) {
metrics.incrNumSnapshotRequestsFailed();
+ reconTaskUpdater.setLastTaskRunStatus(-1);
+ reconTaskUpdater.recordRunCompletion();
LOG.error("Unable to update Recon's metadata with new OM DB. ", e);
// Update health status in ReconContext
reconContext.updateHealthStatus(new AtomicBoolean(false));
@@ -699,7 +723,8 @@ public class OzoneManagerServiceProviderImpl
* Get OM RocksDB's latest sequence number.
* @return latest sequence number.
*/
- private long getCurrentOMDBSequenceNumber() {
+ @VisibleForTesting
+ public long getCurrentOMDBSequenceNumber() {
return omMetadataManager.getLastSequenceNumberFromDB();
}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ContainerKeyMapperTask.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ContainerKeyMapperTask.java
index bf34c9f893..3202f6aa8b 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ContainerKeyMapperTask.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ContainerKeyMapperTask.java
@@ -208,7 +208,7 @@ public class ContainerKeyMapperTask implements ReconOmTask {
Map<Long, Long> containerKeyCountMap = new HashMap<>();
// List of the deleted (container, key) pair's
List<ContainerKeyPrefix> deletedKeyCountList = new ArrayList<>();
-
+ long startTime = System.currentTimeMillis();
while (eventIterator.hasNext()) {
OMDBUpdateEvent<String, OmKeyInfo> omdbUpdateEvent = eventIterator.next();
// Filter event inside process method to avoid duping
@@ -258,8 +258,8 @@ public class ContainerKeyMapperTask implements ReconOmTask {
LOG.error("Unable to write Container Key Prefix data in Recon DB.", e);
return new ImmutablePair<>(getTaskName(), false);
}
- LOG.debug("{} successfully processed {} OM DB update event(s).",
- getTaskName(), eventCount);
+ LOG.debug("{} successfully processed {} OM DB update event(s) in {} milliseconds.",
+ getTaskName(), eventCount, (System.currentTimeMillis() - startTime));
return new ImmutablePair<>(getTaskName(), true);
}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ContainerSizeCountTask.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ContainerSizeCountTask.java
index f95bc3a855..de55fd4fe9 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ContainerSizeCountTask.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ContainerSizeCountTask.java
@@ -19,16 +19,18 @@
package org.apache.hadoop.ozone.recon.tasks;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.container.ContainerInfo;
import org.apache.hadoop.hdds.scm.container.ContainerManager;
import org.apache.hadoop.ozone.recon.ReconUtils;
import org.apache.hadoop.ozone.recon.scm.ReconScmTask;
import org.apache.hadoop.ozone.recon.spi.StorageContainerServiceProvider;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.hadoop.ozone.recon.schema.ContainerSchemaDefinition;
import org.hadoop.ozone.recon.schema.UtilizationSchemaDefinition;
import org.hadoop.ozone.recon.schema.tables.daos.ContainerCountBySizeDao;
-import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
import org.hadoop.ozone.recon.schema.tables.pojos.ContainerCountBySize;
import org.jooq.DSLContext;
import org.jooq.Record1;
@@ -65,20 +67,22 @@ public class ContainerSizeCountTask extends ReconScmTask {
private Map<ContainerSchemaDefinition.UnHealthyContainerStates, Map<String, Long>>
unhealthyContainerStateStatsMap;
private ReadWriteLock lock = new ReentrantReadWriteLock(true);
+ private final ReconTaskStatusUpdater taskStatusUpdater;
public ContainerSizeCountTask(
ContainerManager containerManager,
StorageContainerServiceProvider scmClient,
- ReconTaskStatusDao reconTaskStatusDao,
ReconTaskConfig reconTaskConfig,
ContainerCountBySizeDao containerCountBySizeDao,
- UtilizationSchemaDefinition utilizationSchemaDefinition) {
- super(reconTaskStatusDao);
+ UtilizationSchemaDefinition utilizationSchemaDefinition,
+ ReconTaskStatusUpdaterManager taskStatusUpdaterManager) {
+ super(taskStatusUpdaterManager);
this.scmClient = scmClient;
this.containerManager = containerManager;
this.containerCountBySizeDao = containerCountBySizeDao;
this.dslContext = utilizationSchemaDefinition.getDSLContext();
interval = reconTaskConfig.getContainerSizeCountTaskInterval().toMillis();
+ this.taskStatusUpdater = getTaskStatusUpdater();
}
@@ -93,21 +97,8 @@ public class ContainerSizeCountTask extends ReconScmTask {
while (canRun()) {
wait(interval);
long startTime, endTime, duration, durationMilliseconds;
- final List<ContainerInfo> containers = containerManager.getContainers();
- if (processedContainers.isEmpty()) {
- try {
- int execute =
- dslContext.truncate(CONTAINER_COUNT_BY_SIZE).execute();
- LOG.debug("Deleted {} records from {}", execute,
- CONTAINER_COUNT_BY_SIZE);
- } catch (Exception e) {
- LOG.error("An error occurred while truncating the table {}: {}",
- CONTAINER_COUNT_BY_SIZE, e.getMessage(), e);
- return;
- }
- }
startTime = System.nanoTime();
- process(containers);
+ initializeAndRunTask();
endTime = System.nanoTime();
duration = endTime - startTime;
durationMilliseconds = duration / 1_000_000;
@@ -119,11 +110,13 @@ public class ContainerSizeCountTask extends ReconScmTask {
if (t instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
+ taskStatusUpdater.setLastTaskRunStatus(-1);
+ taskStatusUpdater.recordRunCompletion();
}
}
private void process(ContainerInfo container,
- Map<ContainerSizeCountKey, Long> map) {
+ Map<ContainerSizeCountKey, Long> map) {
final ContainerID id = container.containerID();
final long usedBytes = container.getUsedBytes();
final long currentSize;
@@ -143,8 +136,18 @@ public class ContainerSizeCountTask extends ReconScmTask {
incrementContainerSizeCount(currentSize, map);
}
+ @Override
+ protected void runTask() throws Exception {
+ final List<ContainerInfo> containers = containerManager.getContainers();
+ if (processedContainers.isEmpty()) {
+ int execute = dslContext.truncate(CONTAINER_COUNT_BY_SIZE).execute();
+ LOG.debug("Deleted {} records from {}", execute, CONTAINER_COUNT_BY_SIZE);
+ }
+ processContainers(containers);
+ }
+
/**
- * The process() function is responsible for updating the counts of
+ * The processContainers() function is responsible for updating the counts of
* containers being tracked in a containerSizeCountMap based on the
* ContainerInfo objects in the list containers. It then iterates through
* the list of containers and does the following for each container:
@@ -168,8 +171,11 @@ public class ContainerSizeCountTask extends ReconScmTask {
* size counts. Finally, the counts in the containerSizeCountMap are written
* to the database using the writeCountsToDB() function.
*/
- public void process(List<ContainerInfo> containers) {
+ @VisibleForTesting
+ public void processContainers(List<ContainerInfo> containers) {
lock.writeLock().lock();
+ boolean processingFailed = false;
+
try {
final Map<ContainerSizeCountKey, Long> containerSizeCountMap
= new HashMap<>();
@@ -186,6 +192,7 @@ public class ContainerSizeCountTask extends ReconScmTask {
try {
process(container, containerSizeCountMap);
} catch (Exception e) {
+ processingFailed = true;
// FIXME: it is a bug if there is an exception.
LOG.error("FIXME: Failed to process " + container, e);
}
@@ -198,6 +205,7 @@ public class ContainerSizeCountTask extends ReconScmTask {
writeCountsToDB(false, containerSizeCountMap);
containerSizeCountMap.clear();
LOG.debug("Completed a 'process' run of ContainerSizeCountTask.");
+ taskStatusUpdater.setLastTaskRunStatus(processingFailed ? -1 : 0);
} finally {
lock.writeLock().unlock();
}
@@ -258,11 +266,6 @@ public class ContainerSizeCountTask extends ReconScmTask {
containerCountBySizeDao.update(updateInDb);
}
- @Override
- public String getTaskName() {
- return "ContainerSizeCountTask";
- }
-
/**
*
* Handles the deletion of containers by updating the tracking of processed containers
@@ -305,7 +308,7 @@ public class ContainerSizeCountTask extends ReconScmTask {
* @param containerSize to calculate the upperSizeBound
*/
private static void incrementContainerSizeCount(long containerSize,
- Map<ContainerSizeCountKey, Long> containerSizeCountMap) {
+ Map<ContainerSizeCountKey, Long> containerSizeCountMap) {
updateContainerSizeCount(containerSize, 1, containerSizeCountMap);
}
@@ -326,12 +329,12 @@ public class ContainerSizeCountTask extends ReconScmTask {
* @param containerSize to calculate the upperSizeBound
*/
private static void decrementContainerSizeCount(long containerSize,
- Map<ContainerSizeCountKey, Long> containerSizeCountMap) {
+ Map<ContainerSizeCountKey, Long> containerSizeCountMap) {
updateContainerSizeCount(containerSize, -1, containerSizeCountMap);
}
private static void updateContainerSizeCount(long containerSize, int delta,
- Map<ContainerSizeCountKey, Long> containerSizeCountMap) {
+ Map<ContainerSizeCountKey, Long> containerSizeCountMap) {
ContainerSizeCountKey key = getContainerSizeCountKey(containerSize);
containerSizeCountMap.compute(key,
(k, previous) -> previous != null ? previous + delta : delta);
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/FileSizeCountTask.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/FileSizeCountTask.java
index cbbbe6c373..6906d4bbb8 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/FileSizeCountTask.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/FileSizeCountTask.java
@@ -149,6 +149,7 @@ public class FileSizeCountTask implements ReconOmTask {
Map<FileSizeCountKey, Long> fileSizeCountMap = new HashMap<>();
final Collection<String> taskTables = getTaskTables();
+ long startTime = System.currentTimeMillis();
while (eventIterator.hasNext()) {
OMDBUpdateEvent<String, Object> omdbUpdateEvent = eventIterator.next();
// Filter event inside process method to avoid duping
@@ -198,7 +199,8 @@ public class FileSizeCountTask implements ReconOmTask {
}
}
writeCountsToDB(false, fileSizeCountMap);
- LOG.debug("Completed a 'process' run of FileSizeCountTask.");
+ LOG.debug("{} successfully processed in {} milliseconds",
+ getTaskName(), (System.currentTimeMillis() - startTime));
return new ImmutablePair<>(getTaskName(), true);
}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/NSSummaryTask.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/NSSummaryTask.java
index 256e0c687b..5048050f9d 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/NSSummaryTask.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/NSSummaryTask.java
@@ -100,6 +100,7 @@ public class NSSummaryTask implements ReconOmTask {
@Override
public Pair<String, Boolean> process(OMUpdateEventBatch events) {
+ long startTime = System.currentTimeMillis();
boolean success = nsSummaryTaskWithFSO.processWithFSO(events);
if (!success) {
LOG.error("processWithFSO failed.");
@@ -112,6 +113,8 @@ public class NSSummaryTask implements ReconOmTask {
if (!success) {
LOG.error("processWithOBS failed.");
}
+ LOG.debug("{} successfully processed in {} milliseconds",
+ getTaskName(), (System.currentTimeMillis() - startTime));
return new ImmutablePair<>(getTaskName(), success);
}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OMDBUpdatesHandler.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OMDBUpdatesHandler.java
index d1f98c49bd..d586718ffb 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OMDBUpdatesHandler.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OMDBUpdatesHandler.java
@@ -51,12 +51,21 @@ public class OMDBUpdatesHandler extends ManagedWriteBatch.Handler {
private Map<String, Map<Object, OMDBUpdateEvent>> omdbLatestUpdateEvents = new HashMap<>();
private final OMDBDefinition omdbDefinition = OMDBDefinition.get();
private final OmUpdateEventValidator omUpdateEventValidator = new OmUpdateEventValidator(omdbDefinition);
+ private long batchSequenceNumber; // Store the current sequence number for the batch
public OMDBUpdatesHandler(OMMetadataManager metadataManager) {
omMetadataManager = metadataManager;
tablesNames = metadataManager.getStore().getTableNames();
}
+ public void setLatestSequenceNumber(long sequenceNumber) {
+ this.batchSequenceNumber = sequenceNumber;
+ }
+
+ public long getLatestSequenceNumber() {
+ return this.batchSequenceNumber;
+ }
+
@Override
public void put(int cfIndex, byte[] keyBytes, byte[] valueBytes) {
try {
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OMUpdateEventBatch.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OMUpdateEventBatch.java
index 3ed50a4dd0..1a724162eb 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OMUpdateEventBatch.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OMUpdateEventBatch.java
@@ -27,9 +27,11 @@ import java.util.List;
public class OMUpdateEventBatch {
private final List<OMDBUpdateEvent> events;
+ private final long batchSequenceNumber;
- public OMUpdateEventBatch(List<OMDBUpdateEvent> e) {
+ public OMUpdateEventBatch(List<OMDBUpdateEvent> e, long batchSequenceNumber) {
events = e;
+ this.batchSequenceNumber = batchSequenceNumber;
}
/**
@@ -37,11 +39,7 @@ public class OMUpdateEventBatch {
* @return Event Info instance.
*/
long getLastSequenceNumber() {
- if (events.isEmpty()) {
- return -1;
- } else {
- return events.get(events.size() - 1).getSequenceNumber();
- }
+ return this.batchSequenceNumber;
}
/**
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OmTableInsightTask.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OmTableInsightTask.java
index 3dc91d0dfc..37a0e16e93 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OmTableInsightTask.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OmTableInsightTask.java
@@ -164,6 +164,7 @@ public class OmTableInsightTask implements ReconOmTask {
final Collection<String> taskTables = getTaskTables();
// Process each update event
+ long startTime = System.currentTimeMillis();
while (eventIterator.hasNext()) {
OMDBUpdateEvent<String, Object> omdbUpdateEvent = eventIterator.next();
String tableName = omdbUpdateEvent.getTable();
@@ -208,7 +209,8 @@ public class OmTableInsightTask implements ReconOmTask {
if (!replicatedSizeMap.isEmpty()) {
writeDataToDB(replicatedSizeMap);
}
- LOG.debug("Completed a 'process' run of OmTableInsightTask.");
+ LOG.debug("{} successfully processed in {} milliseconds",
+ getTaskName(), (System.currentTimeMillis() - startTime));
return new ImmutablePair<>(getTaskName(), true);
}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskController.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskController.java
index 1a514ceb90..8e0c6f4b1c 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskController.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskController.java
@@ -22,7 +22,6 @@ import java.util.Map;
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
-import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
/**
* Controller used by Recon to manage Tasks that are waiting on Recon events.
@@ -38,18 +37,15 @@ public interface ReconTaskController {
/**
* Pass on a set of OM DB update events to the registered tasks.
* @param events set of events
- * @throws InterruptedException InterruptedException
*/
void consumeOMEvents(OMUpdateEventBatch events,
- OMMetadataManager omMetadataManager)
- throws InterruptedException;
+ OMMetadataManager omMetadataManager);
/**
* Pass on the handle to a new OM DB instance to the registered tasks.
* @param omMetadataManager OM Metadata Manager instance
*/
- void reInitializeTasks(ReconOMMetadataManager omMetadataManager)
- throws InterruptedException;
+ void reInitializeTasks(ReconOMMetadataManager omMetadataManager);
/**
* Get set of registered tasks.
@@ -57,12 +53,6 @@ public interface ReconTaskController {
*/
Map<String, ReconOmTask> getRegisteredTasks();
- /**
- * Get instance of ReconTaskStatusDao.
- * @return instance of ReconTaskStatusDao
- */
- ReconTaskStatusDao getReconTaskStatusDao();
-
/**
* Start the task scheduler.
*/
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java
index 6440509585..32b440badb 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java
@@ -28,19 +28,23 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
-import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
-import org.hadoop.ozone.recon.schema.tables.pojos.ReconTaskStatus;
+import org.apache.hadoop.ozone.recon.tasks.types.NamedCallableTask;
+import org.apache.hadoop.ozone.recon.tasks.types.TaskExecutionException;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -59,16 +63,16 @@ public class ReconTaskControllerImpl implements ReconTaskController {
private final int threadCount;
private Map<String, AtomicInteger> taskFailureCounter = new HashMap<>();
private static final int TASK_FAILURE_THRESHOLD = 2;
- private ReconTaskStatusDao reconTaskStatusDao;
+ private final ReconTaskStatusUpdaterManager taskStatusUpdaterManager;
@Inject
public ReconTaskControllerImpl(OzoneConfiguration configuration,
- ReconTaskStatusDao reconTaskStatusDao,
- Set<ReconOmTask> tasks) {
+ Set<ReconOmTask> tasks,
+ ReconTaskStatusUpdaterManager taskStatusUpdaterManager) {
reconOmTasks = new HashMap<>();
threadCount = configuration.getInt(OZONE_RECON_TASK_THREAD_COUNT_KEY,
OZONE_RECON_TASK_THREAD_COUNT_DEFAULT);
- this.reconTaskStatusDao = reconTaskStatusDao;
+ this.taskStatusUpdaterManager = taskStatusUpdaterManager;
for (ReconOmTask task : tasks) {
registerTask(task);
}
@@ -83,12 +87,6 @@ public class ReconTaskControllerImpl implements ReconTaskController {
reconOmTasks.put(taskName, task);
// Store Task in Task failure tracker.
taskFailureCounter.put(taskName, new AtomicInteger(0));
- // Create DB record for the task.
- ReconTaskStatus reconTaskStatusRecord = new ReconTaskStatus(taskName,
- 0L, 0L);
- if (!reconTaskStatusDao.existsById(taskName)) {
- reconTaskStatusDao.insert(reconTaskStatusRecord);
- }
}
/**
@@ -97,55 +95,46 @@ public class ReconTaskControllerImpl implements ReconTaskController {
* reprocess call more than 2 times across events, it is unregistered
* (ignored).
* @param events set of events
- * @throws InterruptedException
*/
@Override
- public synchronized void consumeOMEvents(OMUpdateEventBatch events,
- OMMetadataManager omMetadataManager)
- throws InterruptedException {
+ public synchronized void consumeOMEvents(OMUpdateEventBatch events, OMMetadataManager omMetadataManager) {
+ if (!events.isEmpty()) {
+ Collection<NamedCallableTask<Pair<String, Boolean>>> tasks = new ArrayList<>();
+ List<String> failedTasks = new ArrayList<>();
+ for (Map.Entry<String, ReconOmTask> taskEntry :
+ reconOmTasks.entrySet()) {
+ ReconOmTask task = taskEntry.getValue();
+ ReconTaskStatusUpdater taskStatusUpdater = taskStatusUpdaterManager.getTaskStatusUpdater(task.getTaskName());
+ taskStatusUpdater.recordRunStart();
+ // events passed to process method is no longer filtered
+ tasks.add(new NamedCallableTask<>(task.getTaskName(), () -> task.process(events)));
+ }
+ processTasks(tasks, events, failedTasks);
- try {
- if (!events.isEmpty()) {
- Collection<Callable<Pair<String, Boolean>>> tasks = new ArrayList<>();
- for (Map.Entry<String, ReconOmTask> taskEntry :
- reconOmTasks.entrySet()) {
- ReconOmTask task = taskEntry.getValue();
+ // Retry processing failed tasks
+ List<String> retryFailedTasks = new ArrayList<>();
+ if (!failedTasks.isEmpty()) {
+ tasks.clear();
+ for (String taskName : failedTasks) {
+ ReconOmTask task = reconOmTasks.get(taskName);
// events passed to process method is no longer filtered
- tasks.add(() -> task.process(events));
- }
-
- List<Future<Pair<String, Boolean>>> results =
- executorService.invokeAll(tasks);
- List<String> failedTasks = processTaskResults(results, events);
-
- // Retry
- List<String> retryFailedTasks = new ArrayList<>();
- if (!failedTasks.isEmpty()) {
- tasks.clear();
- for (String taskName : failedTasks) {
- ReconOmTask task = reconOmTasks.get(taskName);
- // events passed to process method is no longer filtered
- tasks.add(() -> task.process(events));
- }
- results = executorService.invokeAll(tasks);
- retryFailedTasks = processTaskResults(results, events);
+ tasks.add(new NamedCallableTask<>(task.getTaskName(),
+ () -> task.process(events)));
}
+ processTasks(tasks, events, retryFailedTasks);
+ }
- // Reprocess the failed tasks.
- if (!retryFailedTasks.isEmpty()) {
- tasks.clear();
- for (String taskName : failedTasks) {
- ReconOmTask task = reconOmTasks.get(taskName);
- tasks.add(() -> task.reprocess(omMetadataManager));
- }
- results = executorService.invokeAll(tasks);
- List<String> reprocessFailedTasks =
- processTaskResults(results, events);
- ignoreFailedTasks(reprocessFailedTasks);
+ // Reprocess the failed tasks.
+ if (!retryFailedTasks.isEmpty()) {
+ tasks.clear();
+ for (String taskName : failedTasks) {
+ ReconOmTask task = reconOmTasks.get(taskName);
+ tasks.add(new NamedCallableTask<>(task.getTaskName(), () -> task.reprocess(omMetadataManager)));
}
+ List<String> reprocessFailedTasks = new ArrayList<>();
+ processTasks(tasks, events, reprocessFailedTasks);
+ ignoreFailedTasks(reprocessFailedTasks);
}
- } catch (ExecutionException e) {
- LOG.error("Unexpected error : ", e);
}
}
@@ -166,45 +155,57 @@ public class ReconTaskControllerImpl implements ReconTaskController {
}
@Override
- public synchronized void reInitializeTasks(
- ReconOMMetadataManager omMetadataManager) throws InterruptedException {
- try {
- Collection<Callable<Pair<String, Boolean>>> tasks = new ArrayList<>();
- for (Map.Entry<String, ReconOmTask> taskEntry :
- reconOmTasks.entrySet()) {
- ReconOmTask task = taskEntry.getValue();
- tasks.add(() -> task.reprocess(omMetadataManager));
- }
- List<Future<Pair<String, Boolean>>> results =
- executorService.invokeAll(tasks);
- for (Future<Pair<String, Boolean>> f : results) {
- String taskName = f.get().getLeft();
- if (!f.get().getRight()) {
- LOG.info("Init failed for task {}.", taskName);
- } else {
- //store the timestamp for the task
- ReconTaskStatus reconTaskStatusRecord = new ReconTaskStatus(taskName,
- System.currentTimeMillis(),
- omMetadataManager.getLastSequenceNumberFromDB());
- reconTaskStatusDao.update(reconTaskStatusRecord);
- }
- }
- } catch (ExecutionException e) {
- LOG.error("Unexpected error : ", e);
+ public synchronized void reInitializeTasks(ReconOMMetadataManager omMetadataManager) {
+ Collection<NamedCallableTask<Pair<String, Boolean>>> tasks = new ArrayList<>();
+ for (Map.Entry<String, ReconOmTask> taskEntry :
+ reconOmTasks.entrySet()) {
+ ReconOmTask task = taskEntry.getValue();
+ ReconTaskStatusUpdater taskStatusUpdater = taskStatusUpdaterManager.getTaskStatusUpdater(task.getTaskName());
+ taskStatusUpdater.recordRunStart();
+ tasks.add(new NamedCallableTask<>(task.getTaskName(), () -> task.reprocess(omMetadataManager)));
}
- }
- /**
- * Store the last completed event sequence number and timestamp to the DB
- * for that task.
- * @param taskName taskname to be updated.
- * @param lastSequenceNumber contains the new sequence number.
- */
- private void storeLastCompletedTransaction(
- String taskName, long lastSequenceNumber) {
- ReconTaskStatus reconTaskStatusRecord = new ReconTaskStatus(taskName,
- System.currentTimeMillis(), lastSequenceNumber);
- reconTaskStatusDao.update(reconTaskStatusRecord);
+ try {
+ CompletableFuture.allOf(tasks.stream()
+ .map(task -> CompletableFuture.supplyAsync(() -> {
+ try {
+ return task.call();
+ } catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ // Wrap the exception with the task name
+ throw new TaskExecutionException(task.getTaskName(), e);
+ }
+ }, executorService).thenAccept(result -> {
+ String taskName = result.getLeft();
+ ReconTaskStatusUpdater taskStatusUpdater = taskStatusUpdaterManager.getTaskStatusUpdater(taskName);
+ if (!result.getRight()) {
+ LOG.error("Init failed for task {}.", taskName);
+ taskStatusUpdater.setLastTaskRunStatus(-1);
+ } else {
+ taskStatusUpdater.setLastTaskRunStatus(0);
+ taskStatusUpdater.setLastUpdatedSeqNumber(omMetadataManager.getLastSequenceNumberFromDB());
+ }
+ taskStatusUpdater.recordRunCompletion();
+ }).exceptionally(ex -> {
+ LOG.error("Task failed with exception: ", ex);
+ if (ex.getCause() instanceof TaskExecutionException) {
+ TaskExecutionException taskEx = (TaskExecutionException) ex.getCause();
+ String taskName = taskEx.getTaskName();
+ LOG.error("The above error occurred while trying to execute task: {}", taskName);
+
+ ReconTaskStatusUpdater taskStatusUpdater = taskStatusUpdaterManager.getTaskStatusUpdater(taskName);
+ taskStatusUpdater.setLastTaskRunStatus(-1);
+ taskStatusUpdater.recordRunCompletion();
+ }
+ return null;
+ })).toArray(CompletableFuture[]::new)).join();
+ } catch (CompletionException ce) {
+ LOG.error("Completing all tasks failed with exception ", ce);
+ } catch (CancellationException ce) {
+ LOG.error("Some tasks were cancelled with exception", ce);
+ }
}
@Override
@@ -212,11 +213,6 @@ public class ReconTaskControllerImpl implements ReconTaskController {
return reconOmTasks;
}
- @Override
- public ReconTaskStatusDao getReconTaskStatusDao() {
- return reconTaskStatusDao;
- }
-
@Override
public synchronized void start() {
LOG.info("Starting Recon Task Controller.");
@@ -234,28 +230,59 @@ public class ReconTaskControllerImpl implements ReconTaskController {
}
/**
- * Wait on results of all tasks.
- * @param results Set of Futures.
- * @param events Events.
- * @return List of failed task names
- * @throws ExecutionException execution Exception
- * @throws InterruptedException Interrupted Exception
+ * For a given list of {@link Callable} tasks process them and add any failed task to the provided list.
+ * The tasks are executed in parallel, but will wait for the tasks to complete i.e. the longest
+ * time taken by this method will be the time taken by the longest task in the list.
+ * @param tasks A list of tasks to execute.
+ * @param events A batch of {@link OMUpdateEventBatch} events to fetch sequence number of last event in batch.
+ * @param failedTasks Reference of the list to which we want to add the failed tasks for retry/reprocessing
*/
- private List<String> processTaskResults(List<Future<Pair<String, Boolean>>>
- results,
- OMUpdateEventBatch events)
- throws ExecutionException, InterruptedException {
- List<String> failedTasks = new ArrayList<>();
- for (Future<Pair<String, Boolean>> f : results) {
- String taskName = f.get().getLeft();
- if (!f.get().getRight()) {
- LOG.info("Failed task : {}", taskName);
- failedTasks.add(f.get().getLeft());
- } else {
- taskFailureCounter.get(taskName).set(0);
- storeLastCompletedTransaction(taskName, events.getLastSequenceNumber());
- }
+ private void processTasks(Collection<NamedCallableTask<Pair<String, Boolean>>> tasks,
+ OMUpdateEventBatch events, List<String> failedTasks) {
+ List<CompletableFuture<Void>> futures = tasks.stream()
+ .map(task -> CompletableFuture.supplyAsync(() -> {
+ try {
+ return task.call();
+ } catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ // Wrap the exception with the task name
+ throw new TaskExecutionException(task.getTaskName(), e);
+ }
+ }, executorService).thenAccept(result -> {
+ String taskName = result.getLeft();
+ ReconTaskStatusUpdater taskStatusUpdater = taskStatusUpdaterManager.getTaskStatusUpdater(taskName);
+ if (!result.getRight()) {
+ LOG.error("Task {} failed", taskName);
+ failedTasks.add(result.getLeft());
+ taskStatusUpdater.setLastTaskRunStatus(-1);
+ } else {
+ taskFailureCounter.get(taskName).set(0);
+ taskStatusUpdater.setLastTaskRunStatus(0);
+ taskStatusUpdater.setLastUpdatedSeqNumber(events.getLastSequenceNumber());
+ }
+ taskStatusUpdater.recordRunCompletion();
+ }).exceptionally(ex -> {
+ LOG.error("Task failed with exception: ", ex);
+ if (ex.getCause() instanceof TaskExecutionException) {
+ TaskExecutionException taskEx = (TaskExecutionException) ex.getCause();
+ String taskName = taskEx.getTaskName();
+ LOG.error("The above error occurred while trying to execute task: {}", taskName);
+
+ ReconTaskStatusUpdater taskStatusUpdater = taskStatusUpdaterManager.getTaskStatusUpdater(taskName);
+ taskStatusUpdater.setLastTaskRunStatus(-1);
+ taskStatusUpdater.recordRunCompletion();
+ }
+ return null;
+ })).collect(Collectors.toList());
+
+ try {
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+ } catch (CompletionException ce) {
+ LOG.error("Completing all tasks failed with exception ", ce);
+ } catch (CancellationException ce) {
+ LOG.error("Some tasks were cancelled with exception", ce);
}
- return failedTasks;
}
}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/types/NamedCallableTask.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/types/NamedCallableTask.java
new file mode 100644
index 0000000000..f91ab5255d
--- /dev/null
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/types/NamedCallableTask.java
@@ -0,0 +1,49 @@
+/*
+ * 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.recon.tasks.types;
+
+import java.util.concurrent.Callable;
+
+/**
+ * This class is a wrapper over the {@link java.util.concurrent.Callable} interface.
+ * <br>
+ * If there is any runtime exception which occurs during execution of a task via the executor,
+ * we lose all data regarding the task, so we do not have information on which task actually failed.
+ * <br>
+ * This class is useful, as it will associate a background task to a task name so that we can efficiently
+ * record any failure.
+ */
+public class NamedCallableTask<V> implements Callable<V> {
+ private final String taskName;
+ private final Callable<V> task;
+
+ public NamedCallableTask(String taskName, Callable<V> task) {
+ this.taskName = taskName;
+ this.task = task;
+ }
+
+ public String getTaskName() {
+ return this.taskName;
+ }
+
+ @Override
+ public V call() throws Exception {
+ return task.call();
+ }
+}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/types/TaskExecutionException.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/types/TaskExecutionException.java
new file mode 100644
index 0000000000..a83e643c5b
--- /dev/null
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/types/TaskExecutionException.java
@@ -0,0 +1,35 @@
+/*
+ * 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.recon.tasks.types;
+
+/**
+ * Wrapper over {@link RuntimeException} to associate an exception to a task name.
+ */
+public class TaskExecutionException extends RuntimeException {
+ private final String taskName;
+
+ public TaskExecutionException(String taskName, Throwable cause) {
+ super(cause);
+ this.taskName = taskName;
+ }
+
+ public String getTaskName() {
+ return this.taskName;
+ }
+}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/types/package-info.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/types/package-info.java
new file mode 100644
index 0000000000..68f107ddf1
--- /dev/null
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/types/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+/**
+ * The classes in this package contains a few Wrapper classes for better task tracking.
+ */
+
+package org.apache.hadoop.ozone.recon.tasks.types;
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/updater/ReconTaskStatusUpdater.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/updater/ReconTaskStatusUpdater.java
new file mode 100644
index 0000000000..8708fe5bc3
--- /dev/null
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/updater/ReconTaskStatusUpdater.java
@@ -0,0 +1,123 @@
+/*
+ * 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.recon.tasks.updater;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
+import org.hadoop.ozone.recon.schema.tables.pojos.ReconTaskStatus;
+import org.jooq.exception.DataAccessException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class provides utilities to update/modify Recon Task related data
+ * like updating table, incrementing counter etc.
+ */
+public class ReconTaskStatusUpdater {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ReconTaskStatusUpdater.class);
+
+ private ReconTaskStatus reconTaskStatus;
+
+ private ReconTaskStatusDao reconTaskStatusDao;
+
+ private String taskName;
+
+ @VisibleForTesting
+ public ReconTaskStatusUpdater(ReconTaskStatusDao reconTaskStatusDao,
+ String taskName) {
+ this.taskName = taskName;
+ this.reconTaskStatusDao = reconTaskStatusDao;
+ this.reconTaskStatus = new ReconTaskStatus(taskName, 0L, 0L, 0, 0);
+ }
+
+ public ReconTaskStatusUpdater(ReconTaskStatusDao reconTaskStatusDao, ReconTaskStatus task) {
+ this.taskName = task.getTaskName();
+ this.reconTaskStatusDao = reconTaskStatusDao;
+ this.reconTaskStatus = new ReconTaskStatus(taskName, task.getLastUpdatedTimestamp(),
+ task.getLastUpdatedSeqNumber(), task.getLastTaskRunStatus(), task.getIsCurrentTaskRunning());
+ }
+
+ public void setTaskName(String taskName) {
+ this.taskName = taskName;
+ this.reconTaskStatus.setTaskName(taskName);
+ }
+
+ public void setLastUpdatedSeqNumber(long lastUpdatedSeqNumber) {
+ this.reconTaskStatus.setLastUpdatedSeqNumber(lastUpdatedSeqNumber);
+ }
+
+ public void setLastUpdatedTimestamp(long lastUpdatedTimestamp) {
+ this.reconTaskStatus.setLastUpdatedTimestamp(lastUpdatedTimestamp);
+ }
+
+ public void setLastTaskRunStatus(int lastTaskRunStatus) {
+ this.reconTaskStatus.setLastTaskRunStatus(lastTaskRunStatus);
+ }
+
+ public void setIsCurrentTaskRunning(int isCurrentTaskRunning) {
+ this.reconTaskStatus.setIsCurrentTaskRunning(isCurrentTaskRunning);
+ }
+
+ /**
+ * Helper function to update TASK_STATUS table with task start values.
+ * Set the isCurrentTaskRunning as true, update the timestamp.
+ * Call this function before the actual task processing starts to update table in DB.
+ */
+ public void recordRunStart() {
+ try {
+ this.reconTaskStatus.setIsCurrentTaskRunning(1);
+ this.reconTaskStatus.setLastUpdatedTimestamp(System.currentTimeMillis());
+ updateDetails();
+ } catch (DataAccessException e) {
+ LOG.error("Failed to update table for start of task: {}", this.reconTaskStatus.getTaskName());
+ }
+ }
+
+ /**
+ * Helper function to update TASK_STATUS table with task end values.
+ * Set isCurrentTaskRunning as false, update the timestamp.
+ * Call this function after the actual task processing ends to update table in DB.
+ * It is expected that the task status result (successful/0, failure/-1) is already set
+ * before calling.
+ */
+ public void recordRunCompletion() {
+ try {
+ this.reconTaskStatus.setIsCurrentTaskRunning(0);
+ this.reconTaskStatus.setLastUpdatedTimestamp(System.currentTimeMillis());
+ updateDetails();
+ } catch (DataAccessException e) {
+ LOG.error("Failed to update table for task: {}", this.reconTaskStatus.getTaskName());
+ }
+ }
+
+ /**
+ * Utility function to update table with task details and update the counter if needed.
+ */
+ public void updateDetails() {
+ if (!reconTaskStatusDao.existsById(this.taskName)) {
+ // First time getting the task, so insert value
+ reconTaskStatusDao.insert(this.reconTaskStatus);
+ LOG.info("Registered Task: {}", this.taskName);
+ } else {
+ // We already have row for the task in the table, update the row
+ reconTaskStatusDao.update(this.reconTaskStatus);
+ }
+ }
+}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/updater/ReconTaskStatusUpdaterManager.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/updater/ReconTaskStatusUpdaterManager.java
new file mode 100644
index 0000000000..1e7ee75f24
--- /dev/null
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/updater/ReconTaskStatusUpdaterManager.java
@@ -0,0 +1,69 @@
+/*
+ * 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.recon.tasks.updater;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
+import org.hadoop.ozone.recon.schema.tables.pojos.ReconTaskStatus;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * This class provides caching for ReconTaskStatusUpdater instances.
+ * For each task we maintain a map of updater instance and provide it to consumers
+ * to update.
+ * Here we also make a single call to the TASK_STATUS_TABLE to check if previous values are present
+ * for a task in the DB to avoid overwrite to initial state
+ */
+@Singleton
+public class ReconTaskStatusUpdaterManager {
+ private final ReconTaskStatusDao reconTaskStatusDao;
+ // Act as a cache for the task updater instancesF
+ private final Map<String, ReconTaskStatusUpdater> updaterCache;
+
+ @Inject
+ public ReconTaskStatusUpdaterManager(
+ ReconTaskStatusDao reconTaskStatusDao
+ ) {
+ this.reconTaskStatusDao = reconTaskStatusDao;
+ this.updaterCache = new ConcurrentHashMap<>();
+
+ // Fetch the tasks present in the DB already
+ List<ReconTaskStatus> tasks = reconTaskStatusDao.findAll();
+ for (ReconTaskStatus task: tasks) {
+ updaterCache.put(task.getTaskName(),
+ new ReconTaskStatusUpdater(reconTaskStatusDao, task));
+ }
+ }
+
+ /**
+ * Gets the updater for the provided task name and updates DB with initial values
+ * if the task is not already present in DB.
+ * @param taskName The name of the task for which we want to get instance of the updater
+ * @return An instance of {@link ReconTaskStatusUpdater} for the provided task name.
+ */
+ public ReconTaskStatusUpdater getTaskStatusUpdater(String taskName) {
+ // If the task is not already present in the DB then we can initialize using initial values
+ return updaterCache.computeIfAbsent(taskName, (name) ->
+ new ReconTaskStatusUpdater(reconTaskStatusDao, name));
+ }
+}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/updater/package-info.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/updater/package-info.java
new file mode 100644
index 0000000000..1aeedb88a9
--- /dev/null
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/updater/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * The classes in this package contains the utility methods and classes to update various task status.
+ */
+package org.apache.hadoop.ozone.recon.tasks.updater;
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconLayoutFeature.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconLayoutFeature.java
index 52739efe1a..0539323791 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconLayoutFeature.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconLayoutFeature.java
@@ -32,7 +32,8 @@ import java.util.Set;
*/
public enum ReconLayoutFeature {
// Represents the starting point for Recon's layout versioning system.
- INITIAL_VERSION(0, "Recon Layout Versioning Introduction");
+ INITIAL_VERSION(0, "Recon Layout Versioning Introduction"),
+ TASK_STATUS_STATISTICS(1, "Recon Task Status Statistics Tracking Introduced");
private final int version;
private final String description;
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconTaskStatusTableUpgradeAction.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconTaskStatusTableUpgradeAction.java
new file mode 100644
index 0000000000..f613b21672
--- /dev/null
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconTaskStatusTableUpgradeAction.java
@@ -0,0 +1,104 @@
+/*
+ * 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.recon.upgrade;
+
+import org.apache.hadoop.ozone.recon.scm.ReconStorageContainerManagerFacade;
+import org.jooq.DSLContext;
+import org.jooq.exception.DataAccessException;
+import org.jooq.impl.DSL;
+import org.jooq.impl.SQLDataType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.SQLException;
+
+import static org.hadoop.ozone.recon.codegen.SqlDbUtils.TABLE_EXISTS_CHECK;
+import static org.hadoop.ozone.recon.schema.ReconTaskSchemaDefinition.RECON_TASK_STATUS_TABLE_NAME;
+
+
+/**
+ * Upgrade action for TASK_STATUS_STATISTICS feature layout change, which adds
+ * <code>last_task_run_status</code> and <code>current_task_run_status</code> columns to
+ * {@link org.hadoop.ozone.recon.schema.ReconTaskSchemaDefinition} in case it is missing .
+ */
+@UpgradeActionRecon(feature = ReconLayoutFeature.TASK_STATUS_STATISTICS,
+ type = ReconUpgradeAction.UpgradeActionType.FINALIZE)
+public class ReconTaskStatusTableUpgradeAction implements ReconUpgradeAction {
+
+ public static final Logger LOG = LoggerFactory.getLogger(ReconTaskStatusTableUpgradeAction.class);
+
+ /**
+ * Utility function to add provided column to RECON_TASK_STATUS table as INTEGER type.
+ * @param dslContext Stores {@link DSLContext} to perform alter operations
+ * @param columnName Name of the column to be inserted to the table
+ */
+ private void addColumnToTable(DSLContext dslContext, String columnName) {
+ //Column is set as nullable to avoid any errors.
+ dslContext.alterTable(RECON_TASK_STATUS_TABLE_NAME)
+ .addColumn(columnName, SQLDataType.INTEGER.nullable(true)).execute();
+ }
+
+ /**
+ * Utility function to set the provided column as Non-Null to enforce constraints in RECON_TASK_STATUS table.
+ * @param dslContext Stores {@link DSLContext} to perform alter operations
+ * @param columnName Name of the column to set as non-null
+ */
+ private void setColumnAsNonNullable(DSLContext dslContext, String columnName) {
+ dslContext.alterTable(RECON_TASK_STATUS_TABLE_NAME)
+ .alterColumn(columnName).setNotNull().execute();
+ }
+
+ @Override
+ public void execute(ReconStorageContainerManagerFacade scmFacade) throws DataAccessException {
+ DataSource dataSource = scmFacade.getDataSource();
+ try (Connection conn = dataSource.getConnection()) {
+ if (!TABLE_EXISTS_CHECK.test(conn, RECON_TASK_STATUS_TABLE_NAME)) {
+ return;
+ }
+
+ DSLContext dslContext = DSL.using(conn);
+ // JOOQ doesn't support Derby DB officially, there is no way to run 'ADD COLUMN' command in single call
+ // for multiple columns. Hence, we run it as two separate steps.
+ LOG.info("Adding 'last_task_run_status' column to task status table");
+ addColumnToTable(dslContext, "last_task_run_status");
+ LOG.info("Adding 'is_current_task_running' column to task status table");
+ addColumnToTable(dslContext, "is_current_task_running");
+
+ //Handle previous table values with new columns default values
+ int updatedRowCount = dslContext.update(DSL.table(RECON_TASK_STATUS_TABLE_NAME))
+ .set(DSL.field("last_task_run_status", SQLDataType.INTEGER), 0)
+ .set(DSL.field("is_current_task_running", SQLDataType.INTEGER), 0)
+ .execute();
+ LOG.info("Updated {} rows with default value for new columns", updatedRowCount);
+
+ // Now we will set the column as not-null to enforce constraints
+ setColumnAsNonNullable(dslContext, "last_task_run_status");
+ setColumnAsNonNullable(dslContext, "is_current_task_running");
+ } catch (SQLException | DataAccessException ex) {
+ LOG.error("Error while upgrading RECON_TASK_STATUS table.", ex);
+ }
+ }
+
+ @Override public UpgradeActionType getType() {
+ return UpgradeActionType.FINALIZE;
+ }
+}
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestEndpoints.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestEndpoints.java
index f1dafa2c75..5b6d22bf26 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestEndpoints.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestEndpoints.java
@@ -942,7 +942,7 @@ public class TestEndpoints extends AbstractReconSqlDBTest {
containers.add(omContainerInfo2);
// Process the list of containers through the container size count task
- containerSizeCountTask.process(containers);
+ containerSizeCountTask.processContainers(containers);
// Test fetching all container counts
Response response = utilizationEndpoint.getContainerCounts(0L);
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestTaskStatusService.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestTaskStatusService.java
index 741dcf3be4..b75fde08c4 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestTaskStatusService.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestTaskStatusService.java
@@ -18,21 +18,23 @@
package org.apache.hadoop.ozone.recon.api;
-import static org.junit.jupiter.api.Assertions.assertEquals;
import com.google.inject.AbstractModule;
import com.google.inject.Injector;
-
import org.apache.hadoop.ozone.recon.persistence.AbstractReconSqlDBTest;
import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
import org.hadoop.ozone.recon.schema.tables.pojos.ReconTaskStatus;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import javax.ws.rs.core.Response;
import java.util.ArrayList;
import java.util.List;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+
/**
* Test for Task Status Service.
*/
@@ -44,7 +46,7 @@ public class TestTaskStatusService extends AbstractReconSqlDBTest {
}
@BeforeEach
- public void setUp() {
+ public void setUp() throws Exception {
Injector parentInjector = getInjector();
parentInjector.createChildInjector(new AbstractModule() {
@Override
@@ -55,18 +57,19 @@ public class TestTaskStatusService extends AbstractReconSqlDBTest {
});
}
- @Test
- public void testGetTaskTimes() {
+ @ParameterizedTest
+ @ValueSource(ints = {0, 1, -1})
+ public void testTaskTableValues(int lastTaskRunStatus) {
ReconTaskStatusDao reconTaskStatusDao = getDao(ReconTaskStatusDao.class);
ReconTaskStatus reconTaskStatusRecord = new ReconTaskStatus(
- "Dummy_Task", System.currentTimeMillis(), 0L);
+ "Dummy_Task", System.currentTimeMillis(), 0L, lastTaskRunStatus, 0);
reconTaskStatusDao.insert(reconTaskStatusRecord);
List<ReconTaskStatus> resultList = new ArrayList<>();
resultList.add(reconTaskStatusRecord);
- Response response = taskStatusService.getTaskTimes();
+ Response response = taskStatusService.getTaskStats();
List<ReconTaskStatus> responseList = (List<ReconTaskStatus>)
response.getEntity();
@@ -76,6 +79,8 @@ public class TestTaskStatusService extends AbstractReconSqlDBTest {
assertEquals(reconTaskStatusRecord.getTaskName(), r.getTaskName());
assertEquals(reconTaskStatusRecord.getLastUpdatedTimestamp(),
r.getLastUpdatedTimestamp());
+ assertEquals(reconTaskStatusRecord.getLastTaskRunStatus(), r.getLastTaskRunStatus());
+ assertEquals(reconTaskStatusRecord.getIsCurrentTaskRunning(), r.getIsCurrentTaskRunning());
}
}
}
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestTriggerDBSyncEndpoint.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestTriggerDBSyncEndpoint.java
index 7ad5dcc58c..480d1348c1 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestTriggerDBSyncEndpoint.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestTriggerDBSyncEndpoint.java
@@ -38,6 +38,8 @@ import org.apache.hadoop.ozone.recon.spi.StorageContainerServiceProvider;
import org.apache.hadoop.ozone.recon.spi.impl.OzoneManagerServiceProviderImpl;
import org.apache.hadoop.ozone.recon.spi.impl.StorageContainerServiceProviderImpl;
import org.apache.hadoop.ozone.recon.tasks.ReconTaskController;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.apache.hadoop.security.authentication.client.AuthenticationException;
import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
import org.junit.jupiter.api.BeforeEach;
@@ -102,6 +104,12 @@ public class TestTriggerDBSyncEndpoint {
ReconUtils reconUtilsMock = mock(ReconUtils.class);
+
+ ReconTaskStatusDao reconTaskStatusDaoMock = mock(ReconTaskStatusDao.class);
+ ReconTaskStatusUpdaterManager taskStatusUpdaterManagerMock = mock(ReconTaskStatusUpdaterManager.class);
+ when(taskStatusUpdaterManagerMock.getTaskStatusUpdater(anyString())).thenReturn(new ReconTaskStatusUpdater(
+ reconTaskStatusDaoMock, "dummyTaskManager"));
+
DBCheckpoint checkpoint = omMetadataManager.getStore()
.getCheckpoint(true);
File tarFile = createTarFile(checkpoint.getCheckpointLocation());
@@ -116,13 +124,11 @@ public class TestTriggerDBSyncEndpoint {
commonUtils.getReconNodeDetails());
ReconTaskController reconTaskController = mock(ReconTaskController.class);
- when(reconTaskController.getReconTaskStatusDao())
- .thenReturn(mock(ReconTaskStatusDao.class));
-
OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
new OzoneManagerServiceProviderImpl(configuration,
- reconOMMetadataManager, reconTaskController, reconUtilsMock,
- ozoneManagerProtocol, new ReconContext(configuration, reconUtilsMock));
+ reconOMMetadataManager, reconTaskController,
+ reconUtilsMock, ozoneManagerProtocol, new ReconContext(configuration, reconUtilsMock),
+ taskStatusUpdaterManagerMock);
ozoneManagerServiceProvider.start();
reconTestInjector =
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/fsck/TestContainerHealthTask.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/fsck/TestContainerHealthTask.java
index 46e4506a5e..8c8b72ea45 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/fsck/TestContainerHealthTask.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/fsck/TestContainerHealthTask.java
@@ -26,6 +26,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.any;
@@ -65,6 +66,8 @@ import org.apache.hadoop.ozone.recon.scm.ReconStorageContainerManagerFacade;
import org.apache.hadoop.ozone.recon.spi.ReconContainerMetadataManager;
import org.apache.hadoop.ozone.recon.spi.StorageContainerServiceProvider;
import org.apache.hadoop.ozone.recon.tasks.ReconTaskConfig;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.apache.ozone.test.LambdaTestUtils;
import org.hadoop.ozone.recon.schema.ContainerSchemaDefinition;
import org.apache.hadoop.ozone.recon.persistence.AbstractReconSqlDBTest;
@@ -189,10 +192,9 @@ public class TestContainerHealthTask extends AbstractReconSqlDBTest {
// Start container health task
ContainerHealthTask containerHealthTask =
new ContainerHealthTask(scmMock.getContainerManager(),
- scmMock.getScmServiceProvider(),
- reconTaskStatusDao, containerHealthSchemaManager,
- placementMock, reconTaskConfig,
- reconContainerMetadataManager, new OzoneConfiguration());
+ scmMock.getScmServiceProvider(), containerHealthSchemaManager,
+ placementMock, reconTaskConfig, reconContainerMetadataManager,
+ new OzoneConfiguration(), getMockTaskStatusUpdaterManager());
containerHealthTask.start();
// Ensure unhealthy container count in DB matches expected
@@ -362,10 +364,9 @@ public class TestContainerHealthTask extends AbstractReconSqlDBTest {
1L)).thenReturn(5L);
ContainerHealthTask containerHealthTask =
new ContainerHealthTask(scmMock.getContainerManager(),
- scmMock.getScmServiceProvider(),
- reconTaskStatusDao, containerHealthSchemaManager,
- placementMock, reconTaskConfig,
- reconContainerMetadataManager, new OzoneConfiguration());
+ scmMock.getScmServiceProvider(), containerHealthSchemaManager,
+ placementMock, reconTaskConfig, reconContainerMetadataManager,
+ new OzoneConfiguration(), getMockTaskStatusUpdaterManager());
containerHealthTask.start();
LambdaTestUtils.await(6000, 1000, () ->
(unHealthyContainersTableHandle.count() == 1));
@@ -543,15 +544,13 @@ public class TestContainerHealthTask extends AbstractReconSqlDBTest {
when(reconContainerMetadataManager.getKeyCountForContainer(2L)).thenReturn(0L);
// Start the container health task
- ReconTaskStatusDao reconTaskStatusDao = getDao(ReconTaskStatusDao.class);
ReconTaskConfig reconTaskConfig = new ReconTaskConfig();
reconTaskConfig.setMissingContainerTaskInterval(Duration.ofSeconds(2));
ContainerHealthTask containerHealthTask =
new ContainerHealthTask(scmMock.getContainerManager(),
- scmMock.getScmServiceProvider(),
- reconTaskStatusDao, containerHealthSchemaManager,
- placementMock, reconTaskConfig,
- reconContainerMetadataManager, new OzoneConfiguration());
+ scmMock.getScmServiceProvider(), containerHealthSchemaManager,
+ placementMock, reconTaskConfig, reconContainerMetadataManager,
+ new OzoneConfiguration(), getMockTaskStatusUpdaterManager());
containerHealthTask.start();
@@ -566,6 +565,15 @@ public class TestContainerHealthTask extends AbstractReconSqlDBTest {
});
}
+ private ReconTaskStatusUpdaterManager getMockTaskStatusUpdaterManager() {
+ ReconTaskStatusUpdaterManager reconTaskStatusUpdaterManager = mock(ReconTaskStatusUpdaterManager.class);
+ when(reconTaskStatusUpdaterManager.getTaskStatusUpdater(anyString())).thenAnswer(inv -> {
+ String taskName = inv.getArgument(0);
+ return new ReconTaskStatusUpdater(getDao(ReconTaskStatusDao.class), taskName);
+ });
+ return reconTaskStatusUpdaterManager;
+ }
+
private Set<ContainerReplica> getMockReplicas(
long containerId, State...states) {
Set<ContainerReplica> replicas = new HashSet<>();
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/TestReconInternalSchemaDefinition.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/TestReconInternalSchemaDefinition.java
index 1230a9ba54..5cc2ec0c1a 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/TestReconInternalSchemaDefinition.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/TestReconInternalSchemaDefinition.java
@@ -60,6 +60,10 @@ public class TestReconInternalSchemaDefinition extends AbstractReconSqlDBTest {
Types.BIGINT));
expectedPairs.add(new ImmutablePair<>("last_updated_seq_number",
Types.BIGINT));
+ expectedPairs.add(new ImmutablePair<>("last_task_run_status",
+ Types.INTEGER));
+ expectedPairs.add(new ImmutablePair<>("is_current_task_running",
+ Types.INTEGER));
List<Pair<String, Integer>> actualPairs = new ArrayList<>();
@@ -69,7 +73,7 @@ public class TestReconInternalSchemaDefinition extends AbstractReconSqlDBTest {
resultSet.getInt("DATA_TYPE")));
}
- assertEquals(3, actualPairs.size());
+ assertEquals(5, actualPairs.size());
assertEquals(expectedPairs, actualPairs);
}
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/TestReconWithDifferentSqlDBs.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/TestReconWithDifferentSqlDBs.java
index 1805368846..4c59860161 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/TestReconWithDifferentSqlDBs.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/TestReconWithDifferentSqlDBs.java
@@ -72,7 +72,7 @@ public class TestReconWithDifferentSqlDBs {
assertNotNull(reconSqlDB.getDao(dao));
});
ReconTaskStatusDao dao = reconSqlDB.getDao(ReconTaskStatusDao.class);
- dao.insert(new ReconTaskStatus("TestTask", 1L, 2L));
+ dao.insert(new ReconTaskStatus("TestTask", 1L, 2L, 1, 0));
assertEquals(1, dao.findAll().size());
int numRows = reconSqlDB.getDslContext().
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/TestSqlSchemaSetup.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/TestSqlSchemaSetup.java
index 4c6ae91998..983ceb1d8a 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/TestSqlSchemaSetup.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/persistence/TestSqlSchemaSetup.java
@@ -26,7 +26,8 @@ import java.sql.SQLException;
import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
import org.hadoop.ozone.recon.schema.tables.pojos.ReconTaskStatus;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
/**
* Class to test basic SQL schema setup.
@@ -41,8 +42,9 @@ public class TestSqlSchemaSetup extends AbstractReconSqlDBTest {
* Make sure schema was created correctly.
* @throws SQLException
*/
- @Test
- public void testSchemaSetup() throws SQLException {
+ @ParameterizedTest
+ @ValueSource(ints = {0, 1, -1})
+ public void testSchemaSetup(int lastTaskRunStatus) throws SQLException {
assertNotNull(getInjector());
assertNotNull(getConfiguration());
assertNotNull(getDslContext());
@@ -51,7 +53,7 @@ public class TestSqlSchemaSetup extends AbstractReconSqlDBTest {
assertNotNull(getDao(dao));
});
ReconTaskStatusDao dao = getDao(ReconTaskStatusDao.class);
- dao.insert(new ReconTaskStatus("TestTask", 1L, 2L));
+ dao.insert(new ReconTaskStatus("TestTask", 1L, 2L, lastTaskRunStatus, 0));
assertEquals(1, dao.findAll().size());
}
}
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java
index 2700034aae..0d5050a934 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java
@@ -75,8 +75,9 @@ import org.apache.hadoop.ozone.recon.tasks.OMDBUpdatesHandler;
import org.apache.hadoop.ozone.recon.tasks.OMUpdateEventBatch;
import org.apache.hadoop.ozone.recon.tasks.ReconTaskController;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
-import org.hadoop.ozone.recon.schema.tables.pojos.ReconTaskStatus;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -138,8 +139,8 @@ public class TestOzoneManagerServiceProviderImpl {
OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
new OzoneManagerServiceProviderImpl(configuration,
- reconOMMetadataManager, reconTaskController, reconUtilsMock,
- ozoneManagerProtocol, reconContext);
+ reconOMMetadataManager, reconTaskController, reconUtilsMock, ozoneManagerProtocol,
+ reconContext, getMockTaskStatusUpdaterManager());
assertNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
.get("/sampleVol/bucketOne/key_one"));
@@ -179,8 +180,8 @@ public class TestOzoneManagerServiceProviderImpl {
OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
new OzoneManagerServiceProviderImpl(configuration,
- reconOMMetadataManager, reconTaskController, reconUtilsMock,
- ozoneManagerProtocol, reconContext);
+ reconOMMetadataManager, reconTaskController, reconUtilsMock, ozoneManagerProtocol,
+ reconContext, getMockTaskStatusUpdaterManager());
assertFalse(ozoneManagerServiceProvider.updateReconOmDBWithNewSnapshot());
@@ -216,8 +217,8 @@ public class TestOzoneManagerServiceProviderImpl {
OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
new OzoneManagerServiceProviderImpl(configuration,
- reconOMMetadataManager, reconTaskController, reconUtilsMock,
- ozoneManagerProtocol, reconContext);
+ reconOMMetadataManager, reconTaskController, reconUtilsMock, ozoneManagerProtocol,
+ reconContext, getMockTaskStatusUpdaterManager());
assertTrue(reconContext.getErrors().contains(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED));
assertTrue(ozoneManagerServiceProvider.updateReconOmDBWithNewSnapshot());
@@ -259,8 +260,8 @@ public class TestOzoneManagerServiceProviderImpl {
OzoneManagerServiceProviderImpl ozoneManagerServiceProvider1 =
new OzoneManagerServiceProviderImpl(configuration,
- reconOMMetadataManager, reconTaskController, reconUtilsMock,
- ozoneManagerProtocol, reconContext);
+ reconOMMetadataManager, reconTaskController, reconUtilsMock, ozoneManagerProtocol,
+ reconContext, getMockTaskStatusUpdaterManager());
assertTrue(ozoneManagerServiceProvider1.updateReconOmDBWithNewSnapshot());
HttpURLConnection httpURLConnectionMock2 = mock(HttpURLConnection.class);
@@ -269,8 +270,8 @@ public class TestOzoneManagerServiceProviderImpl {
.thenReturn(httpURLConnectionMock2);
OzoneManagerServiceProviderImpl ozoneManagerServiceProvider2 =
new OzoneManagerServiceProviderImpl(configuration,
- reconOMMetadataManager, reconTaskController, reconUtilsMock,
- ozoneManagerProtocol, reconContext);
+ reconOMMetadataManager, reconTaskController, reconUtilsMock, ozoneManagerProtocol,
+ reconContext, getMockTaskStatusUpdaterManager());
assertTrue(ozoneManagerServiceProvider2.updateReconOmDBWithNewSnapshot());
}
@@ -315,8 +316,8 @@ public class TestOzoneManagerServiceProviderImpl {
ReconTaskController reconTaskController = getMockTaskController();
OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
new OzoneManagerServiceProviderImpl(configuration,
- reconOMMetadataManager, reconTaskController, reconUtilsMock,
- ozoneManagerProtocol, reconContext);
+ reconOMMetadataManager, reconTaskController, reconUtilsMock, ozoneManagerProtocol,
+ reconContext, getMockTaskStatusUpdaterManager());
DBCheckpoint checkpoint = ozoneManagerServiceProvider
.getOzoneManagerDBSnapshot();
@@ -365,8 +366,8 @@ public class TestOzoneManagerServiceProviderImpl {
OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
new OzoneManagerServiceProviderImpl(configuration,
getTestReconOmMetadataManager(omMetadataManager, dirReconMetadata),
- getMockTaskController(), new ReconUtils(),
- getMockOzoneManagerClient(dbUpdatesWrapper), reconContext);
+ getMockTaskController(), new ReconUtils(), getMockOzoneManagerClient(dbUpdatesWrapper),
+ reconContext, getMockTaskStatusUpdaterManager());
OMDBUpdatesHandler updatesHandler =
new OMDBUpdatesHandler(omMetadataManager);
@@ -436,7 +437,8 @@ public class TestOzoneManagerServiceProviderImpl {
getTestReconOmMetadataManager(omMetadataManager, dirReconMetadata),
getMockTaskController(), new ReconUtils(),
getMockOzoneManagerClientWith4Updates(dbUpdatesWrapper[0],
- dbUpdatesWrapper[1], dbUpdatesWrapper[2], dbUpdatesWrapper[3]), reconContext);
+ dbUpdatesWrapper[1], dbUpdatesWrapper[2], dbUpdatesWrapper[3]),
+ reconContext, getMockTaskStatusUpdaterManager());
assertTrue(dbUpdatesWrapper[0].isDBUpdateSuccess());
assertTrue(dbUpdatesWrapper[1].isDBUpdateSuccess());
@@ -479,20 +481,15 @@ public class TestOzoneManagerServiceProviderImpl {
// Empty OM DB to start with.
ReconOMMetadataManager omMetadataManager = getTestReconOmMetadataManager(
initializeEmptyOmMetadataManager(dirOmMetadata), dirReconMetadata);
- ReconTaskStatusDao reconTaskStatusDaoMock =
- mock(ReconTaskStatusDao.class);
- doNothing().when(reconTaskStatusDaoMock)
- .update(any(ReconTaskStatus.class));
ReconTaskController reconTaskControllerMock = getMockTaskController();
- when(reconTaskControllerMock.getReconTaskStatusDao())
- .thenReturn(reconTaskStatusDaoMock);
doNothing().when(reconTaskControllerMock)
.reInitializeTasks(omMetadataManager);
+ ReconTaskStatusUpdaterManager reconTaskStatusUpdaterManager = getMockTaskStatusUpdaterManager();
OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
- new MockOzoneServiceProvider(configuration, omMetadataManager,
- reconTaskControllerMock, new ReconUtils(), ozoneManagerProtocol, reconContext);
+ new MockOzoneServiceProvider(configuration, omMetadataManager, reconTaskControllerMock,
+ new ReconUtils(), ozoneManagerProtocol, reconContext, reconTaskStatusUpdaterManager);
OzoneManagerSyncMetrics metrics = ozoneManagerServiceProvider.getMetrics();
assertEquals(0, metrics.getNumSnapshotRequests());
@@ -500,11 +497,9 @@ public class TestOzoneManagerServiceProviderImpl {
// Should trigger full snapshot request.
ozoneManagerServiceProvider.syncDataFromOM();
- ArgumentCaptor<ReconTaskStatus> captor =
- ArgumentCaptor.forClass(ReconTaskStatus.class);
- verify(reconTaskStatusDaoMock, times(1))
- .update(captor.capture());
- assertEquals(OmSnapshotRequest.name(), captor.getValue().getTaskName());
+ ArgumentCaptor<String> taskNameCaptor = ArgumentCaptor.forClass(String.class);
+ verify(reconTaskStatusUpdaterManager).getTaskStatusUpdater(taskNameCaptor.capture());
+ assertEquals(OmSnapshotRequest.name(), taskNameCaptor.getValue());
verify(reconTaskControllerMock, times(1))
.reInitializeTasks(omMetadataManager);
assertEquals(1, metrics.getNumSnapshotRequests());
@@ -518,32 +513,26 @@ public class TestOzoneManagerServiceProviderImpl {
// Non-Empty OM DB to start with.
ReconOMMetadataManager omMetadataManager = getTestReconOmMetadataManager(
initializeNewOmMetadataManager(dirOmMetadata), dirReconMetadata);
- ReconTaskStatusDao reconTaskStatusDaoMock =
- mock(ReconTaskStatusDao.class);
- doNothing().when(reconTaskStatusDaoMock)
- .update(any(ReconTaskStatus.class));
ReconTaskController reconTaskControllerMock = getMockTaskController();
- when(reconTaskControllerMock.getReconTaskStatusDao())
- .thenReturn(reconTaskStatusDaoMock);
doNothing().when(reconTaskControllerMock)
.consumeOMEvents(any(OMUpdateEventBatch.class),
any(OMMetadataManager.class));
+ ReconTaskStatusUpdaterManager reconTaskStatusUpdaterManager = getMockTaskStatusUpdaterManager();
OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
- new OzoneManagerServiceProviderImpl(configuration, omMetadataManager,
- reconTaskControllerMock, new ReconUtils(), ozoneManagerProtocol, reconContext);
+ new OzoneManagerServiceProviderImpl(configuration, omMetadataManager, reconTaskControllerMock,
+ new ReconUtils(), ozoneManagerProtocol, reconContext, reconTaskStatusUpdaterManager);
OzoneManagerSyncMetrics metrics = ozoneManagerServiceProvider.getMetrics();
// Should trigger delta updates.
ozoneManagerServiceProvider.syncDataFromOM();
- ArgumentCaptor<ReconTaskStatus> captor =
- ArgumentCaptor.forClass(ReconTaskStatus.class);
- verify(reconTaskStatusDaoMock, times(1))
- .update(captor.capture());
- assertEquals(OmDeltaRequest.name(), captor.getValue().getTaskName());
+ ArgumentCaptor<String> captor =
+ ArgumentCaptor.forClass(String.class);
+ verify(reconTaskStatusUpdaterManager).getTaskStatusUpdater(captor.capture());
+ assertEquals(OmDeltaRequest.name(), captor.getValue());
verify(reconTaskControllerMock, times(1))
.consumeOMEvents(any(OMUpdateEventBatch.class),
@@ -559,32 +548,26 @@ public class TestOzoneManagerServiceProviderImpl {
// Non-Empty OM DB to start with.
ReconOMMetadataManager omMetadataManager = getTestReconOmMetadataManager(
initializeNewOmMetadataManager(dirOmMetadata), dirReconMetadata);
- ReconTaskStatusDao reconTaskStatusDaoMock =
- mock(ReconTaskStatusDao.class);
- doNothing().when(reconTaskStatusDaoMock)
- .update(any(ReconTaskStatus.class));
ReconTaskController reconTaskControllerMock = getMockTaskController();
- when(reconTaskControllerMock.getReconTaskStatusDao())
- .thenReturn(reconTaskStatusDaoMock);
doNothing().when(reconTaskControllerMock)
.reInitializeTasks(omMetadataManager);
+ ReconTaskStatusUpdaterManager reconTaskStatusUpdaterManager = getMockTaskStatusUpdaterManager();
OzoneManagerProtocol protocol = getMockOzoneManagerClientWithThrow();
OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
- new MockOzoneServiceProvider(configuration, omMetadataManager,
- reconTaskControllerMock, new ReconUtils(), protocol, reconContext);
+ new MockOzoneServiceProvider(configuration, omMetadataManager, reconTaskControllerMock,
+ new ReconUtils(), protocol, reconContext, reconTaskStatusUpdaterManager);
OzoneManagerSyncMetrics metrics = ozoneManagerServiceProvider.getMetrics();
// Should trigger full snapshot request.
ozoneManagerServiceProvider.syncDataFromOM();
- ArgumentCaptor<ReconTaskStatus> captor =
- ArgumentCaptor.forClass(ReconTaskStatus.class);
- verify(reconTaskStatusDaoMock, times(1))
- .update(captor.capture());
- assertEquals(OmSnapshotRequest.name(), captor.getValue().getTaskName());
+ ArgumentCaptor<String> captor =
+ ArgumentCaptor.forClass(String.class);
+ verify(reconTaskStatusUpdaterManager).getTaskStatusUpdater(captor.capture());
+ assertEquals(OmSnapshotRequest.name(), captor.getValue());
verify(reconTaskControllerMock, times(1))
.reInitializeTasks(omMetadataManager);
assertEquals(1, metrics.getNumSnapshotRequests());
@@ -632,6 +615,15 @@ public class TestOzoneManagerServiceProviderImpl {
return ozoneManagerProtocolMock;
}
+ private ReconTaskStatusUpdaterManager getMockTaskStatusUpdaterManager() {
+ ReconTaskStatusUpdaterManager reconTaskStatusUpdaterManager = mock(ReconTaskStatusUpdaterManager.class);
+ when(reconTaskStatusUpdaterManager.getTaskStatusUpdater(anyString())).thenAnswer(inv -> new ReconTaskStatusUpdater(
+ mock(ReconTaskStatusDao.class), (String) inv.getArgument(0)));
+ when(reconTaskStatusUpdaterManager.getTaskStatusUpdater(anyString())).thenAnswer(inv ->
+ new ReconTaskStatusUpdater(mock(ReconTaskStatusDao.class), (String) inv.getArgument(0)));
+ return reconTaskStatusUpdaterManager;
+ }
+
private BucketLayout getBucketLayout() {
return BucketLayout.DEFAULT;
}
@@ -648,13 +640,20 @@ class MockOzoneServiceProvider extends OzoneManagerServiceProviderImpl {
ReconTaskController reconTaskController,
ReconUtils reconUtils,
OzoneManagerProtocol ozoneManagerClient,
- ReconContext reconContext) {
- super(configuration, omMetadataManager, reconTaskController, reconUtils,
- ozoneManagerClient, reconContext);
+ ReconContext reconContext,
+ ReconTaskStatusUpdaterManager taskStatusUpdaterManager) {
+ super(configuration, omMetadataManager, reconTaskController, reconUtils, ozoneManagerClient,
+ reconContext, taskStatusUpdaterManager);
}
@Override
public boolean updateReconOmDBWithNewSnapshot() {
return true;
}
+
+ // Override to trigger full snapshot
+ @Override
+ public long getCurrentOMDBSequenceNumber() {
+ return 0;
+ }
}
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestContainerKeyMapperTask.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestContainerKeyMapperTask.java
index 67f962f186..2790a48599 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestContainerKeyMapperTask.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestContainerKeyMapperTask.java
@@ -315,7 +315,7 @@ public class TestContainerKeyMapperTask {
ArrayList<OMDBUpdateEvent>() {{
add(keyEvent1);
add(keyEvent2);
- }});
+ }}, 0L);
ContainerKeyMapperTask containerKeyMapperTask =
new ContainerKeyMapperTask(reconContainerMetadataManager,
@@ -427,7 +427,7 @@ public class TestContainerKeyMapperTask {
add(keyEvent1);
add(keyEvent2);
}
- });
+ }, 0L);
// Process PUT event for both the keys
containerKeyMapperTask.process(omUpdateEventBatch);
@@ -460,7 +460,7 @@ public class TestContainerKeyMapperTask {
{
add(keyEvent3);
}
- });
+ }, 0L);
// Process DELETE event for key2
containerKeyMapperTask.process(omUpdateEventBatch2);
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestContainerSizeCountTask.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestContainerSizeCountTask.java
index a996f167a1..314176e3c6 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestContainerSizeCountTask.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestContainerSizeCountTask.java
@@ -25,8 +25,10 @@ import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.QU
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.CLOSING;
import static org.hadoop.ozone.recon.schema.tables.ContainerCountBySizeTable.CONTAINER_COUNT_BY_SIZE;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import java.time.Duration;
import java.util.ArrayList;
@@ -37,6 +39,8 @@ import org.apache.hadoop.hdds.scm.container.ContainerInfo;
import org.apache.hadoop.hdds.scm.container.ContainerManager;
import org.apache.hadoop.ozone.recon.persistence.AbstractReconSqlDBTest;
import org.apache.hadoop.ozone.recon.spi.StorageContainerServiceProvider;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.hadoop.ozone.recon.schema.UtilizationSchemaDefinition;
import org.hadoop.ozone.recon.schema.tables.daos.ContainerCountBySizeDao;
import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
@@ -52,8 +56,8 @@ public class TestContainerSizeCountTask extends AbstractReconSqlDBTest {
private ContainerManager containerManager;
private StorageContainerServiceProvider scmClient;
- private ReconTaskStatusDao reconTaskStatusDao;
private ReconTaskConfig reconTaskConfig;
+ private ReconTaskStatusUpdaterManager reconTaskStatusUpdaterManager;
private ContainerCountBySizeDao containerCountBySizeDao;
private UtilizationSchemaDefinition utilizationSchemaDefinition;
private ContainerSizeCountTask task;
@@ -69,18 +73,20 @@ public class TestContainerSizeCountTask extends AbstractReconSqlDBTest {
getSchemaDefinition(UtilizationSchemaDefinition.class);
dslContext = utilizationSchemaDefinition.getDSLContext();
containerCountBySizeDao = getDao(ContainerCountBySizeDao.class);
- reconTaskStatusDao = getDao(ReconTaskStatusDao.class);
reconTaskConfig = new ReconTaskConfig();
reconTaskConfig.setContainerSizeCountTaskInterval(Duration.ofSeconds(1));
+ reconTaskStatusUpdaterManager = mock(ReconTaskStatusUpdaterManager.class);
+ when(reconTaskStatusUpdaterManager.getTaskStatusUpdater(anyString())).thenReturn(new ReconTaskStatusUpdater(
+ getDao(ReconTaskStatusDao.class), "mockedTask-" + System.currentTimeMillis()));
containerManager = mock(ContainerManager.class);
scmClient = mock(StorageContainerServiceProvider.class);
task = new ContainerSizeCountTask(
containerManager,
scmClient,
- reconTaskStatusDao,
reconTaskConfig,
containerCountBySizeDao,
- utilizationSchemaDefinition);
+ utilizationSchemaDefinition,
+ reconTaskStatusUpdaterManager);
// Truncate table before running each test
dslContext.truncate(CONTAINER_COUNT_BY_SIZE);
}
@@ -111,7 +117,7 @@ public class TestContainerSizeCountTask extends AbstractReconSqlDBTest {
containers.add(omContainerInfo1);
containers.add(omContainerInfo2);
- task.process(containers);
+ task.processContainers(containers);
// Verify 3 containers are in correct bins.
assertEquals(3, containerCountBySizeDao.count());
@@ -143,7 +149,7 @@ public class TestContainerSizeCountTask extends AbstractReconSqlDBTest {
given(omContainerInfo2.containerID()).willReturn(new ContainerID(2));
given(omContainerInfo2.getUsedBytes()).willReturn(50000L); // 50KB
- task.process(containers);
+ task.processContainers(containers);
// Total size groups added to the database
assertEquals(5, containerCountBySizeDao.count());
@@ -166,7 +172,7 @@ public class TestContainerSizeCountTask extends AbstractReconSqlDBTest {
// Remove the container having size 1.5GB and upperbound 2147483648L
containers.remove(omContainerInfo1);
- task.process(containers);
+ task.processContainers(containers);
recordToFind.value1(2147483648L);
assertEquals(0, containerCountBySizeDao
.findById(recordToFind.value1())
@@ -222,7 +228,7 @@ public class TestContainerSizeCountTask extends AbstractReconSqlDBTest {
containers.add(negativeSizeDeletedContainer);
containers.add(validSizeContainer);
- task.process(containers);
+ task.processContainers(containers);
// Verify that only the valid containers are counted
assertEquals(3, containerCountBySizeDao.count());
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestFileSizeCountTask.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestFileSizeCountTask.java
index 3572f5813e..e2e6f5c600 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestFileSizeCountTask.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestFileSizeCountTask.java
@@ -195,7 +195,7 @@ public class TestFileSizeCountTask extends AbstractReconSqlDBTest {
.build();
OMUpdateEventBatch omUpdateEventBatch =
- new OMUpdateEventBatch(Arrays.asList(event, event2));
+ new OMUpdateEventBatch(Arrays.asList(event, event2), 0L);
fileSizeCountTask.process(omUpdateEventBatch);
// Verify 2 keys are in correct bins.
@@ -250,7 +250,7 @@ public class TestFileSizeCountTask extends AbstractReconSqlDBTest {
.build();
omUpdateEventBatch = new OMUpdateEventBatch(
- Arrays.asList(updateEvent, putEvent, deleteEvent));
+ Arrays.asList(updateEvent, putEvent, deleteEvent), 0L);
fileSizeCountTask.process(omUpdateEventBatch);
assertEquals(4, fileCountBySizeDao.count());
@@ -388,7 +388,7 @@ public class TestFileSizeCountTask extends AbstractReconSqlDBTest {
}
OMUpdateEventBatch omUpdateEventBatch =
- new OMUpdateEventBatch(omDbEventList);
+ new OMUpdateEventBatch(omDbEventList, 0L);
fileSizeCountTask.process(omUpdateEventBatch);
// Verify 2 keys are in correct bins.
@@ -464,7 +464,7 @@ public class TestFileSizeCountTask extends AbstractReconSqlDBTest {
}
}
- omUpdateEventBatch = new OMUpdateEventBatch(omDbEventList);
+ omUpdateEventBatch = new OMUpdateEventBatch(omDbEventList, 0L);
fileSizeCountTask.process(omUpdateEventBatch);
assertEquals(10000, fileCountBySizeDao.count());
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTask.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTask.java
index 485804240d..22f74bc79f 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTask.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTask.java
@@ -267,7 +267,7 @@ public final class TestNSSummaryTask {
.setAction(OMDBUpdateEvent.OMDBUpdateAction.DELETE)
.build();
- return new OMUpdateEventBatch(Arrays.asList(keyEvent1, keyEvent2));
+ return new OMUpdateEventBatch(Arrays.asList(keyEvent1, keyEvent2), 0L);
}
@Test
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithFSO.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithFSO.java
index ba2e749741..8a7bbb85e2 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithFSO.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithFSO.java
@@ -424,7 +424,7 @@ public final class TestNSSummaryTaskWithFSO {
return new OMUpdateEventBatch(Arrays.asList(
keyEvent1, keyEvent2, keyEvent3, keyEvent4, keyEvent5,
keyEvent6, keyEvent7
- ));
+ ), 0L);
}
@Test
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithLegacy.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithLegacy.java
index 5ffd03cbb8..7fca7bc924 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithLegacy.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithLegacy.java
@@ -432,7 +432,7 @@ public final class TestNSSummaryTaskWithLegacy {
return new OMUpdateEventBatch(Arrays.asList(
keyEvent1, keyEvent2, keyEvent3, keyEvent4, keyEvent5,
keyEvent6, keyEvent7
- ));
+ ), 0L);
}
@Test
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithLegacyOBSLayout.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithLegacyOBSLayout.java
index db48036763..cef2b5ec7a 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithLegacyOBSLayout.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithLegacyOBSLayout.java
@@ -326,7 +326,7 @@ public final class TestNSSummaryTaskWithLegacyOBSLayout {
.build();
return new OMUpdateEventBatch(
- Arrays.asList(keyEvent1, keyEvent2, keyEvent3, keyEvent4));
+ Arrays.asList(keyEvent1, keyEvent2, keyEvent3, keyEvent4), 0L);
}
@Test
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithOBS.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithOBS.java
index 8f9d6b2990..6cf81063ce 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithOBS.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestNSSummaryTaskWithOBS.java
@@ -321,7 +321,7 @@ public final class TestNSSummaryTaskWithOBS implements Serializable {
.build();
return new OMUpdateEventBatch(
- Arrays.asList(keyEvent1, keyEvent2, keyEvent3, keyEvent4));
+ Arrays.asList(keyEvent1, keyEvent2, keyEvent3, keyEvent4), 0L);
}
@Test
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestOmTableInsightTask.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestOmTableInsightTask.java
index 56d8fe2131..8bb4c1f724 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestOmTableInsightTask.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestOmTableInsightTask.java
@@ -320,7 +320,7 @@ public class TestOmTableInsightTask extends AbstractReconSqlDBTest {
getOmKeyInfo("vol1", "bucket1", DIR_ONE, (i + 1), false),
DELETED_DIR_TABLE, PUT, null));
}
- OMUpdateEventBatch putEventBatch = new OMUpdateEventBatch(putEvents);
+ OMUpdateEventBatch putEventBatch = new OMUpdateEventBatch(putEvents, 0L);
omTableInsightTask.process(putEventBatch);
assertEquals(5, getCountForTable(DELETED_DIR_TABLE));
@@ -334,7 +334,7 @@ public class TestOmTableInsightTask extends AbstractReconSqlDBTest {
deleteEvents.add(getOMUpdateEvent(paths.get(2),
getOmKeyInfo("vol1", "bucket1", DIR_ONE, 3L, false), DELETED_DIR_TABLE,
DELETE, null));
- OMUpdateEventBatch deleteEventBatch = new OMUpdateEventBatch(deleteEvents);
+ OMUpdateEventBatch deleteEventBatch = new OMUpdateEventBatch(deleteEvents, 0L);
omTableInsightTask.process(deleteEventBatch);
assertEquals(3, getCountForTable(DELETED_DIR_TABLE));
}
@@ -470,7 +470,7 @@ public class TestOmTableInsightTask extends AbstractReconSqlDBTest {
}
// Processing the initial batch of events
- OMUpdateEventBatch initialBatch = new OMUpdateEventBatch(initialEvents);
+ OMUpdateEventBatch initialBatch = new OMUpdateEventBatch(initialEvents, 0L);
omTableInsightTask.process(initialBatch);
// Verifying the count in each table
@@ -499,7 +499,7 @@ public class TestOmTableInsightTask extends AbstractReconSqlDBTest {
// Processing the additional events
OMUpdateEventBatch additionalBatch =
- new OMUpdateEventBatch(additionalEvents);
+ new OMUpdateEventBatch(additionalEvents, 0L);
omTableInsightTask.process(additionalBatch);
// Verifying the final count in each table
for (String tableName : omTableInsightTask.getTaskTables()) {
@@ -528,7 +528,7 @@ public class TestOmTableInsightTask extends AbstractReconSqlDBTest {
putEvents.add(getOMUpdateEvent("item" + i, omKeyInfo, table, PUT, null));
}
- OMUpdateEventBatch putEventBatch = new OMUpdateEventBatch(putEvents);
+ OMUpdateEventBatch putEventBatch = new OMUpdateEventBatch(putEvents, 0L);
omTableInsightTask.process(putEventBatch);
// After 5 PUTs, size should be 5 * 1000 = 5000
@@ -546,7 +546,7 @@ public class TestOmTableInsightTask extends AbstractReconSqlDBTest {
deleteEvents.add(
getOMUpdateEvent("item0", omKeyInfo, OPEN_FILE_TABLE, DELETE, null));
- OMUpdateEventBatch deleteEventBatch = new OMUpdateEventBatch(deleteEvents);
+ OMUpdateEventBatch deleteEventBatch = new OMUpdateEventBatch(deleteEvents, 0L);
omTableInsightTask.process(deleteEventBatch);
// After deleting "item0", size should be 4 * 1000 = 4000
@@ -569,7 +569,7 @@ public class TestOmTableInsightTask extends AbstractReconSqlDBTest {
getOMUpdateEvent("item1", newKeyInfo, tableName, UPDATE, omKeyInfo));
}
- OMUpdateEventBatch updateEventBatch = new OMUpdateEventBatch(updateEvents);
+ OMUpdateEventBatch updateEventBatch = new OMUpdateEventBatch(updateEvents, 0L);
omTableInsightTask.process(updateEventBatch);
// After updating "item1", size should be 4000 - 1000 + 2000 = 5000
@@ -606,7 +606,7 @@ public class TestOmTableInsightTask extends AbstractReconSqlDBTest {
getOMUpdateEvent("item" + i, repeatedOmKeyInfo, DELETED_TABLE, PUT,
null));
}
- OMUpdateEventBatch putEventBatch = new OMUpdateEventBatch(putEvents);
+ OMUpdateEventBatch putEventBatch = new OMUpdateEventBatch(putEvents, 0L);
omTableInsightTask.process(putEventBatch);
// Each of the 5 RepeatedOmKeyInfo object has 5 OmKeyInfo obj,
// so total deleted keys should be 5 * 5 = 25
@@ -622,7 +622,7 @@ public class TestOmTableInsightTask extends AbstractReconSqlDBTest {
deleteEvents.add(
getOMUpdateEvent("item0", repeatedOmKeyInfo, DELETED_TABLE, DELETE,
null));
- OMUpdateEventBatch deleteEventBatch = new OMUpdateEventBatch(deleteEvents);
+ OMUpdateEventBatch deleteEventBatch = new OMUpdateEventBatch(deleteEvents, 0L);
omTableInsightTask.process(deleteEventBatch);
// After deleting "item0" total deleted keys should be 20
assertEquals(20L, getCountForTable(DELETED_TABLE));
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestReconTaskControllerImpl.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestReconTaskControllerImpl.java
index b5e82a48a8..477a5cace7 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestReconTaskControllerImpl.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestReconTaskControllerImpl.java
@@ -21,11 +21,12 @@ package org.apache.hadoop.ozone.recon.tasks;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertSame;
-import static org.mockito.Mockito.any;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.verify;
import java.util.HashSet;
@@ -34,6 +35,8 @@ import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.recon.persistence.AbstractReconSqlDBTest;
import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
import org.hadoop.ozone.recon.schema.tables.pojos.ReconTaskStatus;
import org.junit.jupiter.api.BeforeEach;
@@ -55,8 +58,14 @@ public class TestReconTaskControllerImpl extends AbstractReconSqlDBTest {
public void setUp() {
OzoneConfiguration ozoneConfiguration = new OzoneConfiguration();
reconTaskStatusDao = getDao(ReconTaskStatusDao.class);
- reconTaskController = new ReconTaskControllerImpl(ozoneConfiguration,
- reconTaskStatusDao, new HashSet<>());
+ ReconTaskStatusUpdaterManager reconTaskStatusUpdaterManagerMock = mock(ReconTaskStatusUpdaterManager.class);
+ when(reconTaskStatusUpdaterManagerMock.getTaskStatusUpdater(anyString()))
+ .thenAnswer(i -> {
+ String taskName = i.getArgument(0);
+ return new ReconTaskStatusUpdater(reconTaskStatusDao, taskName);
+ });
+ reconTaskController = new ReconTaskControllerImpl(ozoneConfiguration, new HashSet<>(),
+ reconTaskStatusUpdaterManagerMock);
reconTaskController.start();
}
@@ -90,7 +99,6 @@ public class TestReconTaskControllerImpl extends AbstractReconSqlDBTest {
.process(any());
long endTime = System.currentTimeMillis();
- reconTaskStatusDao = getDao(ReconTaskStatusDao.class);
ReconTaskStatus reconTaskStatus = reconTaskStatusDao.findById("MockTask");
long taskTimeStamp = reconTaskStatus.getLastUpdatedTimestamp();
long seqNumber = reconTaskStatus.getLastUpdatedSeqNumber();
@@ -99,6 +107,38 @@ public class TestReconTaskControllerImpl extends AbstractReconSqlDBTest {
assertEquals(seqNumber, omUpdateEventBatchMock.getLastSequenceNumber());
}
+ @Test
+ public void testTaskRecordsFailureOnException() throws Exception {
+ ReconOmTask reconOmTaskMock = getMockTask("MockTask");
+ OMUpdateEventBatch omUpdateEventBatchMock = mock(OMUpdateEventBatch.class);
+
+ // Throw exception when trying to run task
+ when(reconOmTaskMock.process(any(OMUpdateEventBatch.class)))
+ .thenThrow(new RuntimeException("Mock Failure"));
+ reconTaskController.registerTask(reconOmTaskMock);
+ when(omUpdateEventBatchMock.getLastSequenceNumber()).thenReturn(100L);
+ when(omUpdateEventBatchMock.isEmpty()).thenReturn(false);
+
+ long startTime = System.currentTimeMillis();
+ reconTaskController.consumeOMEvents(
+ omUpdateEventBatchMock,
+ mock(OMMetadataManager.class));
+
+ verify(reconOmTaskMock, times(1))
+ .process(any());
+ long endTime = System.currentTimeMillis();
+
+ ReconTaskStatus reconTaskStatus = reconTaskStatusDao.findById("MockTask");
+ long taskTimeStamp = reconTaskStatus.getLastUpdatedTimestamp();
+ long seqNumber = reconTaskStatus.getLastUpdatedSeqNumber();
+ int taskStatus = reconTaskStatus.getLastTaskRunStatus();
+
+ assertThat(taskTimeStamp).isGreaterThanOrEqualTo(startTime).isLessThanOrEqualTo(endTime);
+ // Task failed so seqNumber should not be updated, and last task status should be -1
+ assertEquals(seqNumber, 0);
+ assertEquals(taskStatus, -1);
+ }
+
@Test
public void testFailedTaskRetryLogic() throws Exception {
String taskName = "Dummy_" + System.currentTimeMillis();
@@ -149,6 +189,7 @@ public class TestReconTaskControllerImpl extends AbstractReconSqlDBTest {
}
//Should be ignored now.
+ Long startTime = System.currentTimeMillis();
reconTaskController.consumeOMEvents(omUpdateEventBatchMock,
omMetadataManagerMock);
assertThat(reconTaskController.getRegisteredTasks()).isEmpty();
@@ -157,7 +198,7 @@ public class TestReconTaskControllerImpl extends AbstractReconSqlDBTest {
ReconTaskStatus dbRecord = reconTaskStatusDao.findById(taskName);
assertEquals(taskName, dbRecord.getTaskName());
- assertEquals(Long.valueOf(0L), dbRecord.getLastUpdatedTimestamp());
+ assertThat(dbRecord.getLastUpdatedTimestamp()).isGreaterThanOrEqualTo(startTime);
assertEquals(Long.valueOf(0L), dbRecord.getLastUpdatedSeqNumber());
}
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestReconTaskStatusUpdater.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestReconTaskStatusUpdater.java
new file mode 100644
index 0000000000..379732d2c8
--- /dev/null
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestReconTaskStatusUpdater.java
@@ -0,0 +1,101 @@
+/*
+ * 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.recon.tasks;
+
+import org.apache.hadoop.ozone.recon.persistence.AbstractReconSqlDBTest;
+import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
+import org.hadoop.ozone.recon.schema.tables.daos.ReconTaskStatusDao;
+import org.hadoop.ozone.recon.schema.tables.pojos.ReconTaskStatus;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * This class contains tests to validate {@link ReconTaskStatusUpdater} class.
+ */
+public class TestReconTaskStatusUpdater extends AbstractReconSqlDBTest {
+ private ReconTaskStatusDao reconTaskStatusDaoMock;
+ private ReconTaskStatusUpdater updater;
+
+ @BeforeEach
+ void setup() {
+ this.reconTaskStatusDaoMock = mock(ReconTaskStatusDao.class);
+ this.updater = mock(ReconTaskStatusUpdater.class);
+
+ doAnswer(inv -> {
+ if (!reconTaskStatusDaoMock.existsById(anyString())) {
+ // First time getting the task, so insert value
+ reconTaskStatusDaoMock.insert(new ReconTaskStatus("task1", 0L, 0L, 0, 0));
+ } else {
+ // We already have row for the task in the table, update the row
+ reconTaskStatusDaoMock.update(new ReconTaskStatus("task1", 0L, 0L, 0, 0));
+ }
+ return null;
+ }).when(updater).updateDetails();
+ doNothing().when(updater).setLastUpdatedSeqNumber(anyLong());
+ doNothing().when(updater).setLastUpdatedTimestamp(anyLong());
+ doNothing().when(updater).setLastTaskRunStatus(anyInt());
+ doNothing().when(updater).setIsCurrentTaskRunning(anyInt());
+ }
+
+ @Test
+ void testUpdateDetailsFirstTime() {
+ when(reconTaskStatusDaoMock.existsById(anyString())).thenReturn(false);
+
+ updater.updateDetails();
+
+ verify(reconTaskStatusDaoMock, times(1)).insert(any(ReconTaskStatus.class));
+ verify(reconTaskStatusDaoMock, never()).update(any(ReconTaskStatus.class));
+ }
+
+ @Test
+ void testUpdateDetailsUpdateExisting() {
+ when(reconTaskStatusDaoMock.existsById(anyString())).thenReturn(true);
+
+ updater.setLastTaskRunStatus(1); // Task success
+ updater.updateDetails();
+
+ verify(reconTaskStatusDaoMock, times(1)).update(any(ReconTaskStatus.class));
+ }
+
+ @Test
+ void testSetters() {
+ updater.setLastUpdatedSeqNumber(100L);
+ updater.setLastUpdatedTimestamp(200L);
+ updater.setLastTaskRunStatus(1);
+ updater.setIsCurrentTaskRunning(0);
+
+ // Verify the fields are updated without throwing exceptions
+ verify(updater, times(1)).setLastUpdatedSeqNumber(anyLong());
+ verify(updater, times(1)).setLastUpdatedTimestamp(anyLong());
+ verify(updater, times(1)).setLastTaskRunStatus(anyInt());
+ verify(updater, times(1)).setIsCurrentTaskRunning(anyInt());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org