LogSegmentCache { public void update(Set<String> segmentsRemoved, Map<String, LogSegmentMetadata> segmentsAdded) { synchronized (logSegments) { for (Map.Entry<String, LogSegmentMetadata> entry : segmentsAdded.entrySet()) { add(entry.getKey(), entry.getValue()); } for (String segment : segmentsRemoved) { remove(segment); } } } 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 testUpdate() { LogSegmentCache cache = new LogSegmentCache("test-update"); 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(); Map<String, LogSegmentMetadata> segmentAdded = Maps.newHashMap(); for (int i = 3; i <= 6; i++) { segmentReceived.add(DLMTestUtil.completedLedgerZNodeNameWithLogSegmentSequenceNumber(i)); if (i == 6) { segmentAdded.put(DLMTestUtil.completedLedgerZNodeNameWithLogSegmentSequenceNumber(i), DLMTestUtil.completedLogSegment("/segment" + i, i, i, i * 100L, 100, i, 99L, 0L)); } } cache.update(segmentRemoved, segmentAdded); for (String segment : segmentRemoved) { assertNull("Segment " + segment + " should be removed.", cache.get(segment)); } for (String segment : segmentReceived) { assertNotNull("Segment " + segment + " should not be removed", cache.get(segment)); } for (Map.Entry<String, LogSegmentMetadata> entry : segmentAdded.entrySet()) { assertEquals("Segment " + entry.getKey() + " should be added.", entry.getValue(), entry.getValue()); } }