Repository: Vantiq/vantiq-extension-sources (language: python)

Issue:
ExtJSdk: RECONNECT_REQUIRED/client.doCoreReconnect() sometimes fails with error from server
Server failure of the form "...because the source ... does not have an associated stream..."

The issue here is that there is a window in the server where a configuration change causes the source to restart (and possibly migrate to a different cluster member).  During that window, the source is unavailable.

Solution is to add retry to the underlying SDK.


Discussion / hints:

Fix patch:
diff --git a/extjsdk/src/main/java/io/vantiq/extjsdk/ExtensionWebSocketClient.java b/extjsdk/src/main/java/io/vantiq/extjsdk/ExtensionWebSocketClient.java
index 266b6205..349a7680 100644
--- a/extjsdk/src/main/java/io/vantiq/extjsdk/ExtensionWebSocketClient.java
+++ b/extjsdk/src/main/java/io/vantiq/extjsdk/ExtensionWebSocketClient.java
@@ -134,6 +134,9 @@ public class ExtensionWebSocketClient {
      * Whether it should automatically send a connection message after receiving a reconnect message
      */
     boolean autoReconnect = false;
+    
+    // Used by the listener to indicate that we've hit a window in reactivation logic.
+    boolean retryConnect = false;
 
     /**
      * The data to be used for authentication. This will be either a {@link String} containing an authentication token or
@@ -740,32 +743,50 @@ public synchronized CompletableFuture<Boolean> connectToSource() {
         }
         return sourceFuture;
     }
-
+    
+    private static final Integer MAX_RETRIES = 10;
     /**
      * Method called by reconnectHandlers that actually does the reconnect work.
      * @return  Returns boolean completable indicating if the reconnect was successful, used by the caller
      */
     public CompletableFuture<Boolean> doCoreReconnect() {
         return CompletableFuture.supplyAsync(() -> {
-            CompletableFuture<Boolean> success = connectToSource();
             boolean isReconnected = false;
-            try {
-                if ( !success.get(10, TimeUnit.SECONDS) ) {
-                    if (!isOpen()) {
-                        log.error("Failed to connect to server url .");
-                    } else if (!isAuthed()) {
-                        log.error("Failed to authenticate within 10 seconds using the given authentication data.");
+            // Ensure that retry is not turned on, possibly left from some previous connect attempt
+            retryConnect = false;
+            for (int retryCount = 0; retryCount < MAX_RETRIES; retryCount++) {
+                // This loop exists only for connection retry attempts.  In most cases, we break out of the loop
+                // immediately.  But when retries are necessary due to the restart window (between deactivation &
+                // reactivation), we will retry the connection attempt.  We limit those attempts since that restart
+                // window should not be very long.
+                
+                CompletableFuture<Boolean> success = connectToSource();
+                try {
+                    if (!success.get(10, TimeUnit.SECONDS)) {
+                        if (!isOpen()) {
+                            log.error("Failed to connect to server url .");
+                        } else if (!isAuthed()) {
+                            log.error("Failed to authenticate within 10 seconds using the given authentication data.");
+                        } else if (retryConnect) {
+                            // First, reset
+                            log.warn("Retrying connection (count: {}) due to hitting reactivation window", retryCount);
+                            retryConnect = false;
+                            continue;
+                        } else {
+                            log.error("Failed to connect within 10 seconds");
+                        }
                     } else {
-                        log.error("Failed to connect within 10 seconds");
+                        isReconnected = true;
                     }
-                } else {
-                    isReconnected = true;
+                } catch (InterruptedException | ExecutionException | TimeoutException e) {
+                    retryConnect = false;  // Just to make certain.
+                    log.error("Could not reconnect to source within 10 seconds: ", e);
                 }
-            } catch (InterruptedException | ExecutionException | TimeoutException e) {
-                log.error("Could not reconnect to source within 10 seconds: ", e);
+                break;
             }
-
+            
             return isReconnected;
+    
         });
     }
     
diff --git a/extjsdk/src/main/java/io/vantiq/extjsdk/ExtensionWebSocketListener.java b/extjsdk/src/main/java/io/vantiq/extjsdk/ExtensionWebSocketListener.java
index 14d91dd8..231d45df 100644
--- a/extjsdk/src/main/java/io/vantiq/extjsdk/ExtensionWebSocketListener.java
+++ b/extjsdk/src/main/java/io/vantiq/extjsdk/ExtensionWebSocketListener.java
@@ -22,6 +22,7 @@
 
 import java.io.EOFException;
 import java.net.ConnectException;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -269,6 +270,21 @@ public void onMessage(@NotNull WebSocket webSocket, ByteString bodyBytes) {
                 if (!client.isConnected() && (Integer) message.getStatus() >= 300) {
                     log.warn("Error occurred attempting to connect to source.");
                     log.debug("Error message was: {}", message);
+                    if (message.getBody() instanceof List) {
+                        List body = (List) message.getBody();
+                        if (body.size() >= 1 && body.get(0) instanceof Map) {
+                            Map errMap = (Map) body.get(0);
+                            if (errMap.containsKey("message")) {
+                                String errMsg = (String) errMap.get("message");
+                                if (errMsg.contains("does not have an associated stream")) {
+                                    // Then we've hit the window where the source is in the midst of
+                                    // re-activation/restart. Mark that the client should retry the connection.
+                                    // The client will handle the reconnect attempt once we mark things appropriately.
+                                    client.retryConnect = true;
+                                }
+                            }
+                        }
+                    }
                     client.sourceFuture.complete(false);
                 } else {
                     client.acknowledgeNotification();


Test patch:
diff --git a/jdbcSource/src/test/java/io/vantiq/extsrc/jdbcSource/TestJDBC.java b/jdbcSource/src/test/java/io/vantiq/extsrc/jdbcSource/TestJDBC.java
index 7e38d98b..aab82ffe 100644
--- a/jdbcSource/src/test/java/io/vantiq/extsrc/jdbcSource/TestJDBC.java
+++ b/jdbcSource/src/test/java/io/vantiq/extsrc/jdbcSource/TestJDBC.java
@@ -59,7 +59,7 @@ public class TestJDBC extends TestJDBCBase {
     static final String TIME = "9:24:18";
     static final String FORMATTED_TIMESTAMP = "2018-08-15T09:24:18.000-0700";
     static final String FORMATTED_TIME = "09:24:18.000-0800";
-    static final String VANTIQ_FORMATTED_TIMESTAMP = "2018-08-15T16:24:18Z";
+    static final String VANTIQ_FORMATTED_TIMESTAMP = "2018-08-15T16:24:18.000Z";
     
     // Queries to test oddball types
     static final String CREATE_TABLE_EXTENDED_TYPES = "create table TestTypes(id int, ts TIMESTAMP, testDate DATE, "
@@ -601,7 +601,7 @@ public void testVantiqDateFormatting() throws VantiqSQLException {
         String vantiqTimestamp = responseMap.get("timestamp").getAsString();
         
         // Check that the date was offset by adding 7 hours (since original date ends in 0700)
-        assert vantiqTimestamp.equals(VANTIQ_FORMATTED_TIMESTAMP);
+        assertEquals("Formatted time stamp", VANTIQ_FORMATTED_TIMESTAMP, vantiqTimestamp);
         
         // Delete the Type from VANTIQ
         deleteType();