Repository: phoenix
Updated Branches:
  refs/heads/4.11-HBase-1.1 25c959812 -> cc034d2b3


PHOENIX-4024 Renew lease thread names should be unique across various 
ConnectionQueryServices instances


Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/cc034d2b
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/cc034d2b
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/cc034d2b

Branch: refs/heads/4.11-HBase-1.1
Commit: cc034d2b311805948e3c27e421e888f9a2ba1141
Parents: 25c9598
Author: Samarth Jain <[email protected]>
Authored: Fri Jul 14 14:29:29 2017 -0700
Committer: Samarth Jain <[email protected]>
Committed: Fri Jul 14 14:29:29 2017 -0700

----------------------------------------------------------------------
 .../query/ConnectionQueryServicesImpl.java      | 25 ++++++++++++++++----
 1 file changed, 20 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc034d2b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
----------------------------------------------------------------------
diff --git 
a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
 
b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 16f6e39..fd1b64f 100644
--- 
a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++ 
b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@ -93,6 +93,7 @@ import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.annotation.concurrent.GuardedBy;
 
@@ -309,6 +310,11 @@ public class ConnectionQueryServicesImpl extends 
DelegateQueryServices implement
     // List of queues instead of a single queue to provide reduced contention 
via lock striping
     private final List<LinkedBlockingQueue<WeakReference<PhoenixConnection>>> 
connectionQueues;
     private ScheduledExecutorService renewLeaseExecutor;
+    /*
+     * We can have multiple instances of ConnectionQueryServices. By making 
the thread factory
+     * static, renew lease thread names will be unique across them.
+     */
+    private static final ThreadFactory renewLeaseThreadFactory = new 
RenewLeaseThreadFactory();
     private final boolean renewLeaseEnabled;
     private final boolean isAutoUpgradeEnabled;
     private final AtomicBoolean upgradeRequired = new AtomicBoolean(false);
@@ -3356,18 +3362,27 @@ public class ConnectionQueryServicesImpl extends 
DelegateQueryServices implement
 
     private void scheduleRenewLeaseTasks() {
         if (isRenewingLeasesEnabled()) {
-            ThreadFactory threadFactory =
-                    new ThreadFactoryBuilder().setDaemon(true)
-                    .setNameFormat("PHOENIX-SCANNER-RENEW-LEASE" + 
"-thread-%s").build();
             renewLeaseExecutor =
-                    Executors.newScheduledThreadPool(renewLeasePoolSize, 
threadFactory);
+                    Executors.newScheduledThreadPool(renewLeasePoolSize, 
renewLeaseThreadFactory);
             for (LinkedBlockingQueue<WeakReference<PhoenixConnection>> q : 
connectionQueues) {
                 renewLeaseExecutor.scheduleAtFixedRate(new RenewLeaseTask(q), 
0,
-                        renewLeaseTaskFrequency, TimeUnit.MILLISECONDS);
+                    renewLeaseTaskFrequency, TimeUnit.MILLISECONDS);
             }
         }
     }
 
+    private static class RenewLeaseThreadFactory implements ThreadFactory {
+        private static final AtomicInteger threadNumber = new AtomicInteger(1);
+        private static final String NAME_PREFIX = 
"PHOENIX-SCANNER-RENEW-LEASE-thread-";
+
+        @Override
+        public Thread newThread(Runnable r) {
+            Thread t = new Thread(r, NAME_PREFIX + 
threadNumber.getAndIncrement());
+            t.setDaemon(true);
+            return t;
+        }
+    }
+
     private static int getSaltBuckets(TableAlreadyExistsException e) {
         PTable table = e.getTable();
         Integer sequenceSaltBuckets = table == null ? null : 
table.getBucketNum();

Reply via email to