- Use double checked locking to ensure that we do not create Policies that we forget to destroy.

- Catch exceptions and close/destroy when necessary.

1 files changed, 15 insertions, 6 deletions

diff --git a/documentapi/src/main/java/com/yahoo/documentapi/messagebus/protocol/StoragePolicy.java b/documentapi/src/main/java/com/yahoo/documentapi/messagebus/protocol/StoragePolicy.java
index 918bd193d89..552f47ec5a6 100644
--- a/documentapi/src/main/java/com/yahoo/documentapi/messagebus/protocol/StoragePolicy.java
+++ b/documentapi/src/main/java/com/yahoo/documentapi/messagebus/protocol/StoragePolicy.java
@@ -349,15 +349,24 @@ public class StoragePolicy extends SlobrokPolicy {
         private final int maxOldClusterVersionBeforeSendingRandom; // Reset cluster version protection
 
         DistributorSelectionLogic(Parameters params, SlobrokPolicy policy) {
-            this.hostFetcher = params.createHostFetcher(policy, params.getRequiredUpPercentageToSendToKnownGoodNodes());
-            this.distribution = params.createDistribution(policy);
-            persistentFailureChecker = new InstabilityChecker(params.getAttemptRandomOnFailuresLimit());
-            maxOldClusterVersionBeforeSendingRandom = params.maxOldClusterStatesSeenBeforeThrowingCachedState();
+            try {
+                hostFetcher = params.createHostFetcher(policy, params.getRequiredUpPercentageToSendToKnownGoodNodes());
+                distribution = params.createDistribution(policy);
+                persistentFailureChecker = new InstabilityChecker(params.getAttemptRandomOnFailuresLimit());
+                maxOldClusterVersionBeforeSendingRandom = params.maxOldClusterStatesSeenBeforeThrowingCachedState();
+            } catch (Throwable e) {
+                destroy();
+                throw e;
+            }
         }
 
         public void destroy() {
-            hostFetcher.close();
-            distribution.close();
+            if (hostFetcher != null) {
+                hostFetcher.close();
+            }
+            if (distribution != null) {
+                distribution.close();
+            }
         }
 
         String getTargetSpec(RoutingContext context, BucketId bucketId) {