Merge pull request #38 from lasanthaS/siddhi-4.x.x

Fix timeout issue while waiting on missed records
siddhi-io · Oct 21, 2019 · 8fecfef · 8fecfef
2 parents 6c428ed + 02e0e2d
commit 8fecfef
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 10 deletions.
diff --git a/...extension/siddhi/io/cdc/source/polling/strategies/WaitOnMissingRecordPollingStrategy.java b/...extension/siddhi/io/cdc/source/polling/strategies/WaitOnMissingRecordPollingStrategy.java
@@ -66,7 +66,6 @@ public void poll() {
         Connection connection = getConnection();
         PreparedStatement statement = null;
         ResultSet resultSet = null;
-        boolean breakOnMissingRecord = false;
         int waitingFor = -1;
         long waitingFrom = -1;
         try {
@@ -115,12 +114,13 @@ public void poll() {
                                 waitingFrom = System.currentTimeMillis();
                             }
 
-                            isTimedout = waitTimeout > -1 && waitingFrom + waitTimeout < System.currentTimeMillis();
+                            long waitEndTimestamp = waitTimeout > -1 ?
+                                    waitingFrom + (waitTimeout * 1000L) : Long.MAX_VALUE;
+                            isTimedout = waitEndTimestamp < System.currentTimeMillis();
                             if (!isTimedout) {
                                 log.debug("Missed record found at " + waitingFor + " in table " + tableName +
                                         ". Hence pausing the process and " + "retry in " + pollingInterval +
                                         " seconds.");
-                                breakOnMissingRecord = true;
                                 break;
                             }
                         }
@@ -150,9 +150,6 @@ public void poll() {
                     CDCPollingUtil.cleanupConnection(resultSet, null, null);
                 }
                 try {
-                    if (breakOnMissingRecord) {
-                        breakOnMissingRecord = false;
-                    }
                     Thread.sleep((long) pollingInterval * 1000);
                 } catch (InterruptedException e) {
                     log.error(buildError("Error while polling the table %s.", tableName), e);

diff --git a/...onent/src/test/java/org/wso2/extension/siddhi/io/cdc/source/TestCaseOfCDCPollingMode.java b/...onent/src/test/java/org/wso2/extension/siddhi/io/cdc/source/TestCaseOfCDCPollingMode.java
@@ -369,10 +369,10 @@ public void receive(Event[] events) {
 
         // Do inserts and wait CDC app to capture the events.
         InputHandler inputHandler = siddhiAppRuntime.getInputHandler("inputStream");
-        Object[] ann = new Object[]{11, "Ann"};
-        Object[] bob = new Object[]{12, "Bob"};
-        Object[] charles = new Object[]{13, "Charles"};
-        Object[] david = new Object[]{14, "David"};
+        Object[] ann = new Object[]{1, "Ann"};
+        Object[] bob = new Object[]{2, "Bob"};
+        Object[] charles = new Object[]{3, "Charles"};
+        Object[] david = new Object[]{4, "David"};
 
         inputHandler.send(ann);
         inputHandler.send(bob);