add check for remove task execution tracking when a node is offline
@@ -50,6 +50,7 @@ import com.orientechnologies.orient.server.distributed.impl.lock.OLockGuard;
import com.orientechnologies.orient.server.distributed.impl.lock.OLockManager;
import com.orientechnologies.orient.server.distributed.impl.lock.OLockManagerImpl;
import com.orientechnologies.orient.server.distributed.impl.lock.OTxPromiseManager;
+import com.orientechnologies.orient.server.distributed.impl.lock.OnLocksAcquired;
import com.orientechnologies.orient.server.distributed.impl.task.OLockKeySource;
import com.orientechnologies.orient.server.distributed.impl.task.OUnreachableServerLocalTask;
import com.orientechnologies.orient.server.distributed.impl.task.transaction.OTransactionUniqueKey;
@@ -350,26 +351,31 @@ public class ODistributedDatabaseImpl implements ODistributedDatabase {
SortedSet<ORID> rids = ((OLockKeySource) task).getRids();
SortedSet<OTransactionUniqueKey> uniqueKeys = ((OLockKeySource) task).getUniqueKeys();
OTransactionId txId = ((OLockKeySource) task).getTransactionId();
- this.lockManager.lock(
- rids,
- uniqueKeys,
- txId,
+
+ OnLocksAcquired acquired =
(guards) -> {
- try {
- this.requestExecutor.submit(
+ Runnable executeTask =
() -> {
try {
execute(request);
} finally {
this.lockManager.unlock(guards);
}
- });
+ };
+ try {
+ this.requestExecutor.submit(executeTask);
} catch (RejectedExecutionException e) {
task.finished(this);
this.lockManager.unlock(guards);
throw e;
}
- });
+ };
+ try {
+ this.lockManager.lock(rids, uniqueKeys, txId, acquired);
+ } catch (OOfflineNodeException e) {
+ task.finished(this);
+ throw e;
+ }
} else {
try {
this.requestExecutor.submit(
