HBASE-13970 NPE during compaction in trunk

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java

@@ -23,6 +23,7 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -221,6 +222,24 @@ public abstract class Compactor {
     return store.getCoprocessorHost().preCompact(store, scanner, scanType, request);
   }
 
+  /**
+   * Used to prevent compaction name conflict when multiple compactions running parallel on the
+   * same store.
+   */
+  private static final AtomicInteger NAME_COUNTER = new AtomicInteger(0);
+
+  private String generateCompactionName() {
+    int counter;
+    for (;;) {
+      counter = NAME_COUNTER.get();
+      int next = counter == Integer.MAX_VALUE ? 0 : counter + 1;
+      if (NAME_COUNTER.compareAndSet(counter, next)) {
+        break;
+      }
+    }
+    return store.getRegionInfo().getRegionNameAsString() + "#"
+        + store.getFamily().getNameAsString() + "#" + counter;
+  }
   /**
    * Performs the compaction.
    * @param scanner Where to read from.
@@ -242,8 +261,7 @@ public abstract class Compactor {
     if (LOG.isDebugEnabled()) {
       lastMillis = EnvironmentEdgeManager.currentTime();
     }
-    String compactionName =
-        store.getRegionInfo().getRegionNameAsString() + "#" + store.getFamily().getNameAsString();
+    String compactionName = generateCompactionName();
     long now = 0;
     boolean hasMore;
     ScannerContext scannerContext =