diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureQueue.java hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureQueue.java index 0dd0c3d..63036ab 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureQueue.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureQueue.java @@ -283,7 +283,7 @@ public class MasterProcedureQueue implements ProcedureRunnableSet { if (queue != null) { lock.lock(); try { - if (queue.isEmpty() && !queue.isLocked()) { + if (queue.isEmpty() && queue.acquireDeleteLock()) { fairq.remove(table); // Remove the table lock @@ -307,7 +307,7 @@ public class MasterProcedureQueue implements ProcedureRunnableSet { void addFront(Procedure proc); void addBack(Procedure proc); Long poll(); - boolean isLocked(); + boolean acquireDeleteLock(); } /** @@ -356,6 +356,14 @@ public class MasterProcedureQueue implements ProcedureRunnableSet { } @Override + public synchronized boolean acquireDeleteLock() { + if (isLocked()) { + return false; + } + wlock = true; + return true; + } + public boolean isLocked() { synchronized (this) { return wlock || rlock > 0; diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureQueue.java hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureQueue.java index 0d00ff2..98116f8 100644 --- hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureQueue.java +++ hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestMasterProcedureQueue.java @@ -21,6 +21,7 @@ package org.apache.hadoop.hbase.master.procedure; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.ConcurrentHashMap; import java.util.ArrayList; @@ -35,7 +36,6 @@ import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.master.TableLockManager; import org.apache.hadoop.hbase.procedure2.Procedure; import org.apache.hadoop.hbase.testclassification.SmallTests; - import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -65,6 +65,56 @@ public class TestMasterProcedureQueue { assertEquals(0, queue.size()); } + @Test + public void testConcurrentCreateDelete() throws Exception { + final MasterProcedureQueue procQueue = queue; + final TableName table = TableName.valueOf("testtb"); + final AtomicBoolean running = new AtomicBoolean(true); + final AtomicBoolean failure = new AtomicBoolean(false); + Thread createThread = new Thread() { + @Override + public void run() { + try { + while (running.get() && !failure.get()) { + if (procQueue.tryAcquireTableWrite(table, "create")) { + procQueue.releaseTableWrite(table); + } + } + } catch (Throwable e) { + LOG.error("create failed", e); + failure.set(true); + } + } + }; + + Thread deleteThread = new Thread() { + @Override + public void run() { + try { + while (running.get() && !failure.get()) { + if (procQueue.tryAcquireTableWrite(table, "delete")) { + procQueue.releaseTableWrite(table); + } + procQueue.markTableAsDeleted(table); + } + } catch (Throwable e) { + LOG.error("delete failed", e); + failure.set(true); + } + } + }; + + createThread.start(); + deleteThread.start(); + for (int i = 0; i < 100 && running.get() && !failure.get(); ++i) { + Thread.sleep(100); + } + running.set(false); + createThread.join(); + deleteThread.join(); + assertEquals(false, failure.get()); + } + /** * Verify simple create/insert/fetch/delete of the table queue. */