diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/ReadOnlyWagedRebalancer.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/ReadOnlyWagedRebalancer.java
index d1075d4..e94148e 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/ReadOnlyWagedRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/ReadOnlyWagedRebalancer.java
@@ -88,10 +88,5 @@
       _bestPossibleAssignment = bestPossibleAssignment;
       return true;
     }
-
-    @Override
-    // BucketDataAccessor will be reused, won't be closed here.
-    public void close() {
-    }
   }
 }
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java
index 521e3d7..c20b3f9 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java
@@ -41,12 +41,13 @@
 import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
 import org.apache.helix.zookeeper.api.client.HelixZkClient;
 import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.serializer.ByteArraySerializer;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordJacksonSerializer;
 import org.apache.helix.zookeeper.datamodel.ZNRecord;
 import org.apache.helix.zookeeper.impl.client.FederatedZkClient;
 import org.apache.helix.zookeeper.impl.factory.DedicatedZkClientFactory;
 import org.apache.helix.zookeeper.util.GZipCompressionUtil;
 import org.apache.helix.zookeeper.zkclient.DataUpdater;
-import org.apache.helix.zookeeper.zkclient.exception.ZkMarshallingError;
 import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
 import org.apache.helix.zookeeper.zkclient.serialize.ZkSerializer;
 import org.slf4j.Logger;
@@ -75,10 +76,11 @@
 
   private final int _bucketSize;
   private final long _versionTTLms;
-  private ZkSerializer _zkSerializer;
-  private RealmAwareZkClient _zkClient;
-  private ZkBaseDataAccessor<byte[]> _zkBaseDataAccessor;
-  private Map<String, ScheduledFuture> _gcTaskFutureMap = new HashMap<>();
+  private final ZkSerializer _zkSerializer;
+  private final RealmAwareZkClient _zkClient;
+  private final ZkBaseDataAccessor<byte[]> _zkBaseDataAccessor;
+  private final Map<String, ScheduledFuture> _gcTaskFutureMap = new HashMap<>();
+  private boolean _usesExternalZkClient = false;
 
   /**
    * Constructor that allows a custom bucket size.
@@ -87,25 +89,21 @@
    * @param versionTTLms in ms
    */
   public ZkBucketDataAccessor(String zkAddr, int bucketSize, long versionTTLms) {
-    _zkClient = createRealmAwareZkClient(zkAddr);
-    _zkClient.setZkSerializer(new ZkSerializer() {
-      @Override
-      public byte[] serialize(Object data) throws ZkMarshallingError {
-        if (data instanceof byte[]) {
-          return (byte[]) data;
-        }
-        throw new HelixException("ZkBucketDataAccesor only supports a byte array as an argument!");
-      }
+    this(createRealmAwareZkClient(zkAddr), bucketSize, versionTTLms, false);
+  }
 
-      @Override
-      public Object deserialize(byte[] data) throws ZkMarshallingError {
-        return data;
-      }
-    });
+  public ZkBucketDataAccessor(RealmAwareZkClient zkClient) {
+    this(zkClient, DEFAULT_BUCKET_SIZE, DEFAULT_VERSION_TTL, true);
+  }
+
+  private ZkBucketDataAccessor(RealmAwareZkClient zkClient, int bucketSize, long versionTTLms,
+      boolean usesExternalZkClient) {
+    _zkClient = zkClient;
     _zkBaseDataAccessor = new ZkBaseDataAccessor<>(_zkClient);
     _zkSerializer = new ZNRecordJacksonSerializer();
     _bucketSize = bucketSize;
     _versionTTLms = versionTTLms;
+    _usesExternalZkClient = usesExternalZkClient;
   }
 
   /**
@@ -135,6 +133,7 @@
       zkClient = DedicatedZkClientFactory.getInstance()
           .buildZkClient(new HelixZkClient.ZkConnectionConfig(zkAddr));
     }
+    zkClient.setZkSerializer(new ByteArraySerializer());
     return zkClient;
   }
 
@@ -258,7 +257,7 @@
 
   @Override
   public void disconnect() {
-    if (!_zkClient.isClosed()) {
+    if (!_usesExternalZkClient && _zkClient != null && !_zkClient.isClosed()) {
       _zkClient.close();
     }
   }
diff --git a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/BestPossibleExternalViewVerifier.java b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/BestPossibleExternalViewVerifier.java
index 3b133a1..1997bea 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/BestPossibleExternalViewVerifier.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/BestPossibleExternalViewVerifier.java
@@ -34,6 +34,7 @@
 import org.apache.helix.PropertyKey;
 import org.apache.helix.controller.common.PartitionStateMap;
 import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
+import org.apache.helix.controller.rebalancer.waged.ReadOnlyWagedRebalancer;
 import org.apache.helix.controller.rebalancer.waged.RebalanceAlgorithm;
 import org.apache.helix.controller.stages.AttributeName;
 import org.apache.helix.controller.stages.BestPossibleStateCalcStage;
@@ -91,6 +92,7 @@
     _resources = resources;
     _expectLiveInstances = expectLiveInstances;
     _dataProvider = new ResourceControllerDataProvider();
+    // _zkClient should be closed with BestPossibleExternalViewVerifier
   }
 
   /**
@@ -105,7 +107,7 @@
   public BestPossibleExternalViewVerifier(RealmAwareZkClient zkClient, String clusterName,
       Set<String> resources, Map<String, Map<String, String>> errStates,
       Set<String> expectLiveInstances) {
-    this(zkClient, clusterName, resources, errStates, expectLiveInstances, 0);
+    this(zkClient, clusterName, errStates, resources, expectLiveInstances, 0, true);
   }
 
   @Deprecated
@@ -114,11 +116,7 @@
       Set<String> expectLiveInstances, int waitTillVerify) {
     // usesExternalZkClient = true because ZkClient is given by the caller
     // at close(), we will not close this ZkClient because it might be being used elsewhere
-    super(zkClient, clusterName, true, waitTillVerify);
-    _errStates = errStates;
-    _resources = resources;
-    _expectLiveInstances = expectLiveInstances;
-    _dataProvider = new ResourceControllerDataProvider();
+    this(zkClient, clusterName, errStates, resources, expectLiveInstances, waitTillVerify, true);
   }
 
   private BestPossibleExternalViewVerifier(RealmAwareZkClient zkClient, String clusterName,
@@ -144,7 +142,6 @@
     private Set<String> _resources;
     private Set<String> _expectLiveInstances;
     private RealmAwareZkClient _zkClient;
-    private boolean _usesExternalZkClient = false; // false by default
 
     public Builder(String clusterName) {
       _clusterName = clusterName;
@@ -155,11 +152,12 @@
         throw new IllegalArgumentException("Cluster name is missing!");
       }
 
+      // _usesExternalZkClient == true
       if (_zkClient != null) {
-        return new BestPossibleExternalViewVerifier(_zkClient, _clusterName, _resources, _errStates,
-            _expectLiveInstances, _waitPeriodTillVerify);
+        return new BestPossibleExternalViewVerifier(_zkClient, _clusterName, _errStates, _resources,
+            _expectLiveInstances, _waitPeriodTillVerify, true);
       }
-
+      // _usesExternalZkClient == false
       if (_realmAwareZkConnectionConfig == null || _realmAwareZkClientConfig == null) {
         // For backward-compatibility
         return new BestPossibleExternalViewVerifier(_zkAddress, _clusterName, _resources,
@@ -170,7 +168,7 @@
       return new BestPossibleExternalViewVerifier(
           createZkClient(RealmAwareZkClient.RealmMode.SINGLE_REALM, _realmAwareZkConnectionConfig,
               _realmAwareZkClientConfig, _zkAddress), _clusterName, _errStates, _resources,
-          _expectLiveInstances, _waitPeriodTillVerify, _usesExternalZkClient);
+          _expectLiveInstances, _waitPeriodTillVerify, false);
     }
 
     public String getClusterName() {
@@ -210,7 +208,6 @@
 
     public Builder setZkClient(RealmAwareZkClient zkClient) {
       _zkClient = zkClient;
-      _usesExternalZkClient = true; // Set the flag since external ZkClient is used
       return this;
     }
   }
@@ -435,18 +432,15 @@
 
     RebalanceUtil.runStage(event, new CurrentStateComputationStage());
     // Note the readOnlyWagedRebalancer is just for one time usage
-    DryrunWagedRebalancer dryrunWagedRebalancer =
-        new DryrunWagedRebalancer(_zkClient.getServers(), cache.getClusterName(),
-            cache.getClusterConfig().getGlobalRebalancePreference());
-    event.addAttribute(AttributeName.STATEFUL_REBALANCER.name(), dryrunWagedRebalancer);
-    try {
+
+    try (ZkBucketDataAccessor zkBucketDataAccessor = new ZkBucketDataAccessor(_zkClient);
+        DryrunWagedRebalancer dryrunWagedRebalancer = new DryrunWagedRebalancer(zkBucketDataAccessor,
+            cache.getClusterName(), cache.getClusterConfig().getGlobalRebalancePreference())) {
+      event.addAttribute(AttributeName.STATEFUL_REBALANCER.name(), dryrunWagedRebalancer);
       RebalanceUtil.runStage(event, new BestPossibleStateCalcStage());
-    } finally {
-      dryrunWagedRebalancer.close();
     }
 
-    BestPossibleStateOutput output = event.getAttribute(AttributeName.BEST_POSSIBLE_STATE.name());
-    return output;
+    return event.getAttribute(AttributeName.BEST_POSSIBLE_STATE.name());
   }
 
   @Override
@@ -456,15 +450,17 @@
        + (_resources != null ? Arrays.toString(_resources.toArray()) : "") + "])";
   }
 
+  // TODO: to clean up, finalize is deprecated in Java 9
   @Override
   public void finalize() {
     close();
+    super.finalize();
   }
 
-  private class DryrunWagedRebalancer extends org.apache.helix.controller.rebalancer.waged.ReadOnlyWagedRebalancer {
-    public DryrunWagedRebalancer(String metadataStoreAddress, String clusterName,
+  private static class DryrunWagedRebalancer extends ReadOnlyWagedRebalancer implements AutoCloseable {
+    public DryrunWagedRebalancer(ZkBucketDataAccessor zkBucketDataAccessor, String clusterName,
         Map<ClusterConfig.GlobalRebalancePreferenceKey, Integer> preferences) {
-      super(new ZkBucketDataAccessor(metadataStoreAddress), clusterName, preferences);
+      super(zkBucketDataAccessor, clusterName, preferences);
     }
 
     @Override
diff --git a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/ZkHelixClusterVerifier.java b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/ZkHelixClusterVerifier.java
index 623a91e..11071d3 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/ZkHelixClusterVerifier.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/ClusterVerifiers/ZkHelixClusterVerifier.java
@@ -44,7 +44,7 @@
 import org.slf4j.LoggerFactory;
 
 public abstract class ZkHelixClusterVerifier
-    implements IZkChildListener, IZkDataListener, HelixClusterVerifier {
+    implements IZkChildListener, IZkDataListener, HelixClusterVerifier, AutoCloseable {
   private static Logger LOG = LoggerFactory.getLogger(ZkHelixClusterVerifier.class);
   protected static int DEFAULT_TIMEOUT = 300 * 1000;
   protected static int DEFAULT_PERIOD = 500;
@@ -229,6 +229,11 @@
     return verifyByPolling(DEFAULT_TIMEOUT, DEFAULT_PERIOD);
   }
 
+  /**
+   * Implement close() for {@link AutoCloseable} and {@link HelixClusterVerifier}.
+   * Non-external resources should be closed in this method to prevent resource leak.
+   */
+  @Override
   public void close() {
     if (_zkClient != null && !_usesExternalZkClient) {
       _zkClient.close();
diff --git a/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java b/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java
index a1cfb66..27921f6 100644
--- a/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java
+++ b/helix-rest/src/main/java/org/apache/helix/rest/server/ServerContext.java
@@ -285,7 +285,7 @@
     if (_zkBucketDataAccessor == null) {
       synchronized (this) {
         if (_zkBucketDataAccessor == null) {
-          _zkBucketDataAccessor = new ZkBucketDataAccessor(_zkAddr);
+          _zkBucketDataAccessor = new ZkBucketDataAccessor(getByteArrayRealmAwareZkClient());
         }
       }
     }