LogSegmentCache { public Pair<Set<String>, Set<String>> diff(Set<String> segmentsReceived) { Set<String> segmentsAdded; Set<String> segmentsRemoved; synchronized (logSegments) { Set<String> segmentsCached = logSegments.keySet(); segmentsAdded = Sets.difference(segmentsReceived, segmentsCached).immutableCopy(); segmentsRemoved = Sets.difference(segmentsCached, segmentsReceived).immutableCopy(); } return Pair.of(segmentsAdded, segmentsRemoved); } LogSegmentCache(String streamName); List<LogSegmentMetadata> getLogSegments(Comparator<LogSegmentMetadata> comparator); void add(String name, LogSegmentMetadata metadata); LogSegmentMetadata get(String name); void update(Set<String> segmentsRemoved,
                       Map<String, LogSegmentMetadata> segmentsAdded); Pair<Set<String>, Set<String>> diff(Set<String> segmentsReceived); LogSegmentMetadata remove(String name);  }
@Test(timeout = 60000) public void testDiff() { LogSegmentCache cache = new LogSegmentCache("test-diff"); for (int i = 1; i <= 5; i++) { LogSegmentMetadata metadata = DLMTestUtil.completedLogSegment("/segment" + i, i, i, i * 100L, 100, i, 99L, 0L); String name = DLMTestUtil.completedLedgerZNodeNameWithLogSegmentSequenceNumber(i); cache.add(name, metadata); } LogSegmentMetadata inprogress = DLMTestUtil.inprogressLogSegment("/inprogress-6", 6, 600L, 6); String name = DLMTestUtil.inprogressZNodeName(6); cache.add(name, inprogress); Set<String> segmentRemoved = Sets.newHashSet(); for (int i = 1; i <= 2; i++) { segmentRemoved.add(DLMTestUtil.completedLedgerZNodeNameWithLogSegmentSequenceNumber(i)); } segmentRemoved.add((DLMTestUtil.inprogressZNodeName(6))); Set<String> segmentReceived = Sets.newHashSet(); Set<String> segmentAdded = Sets.newHashSet(); for (int i = 3; i <= 6; i++) { segmentReceived.add(DLMTestUtil.completedLedgerZNodeNameWithLogSegmentSequenceNumber(i)); if (i == 6) { segmentAdded.add(DLMTestUtil.completedLedgerZNodeNameWithLogSegmentSequenceNumber(i)); } } Pair<Set<String>, Set<String>> segmentChanges = cache.diff(segmentReceived); assertTrue("Should remove " + segmentRemoved + ", but removed " + segmentChanges.getRight(), Sets.difference(segmentRemoved, segmentChanges.getRight()).isEmpty()); assertTrue("Should add " + segmentAdded + ", but added " + segmentChanges.getLeft(), Sets.difference(segmentAdded, segmentChanges.getLeft()).isEmpty()); }