Uploaded image for project: 'Phoenix'
  1. Phoenix
  2. PHOENIX-4809

connectionQueue never cleared in ConnectionQueryServicesImpl when lease renewal is disabled/unsupported

Details

    • Bug
    • Status: Closed
    • Major
    • Resolution: Fixed
    • None
    • 4.14.1, 5.1.0
    • None
    • None

    Description

      When we create a new PhoenixConnection, we update connectionQueues in CQSI:

          @Override
          public void addConnection(PhoenixConnection connection) throws SQLException {
              connectionQueues.get(getQueueIndex(connection)).add(new WeakReference<PhoenixConnection>(connection));
              if (returnSequenceValues) {
                  synchronized (connectionCountLock) {
                      connectionCount++;
                  }
              }
          }

      We use connectionQueues to determine what needs lease renewal done.

      However, when the user closes a connection, this datastructure is never cleaned up.

          @Override
          public void removeConnection(PhoenixConnection connection) throws SQLException {
              if (returnSequenceValues) {
                  ConcurrentMap<SequenceKey,Sequence> formerSequenceMap = null;
                  synchronized (connectionCountLock) {
                      if (--connectionCount <= 0) {
                          if (!this.sequenceMap.isEmpty()) {
                              formerSequenceMap = this.sequenceMap;
                              this.sequenceMap = Maps.newConcurrentMap();
                          }
                      }
                      if (connectionCount < 0) {
                          connectionCount = 0;
                      }
                  }
                  // Since we're using the former sequenceMap, we can do this outside
                  // the lock.
                  if (formerSequenceMap != null) {
                      // When there are no more connections, attempt to return any sequences
                      returnAllSequences(formerSequenceMap);
                  }
              } else if (shouldThrottleNumConnections){ //still need to decrement connection count
                  synchronized (connectionCountLock) {
                      if (connectionCount > 0) {
                          --connectionCount;
                      }
                  }
              }
          }

      Running a test now, but seems to be the case on master.

      Attachments

        1. PHOENIX-4809.001.patch
          6 kB
          Josh Elser

        Activity

          hudson Hudson added a comment -

          FAILURE: Integrated in Jenkins build Phoenix-omid2 #89 (See https://builds.apache.org/job/Phoenix-omid2/89/)
          PHOENIX-4809 Only cache PhoenixConnections when lease renewal is on (elserj: rev b9fc468e5710b7cc47ee086c231f435014c04c7e)

          • (edit) phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
          • (add) phoenix-core/src/it/java/org/apache/phoenix/query/ConnectionCachingIT.java
          hudson Hudson added a comment - FAILURE: Integrated in Jenkins build Phoenix-omid2 #89 (See https://builds.apache.org/job/Phoenix-omid2/89/ ) PHOENIX-4809 Only cache PhoenixConnections when lease renewal is on (elserj: rev b9fc468e5710b7cc47ee086c231f435014c04c7e) (edit) phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java (add) phoenix-core/src/it/java/org/apache/phoenix/query/ConnectionCachingIT.java
          hudson Hudson added a comment -

          FAILURE: Integrated in Jenkins build PreCommit-PHOENIX-Build #1930 (See https://builds.apache.org/job/PreCommit-PHOENIX-Build/1930/)
          PHOENIX-4809 Only cache PhoenixConnections when lease renewal is on (elserj: rev f46e8bbcd48674f7de217feeac2679a593b042be)

          • (add) phoenix-core/src/it/java/org/apache/phoenix/query/ConnectionCachingIT.java
          • (edit) phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
          hudson Hudson added a comment - FAILURE: Integrated in Jenkins build PreCommit-PHOENIX-Build #1930 (See https://builds.apache.org/job/PreCommit-PHOENIX-Build/1930/ ) PHOENIX-4809 Only cache PhoenixConnections when lease renewal is on (elserj: rev f46e8bbcd48674f7de217feeac2679a593b042be) (add) phoenix-core/src/it/java/org/apache/phoenix/query/ConnectionCachingIT.java (edit) phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
          elserj Josh Elser added a comment -

          Thanks for the quick review, Ankit!

          elserj Josh Elser added a comment - Thanks for the quick review, Ankit!
          ankit@apache.org Ankit Singhal added a comment -

          +1 elserj, Thanks for the test as well

          ankit@apache.org Ankit Singhal added a comment - +1 elserj , Thanks for the test as well
          elserj Josh Elser added a comment -

          ankit@apache.org what's your thought on this? Straightforward fix that as you had mentioned earlier. I got an IT to reproduce the problem as well.

          elserj Josh Elser added a comment - ankit@apache.org what's your thought on this? Straightforward fix that as you had mentioned earlier. I got an IT to reproduce the problem as well.
          elserj Josh Elser added a comment -

          Maybe related to lease renewal not being on?

          Ah, yeah, that's the rub. If the RenewLeaseTask doesn't run, nothing drains these queues.

          elserj Josh Elser added a comment - Maybe related to lease renewal not being on? Ah, yeah, that's the rub. If the RenewLeaseTask doesn't run, nothing drains these queues.
          elserj Josh Elser added a comment -

          I ran a quick test locally, the odd thing is that I see fewer WeakReference<PhoenixConnection> objects than I'd expect..

          public class PhoenixQuery {
            private static final String URL = "jdbc:phoenix:localhost:2181:/hbase-1.4";
          
            public static void main(String[] args) throws Exception {
              ExecutorService svc = Executors.newFixedThreadPool(8);
              try (Connection conn = DriverManager.getConnection(URL, "", "")) {
                PhoenixConnection phxConn = conn.unwrap(PhoenixConnection.class);
                ConnectionQueryServicesImpl queryServices = (ConnectionQueryServicesImpl) phxConn.getQueryServices();
                System.out.println(numConnectionsCached(queryServices));
              }
              CountDownLatch latch = new CountDownLatch(8);
              for (int i = 0; i < 8; i++) {
                svc.submit(new Runnable() {
                  @Override public void run() {
                    for (int j = 0; j < 1_000_000; j++) {
                      try (Connection conn = DriverManager.getConnection(URL, "", "")) {
                          conn.close();
                      } catch (Exception e) {
                        System.out.println("Caught error");
                      }
                    }
                    latch.countDown();
                  }
                });
              }
              latch.await();
          
              try (Connection conn = DriverManager.getConnection(URL, "", "")) {
                PhoenixConnection phxConn = conn.unwrap(PhoenixConnection.class);
                ConnectionQueryServicesImpl queryServices = (ConnectionQueryServicesImpl) phxConn.getQueryServices();
                System.out.println(numConnectionsCached(queryServices));
              }
            }
          
            private static long numConnectionsCached(ConnectionQueryServicesImpl cqs) {
              try {
                Field f = ConnectionQueryServicesImpl.class.getDeclaredField("connectionQueues");
                f.setAccessible(true);
                List<LinkedBlockingQueue<WeakReference<PhoenixConnection>>> list = (List<LinkedBlockingQueue<WeakReference<PhoenixConnection>>>) f.get(cqs);
                return list.stream().collect(Collectors.summingLong(LinkedBlockingQueue::size));
              } catch (Exception e) {
                e.printStackTrace(System.err);
                throw new RuntimeException(e);
              }
            }
          }

          I'd expect to see 8M objects cached, but I'm actually seeing 615,696. Something must clean this up somehow.. Maybe related to lease renewal not being on?

          elserj Josh Elser added a comment - I ran a quick test locally, the odd thing is that I see fewer WeakReference<PhoenixConnection> objects than I'd expect.. public class PhoenixQuery {   private static final String URL = "jdbc:phoenix:localhost:2181:/hbase-1.4" ;   public static void main( String [] args) throws Exception {     ExecutorService svc = Executors.newFixedThreadPool(8);     try (Connection conn = DriverManager.getConnection(URL, "", " ")) {       PhoenixConnection phxConn = conn.unwrap(PhoenixConnection.class);       ConnectionQueryServicesImpl queryServices = (ConnectionQueryServicesImpl) phxConn.getQueryServices();       System .out.println(numConnectionsCached(queryServices));     }     CountDownLatch latch = new CountDownLatch(8);     for ( int i = 0; i < 8; i++) {       svc.submit( new Runnable () {         @Override public void run() {           for ( int j = 0; j < 1_000_000; j++) {             try (Connection conn = DriverManager.getConnection(URL, "", " ")) {                 conn.close();             } catch (Exception e) {               System .out.println( "Caught error" );             }           }           latch.countDown();         }       });     }     latch.await();     try (Connection conn = DriverManager.getConnection(URL, "", " ")) {       PhoenixConnection phxConn = conn.unwrap(PhoenixConnection.class);       ConnectionQueryServicesImpl queryServices = (ConnectionQueryServicesImpl) phxConn.getQueryServices();       System .out.println(numConnectionsCached(queryServices));     }   }   private static long numConnectionsCached(ConnectionQueryServicesImpl cqs) {     try {       Field f = ConnectionQueryServicesImpl. class. getDeclaredField( "connectionQueues" );       f.setAccessible( true );       List<LinkedBlockingQueue<WeakReference<PhoenixConnection>>> list = (List<LinkedBlockingQueue<WeakReference<PhoenixConnection>>>) f.get(cqs);       return list.stream().collect(Collectors.summingLong(LinkedBlockingQueue::size));     } catch (Exception e) {       e.printStackTrace( System .err);       throw new RuntimeException(e);     }   } } I'd expect to see 8M objects cached, but I'm actually seeing 615,696. Something must clean this up somehow.. Maybe related to lease renewal not being on?

          People

            elserj Josh Elser
            elserj Josh Elser
            Votes:
            0 Vote for this issue
            Watchers:
            6 Start watching this issue

            Dates

              Created:
              Updated:
              Resolved: