Fix lease scanner issues when Storage unreachable (#434)

This fix is for issue #432. There are two parts:

AzureStorageCheckpointLeaseManager performs certain Storage actions within a forEach. If those actions fail, the StorageException gets wrapped in a NoSuchElementException. Catch those and strip off the NoSuchElementException, then handle the StorageException in the existing way.

The unexpected NoSuchElementExceptions were not being caught anywhere and the scanner thread was dying without rescheduling itself. Added code in PartitionMananger.scan to catch any exceptions that leak out of PartitionScanner and reschedule the scanner unless the host instance is shutting down.

Author: JamesBirdsall

azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/AzureStorageCheckpointLeaseManager.java

@@ -22,6 +22,7 @@
 import java.util.Hashtable;
 import java.util.Iterator;
 import java.util.List;
+import java.util.NoSuchElementException;
 import java.util.concurrent.*;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -317,10 +318,16 @@ public CompletableFuture<List<BaseLease>> getAllLeases() {
 						(bp.getLeaseState() == LeaseState.LEASED)));
 			});
 			future = CompletableFuture.completedFuture(infos);
-		} catch (URISyntaxException | StorageException e) {
-            TRACE_LOGGER.warn(this.hostContext.withHost("Failure while getting lease state details"), e);
+		} catch (Exception e) {
+    		Throwable effective = e;
+    		if (e instanceof NoSuchElementException) {
+    			// If there is a StorageException in the forEach, it arrives wrapped in a NoSuchElementException.
+    			// Strip the misleading NoSuchElementException to provide a meaningful error for the user.
+    			effective = e.getCause();
+    		}
+            TRACE_LOGGER.warn(this.hostContext.withHost("Failure while getting lease state details"), effective);
             future = new CompletableFuture<List<BaseLease>>();
-            future.completeExceptionally(LoggingUtils.wrapException(e, EventProcessorHostActionStrings.GETTING_LEASE));
+            future.completeExceptionally(LoggingUtils.wrapException(effective, EventProcessorHostActionStrings.GETTING_LEASE));
 		}
 
     	return future;

azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/PartitionManager.java

@@ -290,31 +290,42 @@ private Void scan(boolean isFirst) {
         TRACE_LOGGER.debug(this.hostContext.withHost("Starting lease scan"));
         long start = System.currentTimeMillis();
 
-        (new PartitionScanner(this.hostContext, (lease) -> this.pumpManager.addPump(lease), this)).scan(isFirst)
-                .whenCompleteAsync((didSteal, e) ->
-                {
-                    TRACE_LOGGER.debug(this.hostContext.withHost("Scanning took " + (System.currentTimeMillis() - start)));
-                	if ((e != null) && !(e instanceof ClosingException)) {
-                		TRACE_LOGGER.warn(this.hostContext.withHost("Lease scanner got exception"), e);
-                	}
-
-                    onPartitionCheckCompleteTestHook();
-
-                    // Schedule the next scan unless we are shutting down.
-                    if (!this.getIsClosingOrClosed()) {
-                        int seconds = didSteal ? this.hostContext.getPartitionManagerOptions().getFastScanIntervalInSeconds() :
-                                this.hostContext.getPartitionManagerOptions().getSlowScanIntervalInSeconds();
-                        if (isFirst) {
-                            seconds = this.hostContext.getPartitionManagerOptions().getStartupScanDelayInSeconds();
-                        }
-                        synchronized (this.scanFutureSynchronizer) {
-                            this.scanFuture = this.hostContext.getExecutor().schedule(() -> scan(false), seconds, TimeUnit.SECONDS);
-                        }
-                        TRACE_LOGGER.debug(this.hostContext.withHost("Scheduling lease scanner in " + seconds));
-                    } else {
-                        TRACE_LOGGER.warn(this.hostContext.withHost("Not scheduling lease scanner due to shutdown"));
-                    }
-                }, this.hostContext.getExecutor());
+        try {
+	        (new PartitionScanner(this.hostContext, (lease) -> this.pumpManager.addPump(lease), this)).scan(isFirst)
+	                .whenCompleteAsync((didSteal, e) ->
+	                {
+	                    TRACE_LOGGER.debug(this.hostContext.withHost("Scanning took " + (System.currentTimeMillis() - start)));
+	                	if ((e != null) && !(e instanceof ClosingException)) {
+	                		TRACE_LOGGER.warn(this.hostContext.withHost("Lease scanner got exception"), e);
+	                	}
+	
+	                    onPartitionCheckCompleteTestHook();
+	
+	                    // Schedule the next scan unless we are shutting down.
+	                    if (!this.getIsClosingOrClosed()) {
+	                        int seconds = didSteal ? this.hostContext.getPartitionManagerOptions().getFastScanIntervalInSeconds() :
+	                                this.hostContext.getPartitionManagerOptions().getSlowScanIntervalInSeconds();
+	                        if (isFirst) {
+	                            seconds = this.hostContext.getPartitionManagerOptions().getStartupScanDelayInSeconds();
+	                        }
+	                        synchronized (this.scanFutureSynchronizer) {
+	                            this.scanFuture = this.hostContext.getExecutor().schedule(() -> scan(false), seconds, TimeUnit.SECONDS);
+	                        }
+	                        TRACE_LOGGER.debug(this.hostContext.withHost("Scheduling lease scanner in " + seconds));
+	                    } else {
+	                        TRACE_LOGGER.warn(this.hostContext.withHost("Not scheduling lease scanner due to shutdown"));
+	                    }
+	                }, this.hostContext.getExecutor());
+        } catch (Exception e) {
+    		TRACE_LOGGER.error(this.hostContext.withHost("Lease scanner threw directly"), e);
+        	if (!this.getIsClosingOrClosed()) {
+                int seconds = this.hostContext.getPartitionManagerOptions().getSlowScanIntervalInSeconds();
+	            synchronized (this.scanFutureSynchronizer) {
+	                this.scanFuture = this.hostContext.getExecutor().schedule(() -> scan(false), seconds, TimeUnit.SECONDS);
+	            }
+	            TRACE_LOGGER.debug(this.hostContext.withHost("Forced schedule of lease scanner in " + seconds));
+        	}
+        }
 
         return null;
     }