Fix race that causes "lease id cannot be null or empty" illegal argument

exception from storage.

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

@@ -387,7 +387,7 @@ private Boolean acquireLeaseSync(AzureBlobLease lease) throws Exception
     	this.host.logWithHostAndPartition(Level.FINE, lease.getPartitionId(), "Acquiring lease");
 
     	CloudBlockBlob leaseBlob = lease.getBlob();
-    	boolean retval = true;
+    	boolean succeeded = true;
     	String newLeaseId = EventProcessorHost.safeCreateUUID();
     	if ((newLeaseId == null) || newLeaseId.isEmpty())
     	{
@@ -400,31 +400,48 @@ private Boolean acquireLeaseSync(AzureBlobLease lease) throws Exception
 	    	if (leaseBlob.getProperties().getLeaseState() == LeaseState.LEASED)
 	    	{
 	    		this.host.logWithHostAndPartition(Level.FINER, lease.getPartitionId(), "changeLease");
-	    		newToken = leaseBlob.changeLease(newLeaseId, AccessCondition.generateLeaseCondition(lease.getToken()));
+	    		if ((lease.getToken() == null) || lease.getToken().isEmpty())
+	    		{
+	    			// We reach here in a race condition: when this instance of EventProcessorHost scanned the
+	    			// lease blobs, this partition was unowned (token is empty) but between then and now, another
+	    			// instance of EPH has established a lease (getLeaseState() is LEASED). We normally enforce
+	    			// that we only steal the lease if it is still owned by the instance which owned it when we
+	    			// scanned, but we can't do that when we don't know who owns it. The safest thing to do is just
+	    			// fail the acquisition. If that means that one EPH instance gets more partitions than it should,
+	    			// rebalancing will take care of that quickly enough.
+	    			succeeded = false;
+	    		}
+	    		else
+	    		{
+		    		newToken = leaseBlob.changeLease(newLeaseId, AccessCondition.generateLeaseCondition(lease.getToken()));
+	    		}
 	    	}
 	    	else
 	    	{
 	    		this.host.logWithHostAndPartition(Level.FINER, lease.getPartitionId(), "acquireLease");
 	    		newToken = leaseBlob.acquireLease(AzureStorageCheckpointLeaseManager.leaseDurationInSeconds, newLeaseId);
 	    	}
-	    	lease.setToken(newToken);
-	    	lease.setOwner(this.host.getHostName());
-	    	lease.incrementEpoch(); // Increment epoch each time lease is acquired or stolen by a new host
-	    	uploadLease(lease, leaseBlob, AccessCondition.generateLeaseCondition(lease.getToken()), UploadActivity.Acquire);
+	    	if (succeeded)
+	    	{
+		    	lease.setToken(newToken);
+		    	lease.setOwner(this.host.getHostName());
+		    	lease.incrementEpoch(); // Increment epoch each time lease is acquired or stolen by a new host
+		    	uploadLease(lease, leaseBlob, AccessCondition.generateLeaseCondition(lease.getToken()), UploadActivity.Acquire);
+	    	}
     	}
     	catch (StorageException se)
     	{
     		if (wasLeaseLost(se, lease.getPartitionId()))
     		{
-    			retval = false;
+    			succeeded = false;
     		}
     		else
     		{
     			throw se;
     		}
     	}
 
-    	return retval;
+    	return succeeded;
     }
 
     @Override