EventHubs integration part 2 - spotbugs

eng/code-quality-reports/src/main/resources/checkstyle/checkstyle-suppressions.xml

@@ -22,14 +22,39 @@
   <!-- Public API already released with incorrect upper-case naming pattern -->
   <suppress checks="MethodName" files="AsymmetricEncryptionAlgorithm.java"/>
   <suppress checks="MethodName" files="KeyWrapAlgorithm.java"/>
+  <suppress checks="MethodName" files="PartitionSenderImpl.java"/>
   <suppress checks="MethodName" files="SignatureEncoding.java"/>
   <suppress checks="MethodName" files="SymmetricEncryptionAlgorithm.java"/>
   <suppress checks="MethodName" files="RsaSignature.java"/>
 
   <!-- Public API already released with incorrect constant variable naming -->
   <suppress checks="ConstantName" files="AlgorithmResolver.java"/>
+  <suppress checks="ConstantName" files="AmqpErrorCode.java"/>
+  <suppress checks="ConstantName" files="BatchErrorCodeStrings.java"/>
+  <suppress checks="ConstantName" files="ConnectionStringBuilder.java"/>
+  <suppress checks="ConstantName" files="EventProcessorHost.java"/>
+  <suppress checks="ConstantName" files="InMemoryCheckpointManager.java"/>
+  <suppress checks="ConstantName" files="InMemoryLeaseManager.java"/>
+  <suppress checks="ConstantName" files="MessagingFactory.java"/>
+  <suppress checks="ConstantName" files="PartitionManagerOptions.java"/>
+  <suppress checks="ConstantName" files="PartitionScanner.java"/>
   <suppress checks="ConstantName" files="RsaKey.java"/>
   <suppress checks="ConstantName" files="SymmetricKey.java"/>
-  <suppress checks="ConstantName" files="BatchErrorCodeStrings.java"/>
   <suppress checks="ConstantName" files="TaskFailureInformationCodes.java"/>
+
+  <!-- Public API already released with incorrect static variable naming -->
+  <suppress checks="StaticVariableName" files="EventHubClientImpl.java"/>
+
+  <!-- Public API already released with visibility modifier -->
+  <suppress checks="VisibilityModifier" files="BatchOptions.java"/>
+  <suppress checks="VisibilityModifier" files="EventHubClientImpl.java"/>
+
+  <!-- Public API already released with inner assignment-->
+  <suppress checks="InnerAssignment" files="MessageReceiver.java"/>
+
+  <!-- Public API already released with empty block-->
+  <suppress checks="EmptyBlock" files="MessageReceiver.java"/>
+  <suppress checks="EmptyBlock" files="InMemoryLeaseManager.java"/>
+  <suppress checks="EmptyBlock" files="ReactorDispatcher.java"/>
+  <suppress checks="EmptyBlock" files="AzureStorageCheckpointLeaseManager.java"/>
 </suppressions>

eng/code-quality-reports/src/main/resources/spotbugs/spotbugs-exclude.xml

@@ -47,6 +47,14 @@
         <Bug pattern="UUF_UNUSED_PUBLIC_OR_PROTECTED_FIELD"/>
     </Match>
 
+    <!-- Protected field already exists in the public API surface area even
+    though it should be final. -->
+    <Match>
+      <Class name="com.microsoft.azure.eventhubs.impl.EventHubClientImpl"/>
+      <Field name="USER_AGENT"/>
+      <Bug pattern="MS_SHOULD_BE_FINAL"/>
+    </Match>
+
     <!-- These KeyVault classes are publicly released APIs that intentionally return null rather
     than an empty array. -->
     <Match>

eventhubs/data-plane/azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/BaseLease.java

@@ -19,7 +19,7 @@
 public class BaseLease implements Comparable<BaseLease> {
     private final String partitionId;
     private String owner = "";
-	private transient boolean isOwned = false; // do not serialize
+    private transient boolean isOwned = false; // do not serialize
 
     /**
      * Do not use; added only for GSon deserializer
@@ -45,9 +45,9 @@ public BaseLease(String partitionId) {
      * @param isOwned True if the lease is owned, false if not.
      */
     public BaseLease(String partitionId, String owner, boolean isOwned) {
-    	this.partitionId = partitionId;
-    	this.owner = owner;
-    	this.isOwned = isOwned;
+        this.partitionId = partitionId;
+        this.owner = owner;
+        this.isOwned = isOwned;
     }
 
     /**
@@ -85,17 +85,17 @@ public void setOwner(String owner) {
      * @param newState true if the lease is owned, or false if it is not
      */
     public void setIsOwned(boolean newState) {
-    	this.isOwned = newState;
+        this.isOwned = newState;
     }
 
-	/**
-	 * Get the owned state of the lease.
-	 * 
-	 * @return true if the lease is owned, or false if it is not
-	 */
-	public boolean getIsOwned() {
-		return this.isOwned;
-	}
+    /**
+     * Get the owned state of the lease.
+     *
+     * @return true if the lease is owned, or false if it is not
+     */
+    public boolean getIsOwned() {
+        return this.isOwned;
+    }
 
     /**
      * Convenience function for comparing possibleOwner against this.owner
@@ -120,11 +120,11 @@ public String getPartitionId() {
         return this.partitionId;
     }
 
-	// Compares by partition id
-	@Override
-	public int compareTo(BaseLease other) {
-		return this.partitionId.compareTo(other.getPartitionId());
-	}
+    // Compares by partition id
+    @Override
+    public int compareTo(BaseLease other) {
+        return this.partitionId.compareTo(other.getPartitionId());
+    }
 
     String getStateDebug() {
         return "N/A";

eventhubs/data-plane/azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/Closable.java

@@ -11,10 +11,10 @@ class Closable {
 
     // null parent means top-level
     Closable(Closable parent) {
-    	this.syncClose = new Object();
-    	this.parent = parent;
-    	this.isClosing = false;
-    	this.isClosed = false;
+        this.syncClose = new Object();
+        this.parent = parent;
+        this.isClosing = false;
+        this.isClosed = false;
     }
 
     protected final boolean getIsClosed() {
@@ -33,29 +33,29 @@ protected final boolean getIsClosingOrClosed() {
     }
 
     protected final void setClosing() {
-    	synchronized (this.syncClose) {
-    		this.isClosing = true;
-    	}
+        synchronized (this.syncClose) {
+            this.isClosing = true;
+        }
     }
 
     protected final void setClosed() {
-    	synchronized (this.syncClose) {
-    		this.isClosing = false;
-    		this.isClosed = true;
-    	}
+        synchronized (this.syncClose) {
+            this.isClosing = false;
+            this.isClosed = true;
+        }
     }
 
     protected final void throwIfClosingOrClosed(String message) {
-    	if (getIsClosingOrClosed()) {
-    		throw new ClosingException(message);
-    	}
+        if (getIsClosingOrClosed()) {
+            throw new ClosingException(message);
+        }
     }
 
     class ClosingException extends RuntimeException {
-		private static final long serialVersionUID = 1138985585921317036L;
+        private static final long serialVersionUID = 1138985585921317036L;
 
-		ClosingException(String message) {
-    		super(message);
-    	}
+        ClosingException(String message) {
+            super(message);
+        }
     }
 }

eventhubs/data-plane/azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/CompleteLease.java

@@ -22,7 +22,7 @@ public class CompleteLease extends BaseLease {
      * Do not use; added only for GSon deserializer
      */
     protected CompleteLease() {
-    	super();
+        super();
     }
 
     /**
@@ -31,7 +31,7 @@ protected CompleteLease() {
      * @param partitionId Partition id for this lease.
      */
     public CompleteLease(String partitionId) {
-    	super(partitionId);
+        super(partitionId);
     }
 
     /**

eventhubs/data-plane/azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/EventProcessorHost.java

@@ -12,7 +12,14 @@
 import java.net.URISyntaxException;
 import java.security.InvalidKeyException;
 import java.util.UUID;
-import java.util.concurrent.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /***
@@ -500,8 +507,7 @@ public CompletableFuture<Void> unregisterEventProcessor() {
             // If we own the executor, stop it also.
             // Owned executor is also created in constructor.
             if (this.weOwnExecutor) {
-                this.unregistered = this.unregistered.thenRunAsync(() ->
-                {
+                this.unregistered = this.unregistered.thenRunAsync(() -> {
                     // IMPORTANT: run this last stage in the default threadpool!
                     // If a task running in a threadpool waits for that threadpool to terminate, it's going to wait a long time...
 
@@ -532,7 +538,7 @@ static class EventProcessorHostThreadPoolFactory implements ThreadFactory {
         private final String entityName;
         private final String consumerGroupName;
 
-        public EventProcessorHostThreadPoolFactory(
+        EventProcessorHostThreadPoolFactory(
                 String hostName,
                 String entityName,
                 String consumerGroupName) {

eventhubs/data-plane/azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/EventProcessorHostActionStrings.java

@@ -8,27 +8,27 @@
  * They describe what activity was taking place when the exception occurred.  
  */
 public final class EventProcessorHostActionStrings {
-    public final static String ACQUIRING_LEASE = "Acquiring Lease";
-    public final static String CHECKING_CHECKPOINT_STORE = "Checking Checpoint Store Existence";
-    public final static String CHECKING_LEASES = "Checking Leases";
-    public final static String CHECKING_LEASE_STORE = "Checking Lease Store Existence";
-    public final static String CLOSING_EVENT_PROCESSOR = "Closing Event Processor";
-    public final static String CREATING_CHECKPOINTS = "Creating Checkpoint Holders";
-    public final static String CREATING_CHECKPOINT_STORE = "Creating Checkpoint Store";
-    public final static String CREATING_EVENT_HUB_CLIENT = "Creating Event Hub Client";
-    public final static String CREATING_EVENT_PROCESSOR = "Creating Event Processor";
-    public final static String CREATING_LEASES = "Creating Leases";
-    public final static String CREATING_LEASE_STORE = "Creating Lease Store";
-    public final static String DELETING_LEASE = "Deleting Lease";
-    public final static String GETTING_CHECKPOINT = "Getting Checkpoint Details";
-    public final static String GETTING_LEASE = "Getting Lease Details";
-    public final static String INITIALIZING_STORES = "Initializing Stores";
-    public final static String OPENING_EVENT_PROCESSOR = "Opening Event Processor";
-    public final static String PARTITION_MANAGER_CLEANUP = "Partition Manager Cleanup";
-    public final static String PARTITION_MANAGER_MAIN_LOOP = "Partition Manager Main Loop";
-    public final static String RELEASING_LEASE = "Releasing Lease";
-    public final static String RENEWING_LEASE = "Renewing Lease";
-    public final static String STEALING_LEASE = "Stealing Lease";
-    public final static String UPDATING_CHECKPOINT = "Updating Checkpoint";
-    public final static String UPDATING_LEASE = "Updating Lease";
+    public static final String ACQUIRING_LEASE = "Acquiring Lease";
+    public static final String CHECKING_CHECKPOINT_STORE = "Checking Checpoint Store Existence";
+    public static final String CHECKING_LEASES = "Checking Leases";
+    public static final String CHECKING_LEASE_STORE = "Checking Lease Store Existence";
+    public static final String CLOSING_EVENT_PROCESSOR = "Closing Event Processor";
+    public static final String CREATING_CHECKPOINTS = "Creating Checkpoint Holders";
+    public static final String CREATING_CHECKPOINT_STORE = "Creating Checkpoint Store";
+    public static final String CREATING_EVENT_HUB_CLIENT = "Creating Event Hub Client";
+    public static final String CREATING_EVENT_PROCESSOR = "Creating Event Processor";
+    public static final String CREATING_LEASES = "Creating Leases";
+    public static final String CREATING_LEASE_STORE = "Creating Lease Store";
+    public static final String DELETING_LEASE = "Deleting Lease";
+    public static final String GETTING_CHECKPOINT = "Getting Checkpoint Details";
+    public static final String GETTING_LEASE = "Getting Lease Details";
+    public static final String INITIALIZING_STORES = "Initializing Stores";
+    public static final String OPENING_EVENT_PROCESSOR = "Opening Event Processor";
+    public static final String PARTITION_MANAGER_CLEANUP = "Partition Manager Cleanup";
+    public static final String PARTITION_MANAGER_MAIN_LOOP = "Partition Manager Main Loop";
+    public static final String RELEASING_LEASE = "Releasing Lease";
+    public static final String RENEWING_LEASE = "Renewing Lease";
+    public static final String STEALING_LEASE = "Stealing Lease";
+    public static final String UPDATING_CHECKPOINT = "Updating Checkpoint";
+    public static final String UPDATING_LEASE = "Updating Lease";
 }

eventhubs/data-plane/azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/HostContext.java

@@ -8,21 +8,21 @@
 import java.util.concurrent.ScheduledExecutorService;
 
 final class HostContext {
-    final private ScheduledExecutorService executor;
+    private final ScheduledExecutorService executor;
 
     // Ideally we wouldn't need the host, but there are certain things which can be dynamically changed
     // by the user via APIs on the host and which need to be exposed on the HostContext. Passing the
     // call through is easier and safer than trying to keep two copies in sync.
-    final private EventProcessorHost host;
-    final private String hostName;
+    private final EventProcessorHost host;
+    private final String hostName;
 
-    final private String eventHubPath;
-    final private String consumerGroupName;
-    final private String eventHubConnectionString;
-    final private RetryPolicy retryPolicy;
+    private final String eventHubPath;
+    private final String consumerGroupName;
+    private final String eventHubConnectionString;
+    private final RetryPolicy retryPolicy;
 
-    final private ILeaseManager leaseManager;
-    final private ICheckpointManager checkpointManager;
+    private final ILeaseManager leaseManager;
+    private final ICheckpointManager checkpointManager;
 
     // Cannot be final because it is not available at HostContext construction time.
     private EventProcessorOptions eventProcessorOptions = null;

eventhubs/data-plane/azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/ICheckpointManager.java

@@ -27,21 +27,21 @@ public interface ICheckpointManager {
      *
      * @return CompletableFuture {@literal ->} true if it exists, false if not
      */
-    public CompletableFuture<Boolean> checkpointStoreExists();
+    CompletableFuture<Boolean> checkpointStoreExists();
 
     /***
      * Create the checkpoint store if it doesn't exist. Do nothing if it does exist.
      *
      * @return CompletableFuture {@literal ->} null on success, completes exceptionally on error.
      */
-    public CompletableFuture<Void> createCheckpointStoreIfNotExists();
+    CompletableFuture<Void> createCheckpointStoreIfNotExists();
 
     /**
      * Deletes the checkpoint store.
      *
      * @return CompletableFuture {@literal ->} null on success, completes exceptionally on error.
      */
-    public CompletableFuture<Void> deleteCheckpointStore();
+    CompletableFuture<Void> deleteCheckpointStore();
 
     /***
      * Get the checkpoint data associated with the given partition. Could return null if no checkpoint has
@@ -51,11 +51,11 @@ public interface ICheckpointManager {
      *
      * @return CompletableFuture {@literal ->} checkpoint info, or null. Completes exceptionally on error.
      */
-    public CompletableFuture<Checkpoint> getCheckpoint(String partitionId);
+    CompletableFuture<Checkpoint> getCheckpoint(String partitionId);
 
     /***
-	 * Creates the checkpoint HOLDERs for the given partitions. Does nothing for any checkpoint HOLDERs
-	 * that already exist.
+     * Creates the checkpoint HOLDERs for the given partitions. Does nothing for any checkpoint HOLDERs
+     * that already exist.
      *
      * The semantics of this are complicated because it is possible to use the same store for both
      * leases and checkpoints (the Azure Storage implementation does so) and it is required to
@@ -67,7 +67,7 @@ public interface ICheckpointManager {
      * @param partitionIds  List of partitions to create checkpoint HOLDERs for.
      * @return CompletableFuture {@literal ->} null on success, completes exceptionally on error.
      */
-    public CompletableFuture<Void> createAllCheckpointsIfNotExists(List<String> partitionIds);
+    CompletableFuture<Void> createAllCheckpointsIfNotExists(List<String> partitionIds);
 
     /***
      * Update the checkpoint in the store with the offset/sequenceNumber in the provided checkpoint.
@@ -81,7 +81,7 @@ public interface ICheckpointManager {
      * @param checkpoint  offset/sequenceNumber and partition id to update the store with.
      * @return CompletableFuture {@literal ->} null on success. Completes exceptionally on error.
      */
-    public CompletableFuture<Void> updateCheckpoint(CompleteLease lease, Checkpoint checkpoint);
+    CompletableFuture<Void> updateCheckpoint(CompleteLease lease, Checkpoint checkpoint);
 
     /***
      * Delete the stored checkpoint data for the given partition. If there is no stored checkpoint for the
@@ -91,5 +91,5 @@ public interface ICheckpointManager {
      * @param partitionId  id of partition to delete checkpoint from store
      * @return CompletableFuture {@literal ->} null on success. Completes exceptionally on error.
      */
-    public CompletableFuture<Void> deleteCheckpoint(String partitionId);
+    CompletableFuture<Void> deleteCheckpoint(String partitionId);
 }