nodes will not try and download segments during startup

server/src/main/java/com/metamx/druid/coordination/ServerManager.java

@@ -104,6 +104,11 @@ public class ServerManager implements QuerySegmentWalker
     }
   }
 
+  public boolean isSegmentLoaded(final DataSegment segment) throws SegmentLoadingException
+  {
+    return segmentLoader.isSegmentLoaded(segment);
+  }
+
   public void loadSegment(final DataSegment segment) throws SegmentLoadingException
   {
     final Segment adapter;

server/src/main/java/com/metamx/druid/coordination/ZkCoordinator.java

@@ -215,9 +215,12 @@ public class ZkCoordinator implements DataSegmentChangeHandler
     for (File file : baseDir.listFiles()) {
       log.info("Loading segment cache file [%s]", file);
       try {
-        addSegment(jsonMapper.readValue(file, DataSegment.class));
+        DataSegment segment = jsonMapper.readValue(file, DataSegment.class);
+        if (serverManager.isSegmentLoaded(segment)) {
+          addSegment(segment);
+        }
       }
-      catch (IOException e) {
+      catch (Exception e) {
         log.error(e, "Exception occurred reading file [%s]", file);
         emitter.emit(
             new AlertEvent.Builder().build(

server/src/main/java/com/metamx/druid/loading/DelegatingSegmentLoader.java

@@ -43,6 +43,12 @@ public class DelegatingSegmentLoader implements SegmentLoader
     this.loaderTypes = loaderTypes;
   }
 
+  @Override
+  public boolean isSegmentLoaded(DataSegment segment) throws SegmentLoadingException
+  {
+    return getLoader(segment.getLoadSpec()).isSegmentLoaded(segment);
+  }
+
   @Override
   public Segment getSegment(DataSegment segment) throws SegmentLoadingException
   {

server/src/main/java/com/metamx/druid/loading/SegmentLoader.java

@@ -26,6 +26,7 @@ import com.metamx.druid.index.Segment;
  */
 public interface SegmentLoader
 {
+  public boolean isSegmentLoaded(DataSegment segment) throws SegmentLoadingException;
   public Segment getSegment(DataSegment loadSpec) throws SegmentLoadingException;
   public void cleanup(DataSegment loadSpec) throws SegmentLoadingException;
 }

server/src/main/java/com/metamx/druid/loading/SingleSegmentLoader.java

@@ -55,6 +55,21 @@ public class SingleSegmentLoader implements SegmentLoader
     this.config = config;
   }
 
+  @Override
+  public boolean isSegmentLoaded(final DataSegment segment)
+  {
+    File localStorageDir = new File(config.getCacheDirectory(), DataSegmentPusherUtil.getStorageDir(segment));
+    if (localStorageDir.exists()) {
+      return true;
+    }
+
+    final File legacyStorageDir = new File(
+        config.getCacheDirectory(),
+        DataSegmentPusherUtil.getLegacyStorageDir(segment)
+    );
+    return legacyStorageDir.exists();
+  }
+
   @Override
   public Segment getSegment(DataSegment segment) throws SegmentLoadingException
   {

server/src/test/java/com/metamx/druid/coordination/ServerManagerTest.java

@@ -84,6 +84,12 @@ public class ServerManagerTest
     serverManager = new ServerManager(
         new SegmentLoader()
         {
+          @Override
+          public boolean isSegmentLoaded(DataSegment segment) throws SegmentLoadingException
+          {
+            return false;
+          }
+
           @Override
           public Segment getSegment(final DataSegment segment)
           {

server/src/test/java/com/metamx/druid/loading/NoopSegmentLoader.java

@@ -29,6 +29,12 @@ import org.joda.time.Interval;
 */
 public class NoopSegmentLoader implements SegmentLoader
 {
+  @Override
+  public boolean isSegmentLoaded(DataSegment segment) throws SegmentLoadingException
+  {
+    return false;
+  }
+
   @Override
   public Segment getSegment(final DataSegment segment) throws SegmentLoadingException
   {