commit 5cb1475a91f41861943c7df689f59bf62a057cf7
author fjy <fangjin@metamarkets.com> Mon Aug 12 18:53:50 2013 -0700
committer fjy <fangjin@metamarkets.com> Mon Aug 12 18:53:50 2013 -0700
tree abf65b28d40258941a93acf31c52fcc153a20865
parent 5ff216838dc1b119dcbe4f31dcf38271e1d7fc4e

fix previous fix because multithread is not correct

client/src/main/java/com/metamx/druid/coordination/BatchDataSegmentAnnouncer.java

diff --git a/client/src/main/java/com/metamx/druid/coordination/BatchDataSegmentAnnouncer.java b/client/src/main/java/com/metamx/druid/coordination/BatchDataSegmentAnnouncer.java
index 10938b6..46aedce 100644
--- a/client/src/main/java/com/metamx/druid/coordination/BatchDataSegmentAnnouncer.java
+++ b/client/src/main/java/com/metamx/druid/coordination/BatchDataSegmentAnnouncer.java
@@ -37,6 +37,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  */
@@ -50,6 +51,7 @@
   private final String liveSegmentLocation;
 
   private final Object lock = new Object();
+  private final AtomicLong counter = new AtomicLong(0);
 
   private final Set<SegmentZNode> availableZNodes = new ConcurrentSkipListSet<SegmentZNode>();
   private final Map<DataSegment, SegmentZNode> segmentLookup = Maps.newConcurrentMap();
@@ -118,15 +120,17 @@
       return;
     }
 
-    segmentZNode.removeSegment(segment);
+    synchronized (lock) {
+      segmentZNode.removeSegment(segment);
 
-    log.info("Unannouncing segment[%s] at path[%s]", segment.getIdentifier(), segmentZNode.getPath());
-    if (segmentZNode.getCount() == 0) {
-      availableZNodes.remove(segmentZNode);
-      announcer.unannounce(segmentZNode.getPath());
-    } else {
-      announcer.update(segmentZNode.getPath(), segmentZNode.getBytes());
-      availableZNodes.add(segmentZNode);
+      log.info("Unannouncing segment[%s] at path[%s]", segment.getIdentifier(), segmentZNode.getPath());
+      if (segmentZNode.getCount() == 0) {
+        availableZNodes.remove(segmentZNode);
+        announcer.unannounce(segmentZNode.getPath());
+      } else {
+        announcer.update(segmentZNode.getPath(), segmentZNode.getBytes());
+        availableZNodes.add(segmentZNode);
+      }
     }
   }
 
@@ -176,7 +180,7 @@
 
   private String makeServedSegmentPath(String zNode)
   {
-    return ZKPaths.makePath(liveSegmentLocation, zNode);
+    return ZKPaths.makePath(liveSegmentLocation, String.format("%s%s", zNode, counter.getAndIncrement()));
   }
 
   private class SegmentZNode implements Comparable<SegmentZNode>