You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by zh...@apache.org on 2024/07/17 01:45:00 UTC
(hbase) branch branch-2.6 updated: HBASE-28683 Only allow one TableProcedureInterface for a single table to run at the same time for some special procedure types (#6046) (#6084)
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/branch-2.6 by this push:
new e1bd63246fd HBASE-28683 Only allow one TableProcedureInterface for a single table to run at the same time for some special procedure types (#6046) (#6084)
e1bd63246fd is described below
commit e1bd63246fdad805c038fb366ab66ee639535e5f
Author: Duo Zhang <zh...@apache.org>
AuthorDate: Wed Jul 17 09:19:11 2024 +0800
HBASE-28683 Only allow one TableProcedureInterface for a single table to run at the same time for some special procedure types (#6046) (#6084)
Signed-off-by: Viraj Jasani <vj...@apache.org>
(cherry picked from commit 50a495f735b88f933311dd9fff7d6ce9364298a6)
(cherry picked from commit 855410fad73ba5dbc7872d5bf7684f29bec5bfc5)
---
.../apache/hadoop/hbase/procedure2/Procedure.java | 2 +-
.../hadoop/hbase/procedure2/ProcedureExecutor.java | 22 ++++
.../master/procedure/MasterProcedureScheduler.java | 72 +++++++++++--
.../procedure/TableProcedureWaitingQueue.java | 117 +++++++++++++++++++++
.../hadoop/hbase/master/procedure/TableQueue.java | 2 +-
...terProcedureSchedulerPerformanceEvaluation.java | 1 +
.../procedure/TestMasterProcedureScheduler.java | 42 +++++---
.../TestMasterProcedureSchedulerConcurrency.java | 5 +-
8 files changed, 241 insertions(+), 22 deletions(-)
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/Procedure.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/Procedure.java
index 3cf3b2f5ee5..0356f806bf4 100644
--- a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/Procedure.java
+++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/Procedure.java
@@ -906,7 +906,7 @@ public abstract class Procedure<TEnvironment> implements Comparable<Procedure<TE
this.wasExecuted = true;
}
- protected synchronized boolean wasExecuted() {
+ public synchronized boolean wasExecuted() {
return wasExecuted;
}
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 e0c50c7dc8a..fb3b9d90bf7 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
@@ -503,6 +503,28 @@ public class ProcedureExecutor<TEnvironment> {
private void pushProceduresAfterLoad(List<Procedure<TEnvironment>> runnableList,
List<Procedure<TEnvironment>> failedList) {
failedList.forEach(scheduler::addBack);
+ // Put the procedures which have been executed first
+ // For table procedures, to prevent concurrent modifications, we only allow one procedure to run
+ // for a single table at the same time, this is done via inserting a waiting queue before
+ // actually add the procedure to run queue. So when loading here, we should add the procedures
+ // which have been executed first, otherwise another procedure which was in the waiting queue
+ // before restarting may be added to run queue first and still cause concurrent modifications.
+ // See HBASE-28263 for the reason why we need this
+ runnableList.sort((p1, p2) -> {
+ if (p1.wasExecuted()) {
+ if (p2.wasExecuted()) {
+ return Long.compare(p1.getProcId(), p2.getProcId());
+ } else {
+ return -1;
+ }
+ } else {
+ if (p2.wasExecuted()) {
+ return 1;
+ } else {
+ return Long.compare(p1.getProcId(), p2.getProcId());
+ }
+ }
+ });
runnableList.forEach(p -> {
p.afterReplay(getEnvironment());
if (!p.hasParent()) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureScheduler.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureScheduler.java
index 9a782d3d4a7..6f1338db944 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureScheduler.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureScheduler.java
@@ -19,7 +19,10 @@ package org.apache.hadoop.hbase.master.procedure;
import java.io.IOException;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Optional;
import java.util.function.Function;
import java.util.function.Supplier;
import org.apache.commons.lang3.builder.ToStringBuilder;
@@ -109,9 +112,17 @@ public class MasterProcedureScheduler extends AbstractProcedureScheduler {
private PeerQueue peerMap = null;
private MetaQueue metaMap = null;
+ private final Function<Long, Procedure<?>> procedureRetriever;
private final SchemaLocking locking;
+ // To prevent multiple Create/Modify/Disable/Enable table procedure run at the same time, we will
+ // keep table procedure in this queue first before actually enqueuing it to tableQueue
+ // Seee HBASE-28683 for more details
+ private final Map<TableName, TableProcedureWaitingQueue> tableProcsWaitingEnqueue =
+ new HashMap<>();
+
public MasterProcedureScheduler(Function<Long, Procedure<?>> procedureRetriever) {
+ this.procedureRetriever = procedureRetriever;
locking = new SchemaLocking(procedureRetriever);
}
@@ -120,11 +131,26 @@ public class MasterProcedureScheduler extends AbstractProcedureScheduler {
push(proc, false, true);
}
+ private boolean shouldWaitBeforeEnqueuing(TableProcedureInterface proc) {
+ return TableQueue.requireTableExclusiveLock(proc);
+ }
+
@Override
protected void enqueue(final Procedure proc, final boolean addFront) {
if (isMetaProcedure(proc)) {
doAdd(metaRunQueue, getMetaQueue(), proc, addFront);
} else if (isTableProcedure(proc)) {
+ TableProcedureInterface tableProc = (TableProcedureInterface) proc;
+ if (shouldWaitBeforeEnqueuing(tableProc)) {
+ TableProcedureWaitingQueue waitingQueue = tableProcsWaitingEnqueue.computeIfAbsent(
+ tableProc.getTableName(), k -> new TableProcedureWaitingQueue(procedureRetriever));
+ if (!waitingQueue.procedureSubmitted(proc)) {
+ // there is a table procedure for this table already enqueued, waiting
+ LOG.debug("There is already a procedure running for table {}, added {} to waiting queue",
+ tableProc.getTableName(), proc);
+ return;
+ }
+ }
doAdd(tableRunQueue, getTableQueue(getTableName(proc)), proc, addFront);
} else if (isServerProcedure(proc)) {
ServerProcedureInterface spi = (ServerProcedureInterface) proc;
@@ -266,6 +292,7 @@ public class MasterProcedureScheduler extends AbstractProcedureScheduler {
// Remove Tables
clear(tableMap, tableRunQueue, TABLE_QUEUE_KEY_COMPARATOR);
tableMap = null;
+ tableProcsWaitingEnqueue.clear();
// Remove Peers
clear(peerMap, peerRunQueue, PEER_QUEUE_KEY_COMPARATOR);
@@ -303,17 +330,46 @@ public class MasterProcedureScheduler extends AbstractProcedureScheduler {
count += queueSize(tableMap);
count += queueSize(peerMap);
count += queueSize(metaMap);
+ for (TableProcedureWaitingQueue waitingQ : tableProcsWaitingEnqueue.values()) {
+ count += waitingQ.waitingSize();
+ }
return count;
}
@Override
public void completionCleanup(final Procedure proc) {
- if (proc instanceof TableProcedureInterface) {
- TableProcedureInterface iProcTable = (TableProcedureInterface) proc;
+ if (isTableProcedure(proc)) {
+ TableProcedureInterface tableProc = (TableProcedureInterface) proc;
+ if (shouldWaitBeforeEnqueuing(tableProc)) {
+ schedLock();
+ try {
+ TableProcedureWaitingQueue waitingQueue =
+ tableProcsWaitingEnqueue.get(tableProc.getTableName());
+ if (waitingQueue != null) {
+ Optional<Procedure<?>> nextProc = waitingQueue.procedureCompleted(proc);
+ if (nextProc.isPresent()) {
+ // enqueue it
+ Procedure<?> next = nextProc.get();
+ LOG.debug("{} completed, enqueue a new procedure {}", proc, next);
+ doAdd(tableRunQueue, getTableQueue(tableProc.getTableName()), next, false);
+ } else {
+ if (waitingQueue.isEmpty()) {
+ // there is no waiting procedures in it, remove
+ tableProcsWaitingEnqueue.remove(tableProc.getTableName());
+ }
+ }
+ } else {
+ // this should not happen normally, warn it
+ LOG.warn("no waiting queue while completing {}, which should not happen", proc);
+ }
+ } finally {
+ schedUnlock();
+ }
+ }
boolean tableDeleted;
if (proc.hasException()) {
Exception procEx = proc.getException().unwrapRemoteException();
- if (iProcTable.getTableOperationType() == TableOperationType.CREATE) {
+ if (tableProc.getTableOperationType() == TableOperationType.CREATE) {
// create failed because the table already exist
tableDeleted = !(procEx instanceof TableExistsException);
} else {
@@ -322,11 +378,10 @@ public class MasterProcedureScheduler extends AbstractProcedureScheduler {
}
} else {
// the table was deleted
- tableDeleted = (iProcTable.getTableOperationType() == TableOperationType.DELETE);
+ tableDeleted = (tableProc.getTableOperationType() == TableOperationType.DELETE);
}
if (tableDeleted) {
- markTableAsDeleted(iProcTable.getTableName(), proc);
- return;
+ markTableAsDeleted(tableProc.getTableName(), proc);
}
} else if (proc instanceof PeerProcedureInterface) {
tryCleanupPeerQueue(getPeerId(proc), proc);
@@ -657,7 +712,9 @@ public class MasterProcedureScheduler extends AbstractProcedureScheduler {
try {
final TableQueue queue = getTableQueue(table);
final LockAndQueue tableLock = locking.getTableLock(table);
- if (queue == null) return true;
+ if (queue == null) {
+ return true;
+ }
if (queue.isEmpty() && tableLock.tryExclusiveLock(procedure)) {
// remove the table from the run-queue and the map
@@ -1042,6 +1099,7 @@ public class MasterProcedureScheduler extends AbstractProcedureScheduler {
serverBucketToString(builder, "serverBuckets[" + i + "]", serverBuckets[i]);
}
builder.append("tableMap", tableMap);
+ builder.append("tableWaitingMap", tableProcsWaitingEnqueue);
builder.append("peerMap", peerMap);
builder.append("metaMap", metaMap);
} finally {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/TableProcedureWaitingQueue.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/TableProcedureWaitingQueue.java
new file mode 100644
index 00000000000..fbd5037c880
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/TableProcedureWaitingQueue.java
@@ -0,0 +1,117 @@
+/*
+ * 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.
+ */
+package org.apache.hadoop.hbase.master.procedure;
+
+import java.util.ArrayDeque;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.function.Function;
+import org.apache.hadoop.hbase.procedure2.Procedure;
+import org.apache.yetus.audience.InterfaceAudience;
+
+import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
+
+/**
+ * To prevent multiple Create/Modify/Disable/Enable table procedures run at the same time, we will
+ * keep table procedure in this queue first before actually enqueuing it to
+ * MasterProcedureScheduler's tableQueue. See HBASE-28683 for more details
+ */
+@InterfaceAudience.Private
+class TableProcedureWaitingQueue {
+
+ private final Function<Long, Procedure<?>> procedureRetriever;
+
+ // whether there is already a table procedure enqueued in ProcedureScheduler.
+ private Procedure<?> enqueuedProc;
+
+ private final Queue<Procedure<?>> queue = new ArrayDeque<>();
+
+ TableProcedureWaitingQueue(Function<Long, Procedure<?>> procedureRetriever) {
+ this.procedureRetriever = procedureRetriever;
+ }
+
+ private boolean isSubProcedure(Procedure<?> proc) {
+ while (proc.hasParent()) {
+ if (proc.getParentProcId() == enqueuedProc.getProcId()) {
+ return true;
+ }
+ proc = Preconditions.checkNotNull(procedureRetriever.apply(proc.getParentProcId()),
+ "can not find parent procedure pid=%s", proc.getParentProcId());
+ }
+ return false;
+ }
+
+ /**
+ * Return whether we can enqueue this procedure to ProcedureScheduler.
+ * <p>
+ * If returns {@code true}, you should enqueue this procedure, otherwise you just need to do
+ * nothing, as we will queue it in the waitingQueue, and you will finally get it again by calling
+ * {@link #procedureCompleted(Procedure)} method in the future.
+ */
+ boolean procedureSubmitted(Procedure<?> proc) {
+ if (enqueuedProc == null) {
+ // no procedure enqueued yet, record it and return
+ enqueuedProc = proc;
+ return true;
+ }
+ if (proc == enqueuedProc) {
+ // the same procedure is enqueued again, this usually because the procedure comes back from
+ // WAITING state, such as all child procedures are finished
+ return true;
+ }
+ // check whether this is a sub procedure of the enqueued procedure
+ if (isSubProcedure(proc)) {
+ return true;
+ }
+ queue.add(proc);
+ return false;
+ }
+
+ /**
+ * Return the next procedure which can be enqueued to ProcedureScheduler.
+ */
+ Optional<Procedure<?>> procedureCompleted(Procedure<?> proc) {
+ Preconditions.checkState(enqueuedProc != null, "enqueued procedure should not be null");
+ if (enqueuedProc == proc) {
+ if (!queue.isEmpty()) {
+ enqueuedProc = queue.poll();
+ return Optional.of(enqueuedProc);
+ } else {
+ enqueuedProc = null;
+ return Optional.empty();
+ }
+ } else {
+ Preconditions.checkState(isSubProcedure(proc),
+ "procedure %s is not a sub procedure of enqueued procedure %s", proc, enqueuedProc);
+ return Optional.empty();
+ }
+ }
+
+ boolean isEmpty() {
+ return enqueuedProc == null;
+ }
+
+ int waitingSize() {
+ return queue.size();
+ }
+
+ @Override
+ public String toString() {
+ return "TableProcedureWaitingQueue [enqueuedProc=" + enqueuedProc + ", queue=" + queue + "]";
+ }
+}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/TableQueue.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/TableQueue.java
index 1784c44aca7..ec85cddafa0 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/TableQueue.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/TableQueue.java
@@ -47,7 +47,7 @@ class TableQueue extends Queue<TableName> {
/**
* @param proc must not be null
*/
- private static boolean requireTableExclusiveLock(TableProcedureInterface proc) {
+ static boolean requireTableExclusiveLock(TableProcedureInterface proc) {
switch (proc.getTableOperationType()) {
case CREATE:
case DELETE:
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureSchedulerPerformanceEvaluation.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureSchedulerPerformanceEvaluation.java
index 5c40a71804b..542faebd38f 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureSchedulerPerformanceEvaluation.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureSchedulerPerformanceEvaluation.java
@@ -123,6 +123,7 @@ public class MasterProcedureSchedulerPerformanceEvaluation extends AbstractHBase
@Override
public void releaseLock(Void env) {
procedureScheduler.wakeTableExclusiveLock(this, getTableName());
+ procedureScheduler.completionCleanup(this);
}
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureScheduler.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureScheduler.java
index 127ff9a5aa6..de8c00cd28d 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureScheduler.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureScheduler.java
@@ -27,7 +27,9 @@ import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
@@ -65,12 +67,15 @@ public class TestMasterProcedureScheduler {
private MasterProcedureScheduler queue;
+ private Map<Long, Procedure<?>> procedures;
+
@Rule
public TestName name = new TestName();
@Before
public void setUp() throws IOException {
- queue = new MasterProcedureScheduler(pid -> null);
+ procedures = new HashMap<>();
+ queue = new MasterProcedureScheduler(procedures::get);
queue.start();
}
@@ -95,7 +100,7 @@ public class TestMasterProcedureScheduler {
// insert items
for (int j = 1; j <= NUM_ITEMS; ++j) {
queue.addBack(new TestTableProcedure(i * 1000 + j, tableName,
- TableProcedureInterface.TableOperationType.EDIT));
+ TableProcedureInterface.TableOperationType.REGION_EDIT));
assertEquals(++count, queue.size());
}
}
@@ -222,7 +227,7 @@ public class TestMasterProcedureScheduler {
assertEquals(null, queue.poll(0));
// Release the write lock and acquire the read lock
- queue.wakeTableExclusiveLock(proc, tableName);
+ releaseTableExclusiveLockAndComplete(proc, tableName);
// Fetch the 2nd item and take the read lock
Procedure<?> rdProc = queue.poll();
@@ -248,7 +253,7 @@ public class TestMasterProcedureScheduler {
assertEquals(null, queue.poll(0));
// Release the write lock and acquire the read lock
- queue.wakeTableExclusiveLock(wrProc, tableName);
+ releaseTableExclusiveLockAndComplete(wrProc, tableName);
// Fetch the 4th item and take the read lock
rdProc = queue.poll();
@@ -641,6 +646,13 @@ public class TestMasterProcedureScheduler {
childProcs);
}
+ private void releaseTableExclusiveLockAndComplete(Procedure<?> proc, TableName tableName) {
+ // release xlock
+ queue.wakeTableExclusiveLock(proc, tableName);
+ // mark the procedure as complete
+ queue.completionCleanup(proc);
+ }
+
private void testInheritedXLockAndChildrenSharedLock(final TableName tableName,
final TestTableProcedure rootProc, final TestRegionProcedure[] childProcs) throws Exception {
queue.addBack(rootProc);
@@ -671,13 +683,13 @@ public class TestMasterProcedureScheduler {
assertEquals(null, queue.poll(0));
// release xlock
- queue.wakeTableExclusiveLock(parentProc, tableName);
+ releaseTableExclusiveLockAndComplete(parentProc, tableName);
// fetch the other xlock proc
Procedure<?> proc = queue.poll();
assertEquals(100, proc.getProcId());
assertEquals(false, queue.waitTableExclusiveLock(proc, tableName));
- queue.wakeTableExclusiveLock(proc, tableName);
+ releaseTableExclusiveLockAndComplete(proc, tableName);
}
@Test
@@ -694,29 +706,35 @@ public class TestMasterProcedureScheduler {
// simulate 3 procedures: 1 (root), (2) child of root, (3) child of proc-2
testInheritedXLockAndChildrenXLock(tableName,
new TestTableProcedure(1, tableName, TableProcedureInterface.TableOperationType.EDIT),
+ new TestTableProcedure(1, 1, 2, tableName, TableProcedureInterface.TableOperationType.EDIT),
new TestTableProcedure(1, 2, 3, tableName, TableProcedureInterface.TableOperationType.EDIT));
}
private void testInheritedXLockAndChildrenXLock(final TableName tableName,
- final TestTableProcedure rootProc, final TestTableProcedure childProc) throws Exception {
+ final TestTableProcedure rootProc, final TestTableProcedure... childProcs) throws Exception {
+ procedures.put(rootProc.getProcId(), rootProc);
+ for (TestTableProcedure childProc : childProcs) {
+ procedures.put(childProc.getProcId(), childProc);
+ }
queue.addBack(rootProc);
// fetch and acquire first xlock proc
Procedure<?> parentProc = queue.poll();
- assertEquals(rootProc, parentProc);
+ assertSame(rootProc, parentProc);
assertEquals(false, queue.waitTableExclusiveLock(parentProc, tableName));
+ TestTableProcedure childProc = childProcs[childProcs.length - 1];
// add child procedure
queue.addFront(childProc);
// fetch the other xlock proc
Procedure<?> proc = queue.poll();
- assertEquals(childProc, proc);
+ assertSame(childProc, proc);
assertEquals(false, queue.waitTableExclusiveLock(proc, tableName));
- queue.wakeTableExclusiveLock(proc, tableName);
+ releaseTableExclusiveLockAndComplete(proc, tableName);
// release xlock
- queue.wakeTableExclusiveLock(parentProc, tableName);
+ releaseTableExclusiveLockAndComplete(proc, tableName);
}
@Test
@@ -744,7 +762,7 @@ public class TestMasterProcedureScheduler {
assertEquals(1, proc.getProcId());
// release the xlock
- queue.wakeTableExclusiveLock(proc, tableName);
+ releaseTableExclusiveLockAndComplete(proc, tableName);
proc = queue.poll();
assertEquals(2, proc.getProcId());
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureSchedulerConcurrency.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureSchedulerConcurrency.java
index dfb1a968b10..5729753b84b 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureSchedulerConcurrency.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureSchedulerConcurrency.java
@@ -264,7 +264,9 @@ public class TestMasterProcedureSchedulerConcurrency {
boolean waiting = true;
while (waiting && queue.size() > 0) {
proc = queue.poll(100000000L);
- if (proc == null) continue;
+ if (proc == null) {
+ continue;
+ }
switch (getTableOperationType(proc)) {
case CREATE:
case DELETE:
@@ -294,6 +296,7 @@ public class TestMasterProcedureSchedulerConcurrency {
default:
throw new UnsupportedOperationException();
}
+ queue.completionCleanup(proc);
}
public TableName getTableName(Procedure proc) {