This is an automated email from the ASF dual-hosted git repository.
mhubail pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
The following commit(s) were added to refs/heads/master by this push:
new 1f59a5e [NO ISSUE][OTH] Make Active Recovery Task Extensible
1f59a5e is described below
commit 1f59a5eec0a3874f9bb701405a592009e64e4e86
Author: Murtadha Hubail <[email protected]>
AuthorDate: Tue Sep 3 01:40:49 2019 +0300
[NO ISSUE][OTH] Make Active Recovery Task Extensible
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
- Allow active entity listeners to create their own
recovery tasks.
- Make lock acquisition in recovery task extensible.
Change-Id: I801eec74f7c1723e8243fe0f36db6148638bde35
Reviewed-on: https://asterix-gerrit.ics.uci.edu/3538
Sonar-Qube: Jenkins <[email protected]>
Contrib: Jenkins <[email protected]>
Tested-by: Jenkins <[email protected]>
Integration-Tests: Jenkins <[email protected]>
Reviewed-by: Murtadha Hubail <[email protected]>
Reviewed-by: Till Westmann <[email protected]>
---
.../app/active/ActiveEntityEventsListener.java | 6 ++-
.../apache/asterix/app/active/RecoveryTask.java | 46 +++++++++++++++-------
2 files changed, 36 insertions(+), 16 deletions(-)
diff --git
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
index 3b847b3..882afc5 100644
---
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
+++
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
@@ -366,7 +366,7 @@ public abstract class ActiveEntityEventsListener implements
IActiveEntityControl
ExecutorService executor =
appCtx.getServiceContext().getControllerService().getExecutor();
setState(ActivityState.TEMPORARILY_FAILED);
LOGGER.log(level, "Recovery task has been submitted");
- rt = new RecoveryTask(appCtx, this, retryPolicyFactory);
+ rt = createRecoveryTask();
executor.submit(rt.recover());
}
}
@@ -685,6 +685,10 @@ public abstract class ActiveEntityEventsListener
implements IActiveEntityControl
return suspended;
}
+ protected RecoveryTask createRecoveryTask() {
+ return new RecoveryTask(appCtx, this, retryPolicyFactory);
+ }
+
@Override
public String toString() {
return "{\"class\":\"" + getClass().getSimpleName() + "\"," +
"\"entityId\":\"" + entityId + "\","
diff --git
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java
index 0172b28..a1989fc 100644
---
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java
+++
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java
@@ -119,13 +119,7 @@ public class RecoveryTask {
}
IMetadataLockManager lockManager =
metadataProvider.getApplicationContext().getMetadataLockManager();
try {
-
lockManager.acquireActiveEntityWriteLock(metadataProvider.getLocks(),
- listener.getEntityId().getDataverse() + '.' +
listener.getEntityId().getEntityName());
- for (Dataset dataset : listener.getDatasets()) {
-
lockManager.acquireDataverseReadLock(metadataProvider.getLocks(),
dataset.getDataverseName());
-
lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(),
- DatasetUtil.getFullyQualifiedName(dataset));
- }
+ acquireRecoveryLocks(lockManager);
synchronized (listener) {
try {
if (!cancelRecovery && listener.getState() ==
ActivityState.TEMPORARILY_FAILED) {
@@ -143,7 +137,7 @@ public class RecoveryTask {
listener.setState(ActivityState.TEMPORARILY_FAILED);
failure = e;
} finally {
- metadataProvider.getLocks().reset();
+ releaseRecoveryLocks(metadataProvider);
}
}
// Recovery task is essntially over now either through failure or
through cancellation(stop)
@@ -163,12 +157,7 @@ public class RecoveryTask {
}
IMetadataLockManager lockManager =
metadataProvider.getApplicationContext().getMetadataLockManager();
try {
-
lockManager.acquireActiveEntityWriteLock(metadataProvider.getLocks(),
- listener.getEntityId().getDataverse() + '.' +
listener.getEntityId().getEntityName());
- for (Dataset dataset : listener.getDatasets()) {
- MetadataLockUtil.modifyDatasetBegin(lockManager,
metadataProvider.getLocks(), dataset.getDatasetName(),
- DatasetUtil.getFullyQualifiedName(dataset));
- }
+ acquirePostRecoveryLocks(lockManager);
synchronized (listener) {
if (!cancelRecovery && listener.getState() ==
ActivityState.TEMPORARILY_FAILED) {
LOGGER.warn("Recovery for {} permanently failed",
listener.getEntityId());
@@ -178,8 +167,35 @@ public class RecoveryTask {
listener.notifyAll();
}
} finally {
- metadataProvider.getLocks().reset();
+ releasePostRecoveryLocks();
}
return null;
}
+
+ protected void acquireRecoveryLocks(IMetadataLockManager lockManager)
throws AlgebricksException {
+ lockManager.acquireActiveEntityWriteLock(metadataProvider.getLocks(),
+ listener.getEntityId().getDataverse() + '.' +
listener.getEntityId().getEntityName());
+ for (Dataset dataset : listener.getDatasets()) {
+ lockManager.acquireDataverseReadLock(metadataProvider.getLocks(),
dataset.getDataverseName());
+
lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(),
+ DatasetUtil.getFullyQualifiedName(dataset));
+ }
+ }
+
+ protected void releaseRecoveryLocks(MetadataProvider metadataProvider) {
+ metadataProvider.getLocks().reset();
+ }
+
+ protected void acquirePostRecoveryLocks(IMetadataLockManager lockManager)
throws AlgebricksException {
+ lockManager.acquireActiveEntityWriteLock(metadataProvider.getLocks(),
+ listener.getEntityId().getDataverse() + '.' +
listener.getEntityId().getEntityName());
+ for (Dataset dataset : listener.getDatasets()) {
+ MetadataLockUtil.modifyDatasetBegin(lockManager,
metadataProvider.getLocks(), dataset.getDatasetName(),
+ DatasetUtil.getFullyQualifiedName(dataset));
+ }
+ }
+
+ protected void releasePostRecoveryLocks() {
+ metadataProvider.getLocks().reset();
+ }
}
