diff --git a/heron/statemgrs/src/java/org/apache/heron/statemgr/zookeeper/curator/CuratorStateManager.java b/heron/statemgrs/src/java/org/apache/heron/statemgr/zookeeper/curator/CuratorStateManager.java
index fe097e5..78129cc 100644
--- a/heron/statemgrs/src/java/org/apache/heron/statemgr/zookeeper/curator/CuratorStateManager.java
+++ b/heron/statemgrs/src/java/org/apache/heron/statemgr/zookeeper/curator/CuratorStateManager.java
@@ -63,6 +63,8 @@
 
 public class CuratorStateManager extends FileSystemStateManager {
   private static final Logger LOG = Logger.getLogger(CuratorStateManager.class.getName());
+  private static final int TUNNEL_SETUP_RETRY = 0;  // 0 means no retry
+  private static final int TUNNEL_SETUP_RETRY_SLEEP_SEC = 5;
 
   private CuratorFramework client;
   private String connectionString;
@@ -83,17 +85,29 @@
         NetworkUtils.TunnelConfig.build(config, NetworkUtils.HeronSystem.STATE_MANAGER);
 
     if (tunnelConfig.isTunnelNeeded()) {
-      Pair<String, List<Process>> tunneledResults = setupZkTunnel(tunnelConfig);
+      for (int setupCount = 0;; ++setupCount) {
+        Pair<String, List<Process>> tunneledResults = setupZkTunnel(tunnelConfig);
+        String newConnectionString = tunneledResults.first;
 
-      String newConnectionString = tunneledResults.first;
-      if (newConnectionString.isEmpty()) {
-        throw new IllegalArgumentException("Failed to connect to tunnel host '"
-            + tunnelConfig.getTunnelHost() + "'");
+        // If tunnel can't be setup correctly. Retry or bail.
+        if (!newConnectionString.isEmpty()) {
+          // Success, use the new connection string
+          connectionString = newConnectionString;
+          tunnelProcesses.addAll(tunneledResults.second);
+          break;
+        } else {
+          if (setupCount < TUNNEL_SETUP_RETRY) {
+            try {
+              TimeUnit.SECONDS.sleep(TUNNEL_SETUP_RETRY_SLEEP_SEC);
+            } catch (InterruptedException ex) {
+              Thread.currentThread().interrupt();
+            }
+          } else {
+            throw new IllegalArgumentException("Failed to connect to tunnel host '"
+                + tunnelConfig.getTunnelHost() + "'");
+          }
+        }
       }
-
-      // Use the new connection string
-      connectionString = newConnectionString;
-      tunnelProcesses.addAll(tunneledResults.second);
     }
 
     // Start it