Merge pull request #329 from metamx/segment-load-idempotent

ZkCoordinator: Make addSegment, addSegments idempotent

server/src/main/java/io/druid/server/coordination/ServerManager.java

@@ -116,7 +116,13 @@ public class ServerManager implements QuerySegmentWalker
     return segmentLoader.isSegmentLoaded(segment);
   }
 
-  public void loadSegment(final DataSegment segment) throws SegmentLoadingException
+  /**
+   * Load a single segment.
+   * @param segment segment to load
+   * @return true if the segment was newly loaded, false if it was already loaded
+   * @throws SegmentLoadingException if the segment cannot be loaded
+   */
+  public boolean loadSegment(final DataSegment segment) throws SegmentLoadingException
   {
     final Segment adapter;
     try {
@@ -150,8 +156,8 @@ public class ServerManager implements QuerySegmentWalker
           segment.getVersion()
       );
       if ((entry != null) && (entry.getChunk(segment.getShardSpec().getPartitionNum()) != null)) {
-        log.info("Told to load a adapter for a segment[%s] that already exists", segment.getIdentifier());
+        log.warn("Told to load a adapter for a segment[%s] that already exists", segment.getIdentifier());
-        throw new SegmentLoadingException("Segment already exists[%s]", segment.getIdentifier());
+        return false;
       }
 
       loadedIntervals.add(
@@ -165,6 +171,7 @@ public class ServerManager implements QuerySegmentWalker
       synchronized (dataSourceCounts) {
         dataSourceCounts.add(dataSource, 1L);
       }
+      return true;
     }
   }
 

server/src/main/java/io/druid/server/coordination/ZkCoordinator.java

@@ -230,14 +230,16 @@ public class ZkCoordinator implements DataSegmentChangeHandler
     try {
       log.info("Loading segment %s", segment.getIdentifier());
 
+      final boolean loaded;
       try {
-        serverManager.loadSegment(segment);
+        loaded = serverManager.loadSegment(segment);
       }
       catch (Exception e) {
         removeSegment(segment);
         throw new SegmentLoadingException(e, "Exception loading segment[%s]", segment.getIdentifier());
       }
 
+      if (loaded) {
         File segmentInfoCacheFile = new File(config.getInfoDir(), segment.getIdentifier());
         if (!segmentInfoCacheFile.exists()) {
           try {
@@ -257,6 +259,7 @@ public class ZkCoordinator implements DataSegmentChangeHandler
         catch (IOException e) {
           throw new SegmentLoadingException(e, "Failed to announce segment[%s]", segment.getIdentifier());
         }
+      }
 
     }
     catch (SegmentLoadingException e) {
@@ -275,8 +278,9 @@ public class ZkCoordinator implements DataSegmentChangeHandler
       for (DataSegment segment : segments) {
         log.info("Loading segment %s", segment.getIdentifier());
 
+        final boolean loaded;
         try {
-          serverManager.loadSegment(segment);
+          loaded = serverManager.loadSegment(segment);
         }
         catch (Exception e) {
           log.error(e, "Exception loading segment[%s]", segment.getIdentifier());
@@ -285,6 +289,7 @@ public class ZkCoordinator implements DataSegmentChangeHandler
           continue;
         }
 
+        if (loaded) {
           File segmentInfoCacheFile = new File(config.getInfoDir(), segment.getIdentifier());
           if (!segmentInfoCacheFile.exists()) {
             try {
@@ -300,6 +305,7 @@ public class ZkCoordinator implements DataSegmentChangeHandler
 
           validSegments.add(segment);
         }
+      }
 
       try {
         announcer.announceSegments(validSegments);