Repository: hbase Updated Branches: refs/heads/branch-2 20a9b8ad0 -> 92b9b0f26
HBASE-21323 Should not skip force updating for a sub procedure even if it has been finished Project: http://git-wip-us.apache.org/repos/asf/hbase/repo Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/92b9b0f2 Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/92b9b0f2 Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/92b9b0f2 Branch: refs/heads/branch-2 Commit: 92b9b0f26d1f3d4b501a2dd1fea395d90720a1d2 Parents: 20a9b8a Author: zhangduo <[email protected]> Authored: Wed Oct 17 20:51:19 2018 +0800 Committer: Duo Zhang <[email protected]> Committed: Thu Oct 18 14:44:10 2018 +0800 ---------------------------------------------------------------------- .../hbase/procedure2/ProcedureExecutor.java | 9 +++++- .../store/wal/TestForceUpdateProcedure.java | 33 ++++++++++++++++++-- 2 files changed, 39 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hbase/blob/92b9b0f2/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java ---------------------------------------------------------------------- diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java index d57fcb5..25c72ba 100644 --- a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java +++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java @@ -376,6 +376,11 @@ public class ProcedureExecutor<TEnvironment> { this(conf, environment, store, new SimpleProcedureScheduler()); } + private boolean isRootFinished(Procedure<?> proc) { + Procedure<?> rootProc = procedures.get(proc.getRootProcId()); + return rootProc == null || rootProc.isFinished(); + } + private void forceUpdateProcedure(long procId) throws IOException { IdLock.Entry lockEntry = procExecutionLock.getLockEntry(procId); try { @@ -384,7 +389,9 @@ public class ProcedureExecutor<TEnvironment> { LOG.debug("No pending procedure with id = {}, skip force updating.", procId); return; } - if (proc.isFinished()) { + // For a sub procedure which root parent has not been finished, we still need to retain the + // wal even if the procedure itself is finished. + if (proc.isFinished() && (!proc.hasParent() || isRootFinished(proc))) { LOG.debug("Procedure {} has already been finished, skip force updating.", proc); return; } http://git-wip-us.apache.org/repos/asf/hbase/blob/92b9b0f2/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestForceUpdateProcedure.java ---------------------------------------------------------------------- diff --git a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestForceUpdateProcedure.java b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestForceUpdateProcedure.java index 0d37b77..df6ee51 100644 --- a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestForceUpdateProcedure.java +++ b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestForceUpdateProcedure.java @@ -123,7 +123,34 @@ public class TestForceUpdateProcedure { @Override protected Procedure<Void>[] execute(Void env) throws ProcedureYieldException, ProcedureSuspendedException, InterruptedException { - return new Procedure[] { new WaitingProcedure() }; + return new Procedure[] { new DummyProcedure(), new WaitingProcedure() }; + } + + @Override + protected void rollback(Void env) throws IOException, InterruptedException { + throw new UnsupportedOperationException(); + } + + @Override + protected boolean abort(Void env) { + return false; + } + + @Override + protected void serializeStateData(ProcedureStateSerializer serializer) throws IOException { + } + + @Override + protected void deserializeStateData(ProcedureStateSerializer serializer) throws IOException { + } + } + + public static final class DummyProcedure extends Procedure<Void> { + + @Override + protected Procedure<Void>[] execute(Void env) + throws ProcedureYieldException, ProcedureSuspendedException, InterruptedException { + return null; } @Override @@ -208,10 +235,12 @@ public class TestForceUpdateProcedure { createStoreAndExecutor(); Map<Class<?>, Procedure<Void>> procMap = new HashMap<>(); EXEC.getActiveProceduresNoCopy().forEach(p -> procMap.put(p.getClass(), p)); - assertEquals(2, procMap.size()); + assertEquals(3, procMap.size()); ParentProcedure parentProc = (ParentProcedure) procMap.get(ParentProcedure.class); assertEquals(ProcedureState.WAITING, parentProc.getState()); WaitingProcedure waitingProc = (WaitingProcedure) procMap.get(WaitingProcedure.class); assertEquals(ProcedureState.WAITING_TIMEOUT, waitingProc.getState()); + DummyProcedure dummyProc = (DummyProcedure) procMap.get(DummyProcedure.class); + assertEquals(ProcedureState.SUCCESS, dummyProc.getState()); } }
