Index: src/main/java/org/apache/jackrabbit/oak/plugins/segment/file/CompactionGainEstimate.java
===================================================================
--- src/main/java/org/apache/jackrabbit/oak/plugins/segment/file/CompactionGainEstimate.java	(revision 1639339)
+++ src/main/java/org/apache/jackrabbit/oak/plugins/segment/file/CompactionGainEstimate.java	(working copy)
@@ -19,14 +19,17 @@
 import static org.apache.jackrabbit.oak.api.Type.BINARIES;
 
 import java.io.File;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.UUID;
 
 import org.apache.jackrabbit.oak.api.Blob;
 import org.apache.jackrabbit.oak.api.PropertyState;
+import org.apache.jackrabbit.oak.plugins.segment.RecordId;
 import org.apache.jackrabbit.oak.plugins.segment.SegmentBlob;
 import org.apache.jackrabbit.oak.plugins.segment.SegmentId;
+import org.apache.jackrabbit.oak.plugins.segment.SegmentNodeState;
 import org.apache.jackrabbit.oak.spi.state.ChildNodeEntry;
-import org.apache.jackrabbit.oak.spi.state.NodeState;
 
 import com.google.common.hash.BloomFilter;
 import com.google.common.hash.Funnel;
@@ -48,24 +51,28 @@
 
     private long reachableSize = 0;
 
-    CompactionGainEstimate(NodeState node, int estimatedBulkCount) {
+    CompactionGainEstimate(SegmentNodeState node, int estimatedBulkCount) {
         uuids = BloomFilter.create(UUID_FUNNEL, estimatedBulkCount);
-        collectBulkSegments(node);
+        collectBulkSegments(node, new HashSet<RecordId>());
     }
 
-    private void collectBulkSegments(NodeState node) {
-        for (PropertyState property : node.getProperties()) {
-            for (Blob blob : property.getValue(BINARIES)) {
-                for (SegmentId id : SegmentBlob.getBulkSegmentIds(blob)) {
-                    uuids.put(new UUID(
-                            id.getMostSignificantBits(),
-                            id.getLeastSignificantBits()));
+    private void collectBulkSegments(SegmentNodeState node,
+            Set<RecordId> visited) {
+        if (!visited.contains(node.getRecordId())) {
+            for (PropertyState property : node.getProperties()) {
+                for (Blob blob : property.getValue(BINARIES)) {
+                    for (SegmentId id : SegmentBlob.getBulkSegmentIds(blob)) {
+                        uuids.put(new UUID(id.getMostSignificantBits(), id
+                                .getLeastSignificantBits()));
+                    }
                 }
             }
+            for (ChildNodeEntry child : node.getChildNodeEntries()) {
+                collectBulkSegments((SegmentNodeState) child.getNodeState(),
+                        visited);
+            }
+            visited.add(node.getRecordId());
         }
-        for (ChildNodeEntry child : node.getChildNodeEntries()) {
-            collectBulkSegments(child.getNodeState());
-        }
     }
 
     /**