WL#9545 Clusterj automatic reconnect on cluster failure

post-push fix for pb2 failures

clusterj-core/src/main/java/com/mysql/clusterj/core/store/ClusterConnection.java
  add method closing() to separate marking dbs as closing versus force close

clusterj-tie/src/main/java/com/mysql/clusterj/tie/ClusterConnectionImpl.java
  implement closing() to mark dbs as closing

clusterj-core/src/main/java/com/mysql/clusterj/core/SessionFactoryImpl.java
  use closing() method to mark all cluster connections then close them

clusterj-test/src/main/java/testsuite/clusterj/ReconnectTest.java
  change misbehaving thread to catch any exception and terminate instead of looping

Author: Craig L Russell

storage/ndb/clusterj/clusterj-core/src/main/java/com/mysql/clusterj/core/SessionFactoryImpl.java

@@ -679,6 +679,19 @@ public void reconnect(int timeout) {
             reconnectThread.start();
         }
     }
+
+    protected static int countSessions(SessionFactoryImpl factory) {
+        return countSessions(factory.getConnectionPoolSessionCounts());
+    }
+
+    protected static int countSessions(List<Integer> sessionCounts) {
+        int result = 0;
+        for (int i: sessionCounts) {
+            result += i;
+        }
+        return result;
+    }
+
     protected static class ReconnectThread implements Runnable {
         SessionFactoryImpl factory;
         ReconnectThread(SessionFactoryImpl factory) {
@@ -689,23 +702,25 @@ public void run() {
             boolean done = false;
             int iterations = factory.CLUSTER_RECONNECT_TIMEOUT;
             while (!done && iterations-- > 0) {
-                done = true;
-                sessionCounts = factory.getConnectionPoolSessionCounts();
-                for (int i: sessionCounts) {
-                    if (i != 0) {
-                        done = false;
-                    }
-                }
+                done = countSessions(sessionCounts) == 0;
                 if (!done) {
                     logger.info(local.message("INFO_Reconnect_wait", sessionCounts.toString()));
                     sleep(1000);
+                    sessionCounts = factory.getConnectionPoolSessionCounts();
                 }
             }
             if (!done) {
                 // timed out waiting for sessions to close
                 logger.warn(local.message("WARN_Reconnect_timeout", sessionCounts.toString()));
             }
             logger.warn(local.message("WARN_Reconnect_closing"));
+            // mark all cluster connections as closing
+            for (ClusterConnection clusterConnection: factory.pooledConnections) {
+                clusterConnection.closing();
+            }
+            // wait for connections to close on their own
+            sleep(1000);
+            // hard close connections that didn't close on their own
             for (ClusterConnection clusterConnection: factory.pooledConnections) {
                 clusterConnection.close();
             }

storage/ndb/clusterj/clusterj-core/src/main/java/com/mysql/clusterj/core/store/ClusterConnection.java

@@ -1,5 +1,5 @@
 /*
- *  Copyright (c) 2010, 2015, Oracle and/or its affiliates. All rights reserved.
+ *  Copyright (c) 2010, 2017, Oracle and/or its affiliates. All rights reserved.
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License as published by
@@ -30,6 +30,8 @@ public interface ClusterConnection {
 
     public void waitUntilReady(int connectTimeoutBefore, int connectTimeoutAfter);
 
+    public void closing();
+
     public void close();
 
     public int dbCount();

storage/ndb/clusterj/clusterj-test/src/main/java/testsuite/clusterj/ReconnectTest.java

@@ -26,6 +26,7 @@
 import java.util.TreeSet;
 
 import com.mysql.clusterj.ClusterJDatastoreException;
+import com.mysql.clusterj.ClusterJException;
 import com.mysql.clusterj.ClusterJUserException;
 import com.mysql.clusterj.Query;
 import com.mysql.clusterj.Session;
@@ -35,7 +36,6 @@
 import testsuite.clusterj.model.Order;
 import testsuite.clusterj.model.OrderLine;
 
-@org.junit.Ignore("disable test until diagnosis of failure")
 public class ReconnectTest extends AbstractClusterJModelTest {
 
     @Override
@@ -240,11 +240,10 @@ public void run() {
                 Query<OrderLine> queryOrder = session.createQuery(queryOrderType);
                 queryOrder.setParameter("orderId", 0);
                 queryOrder.getResultList();
-                } catch (ClusterJUserException cjue) {
+                sleep(100);
+                } catch (ClusterJException cje) {
+                    // the exception might be any of several exceptions when disconnecting/reconnecting
                     done = true;
-                    if (!cjue.getMessage().contains("this Db is closing")) {
-                        if (getDebug()) { System.out.print(" Misbehaving caught exception: " + cjue.getMessage()); }
-                    }
                 }
             }
         }

storage/ndb/clusterj/clusterj-tie/src/main/java/com/mysql/clusterj/tie/ClusterConnectionImpl.java

@@ -103,6 +103,8 @@ public class ClusterConnectionImpl
     /** The dictionary used to create NdbRecords */
     Dictionary dictionaryForNdbRecord = null;
 
+    private boolean isClosing = false;
+
     private long[] autoIncrement;
 
     private static final String USE_SMART_VALUE_HANDLER_NAME = "com.mysql.clusterj.UseSmartValueHandler";
@@ -223,23 +225,31 @@ protected static void throwError(Object returnCode, Ndb_cluster_connection clust
         throw new ClusterJDatastoreException(msg);
     }
 
-    public void close() {
+    public void closing() {
+        this.isClosing = true;
         if (clusterConnection != null) {
             logger.info(local.message("INFO_Close_Cluster_Connection", connectString, nodeId));
             if (dbs.size() > 0) {
                 for (DbImpl db: dbs.keySet()) {
                     // mark all dbs as closing so no more operations will start
                     db.closing();
                 }
-                dbForNdbRecord.closing();
-
-                if (dbs.size() != 0) {
-                    // sleep for 1000 milliseconds to allow operations in other threads to terminate
-                    sleep(1000);
-                    Map<Db, Object> dbsToClose = new IdentityHashMap<Db, Object>(dbs);
-                    for (Db db: dbsToClose.keySet()) {
-                        db.close();
-                    }
+            }
+            dbForNdbRecord.closing();
+        }
+    }
+
+    public void close() {
+        if (clusterConnection != null) {
+            if (!this.isClosing) {
+                this.closing();
+                // sleep for 1000 milliseconds to allow operations in other threads to terminate
+                sleep(1000);
+            }
+            if (dbs.size() != 0) {
+                Map<Db, Object> dbsToClose = new IdentityHashMap<Db, Object>(dbs);
+                for (Db db: dbsToClose.keySet()) {
+                    db.close();
                 }
             }
             for (NdbRecordImpl ndbRecord: ndbRecordImplMap.values()) {

storage/ndb/clusterj/clusterj-tie/src/test/java/testsuite/clusterj/tie/ReconnectTest.java

@@ -17,7 +17,6 @@
 
 package testsuite.clusterj.tie;
 
-@org.junit.Ignore("disable test until diagnosis of failure")
 public class ReconnectTest extends testsuite.clusterj.ReconnectTest {
 
 }