LogSegmentMetadataStoreUpdater implements MetadataUpdater { @Override public Future<LogSegmentMetadata> changeSequenceNumber(LogSegmentMetadata segment, long logSegmentSeqNo) { String newZkPath = segment.getZkPath() .replace(formatLogSegmentSequenceNumber(segment.getLogSegmentSequenceNumber()), formatLogSegmentSequenceNumber(logSegmentSeqNo)); final LogSegmentMetadata newSegment = segment.mutator() .setLogSegmentSequenceNumber(logSegmentSeqNo) .setZkPath(newZkPath) .build(); return addNewSegmentAndDeleteOldSegment(newSegment, segment); } protected  LogSegmentMetadataStoreUpdater(DistributedLogConfiguration conf,
                                             LogSegmentMetadataStore metadataStore); static MetadataUpdater createMetadataUpdater(DistributedLogConfiguration conf,
                                                        LogSegmentMetadataStore metadataStore); @Override Transaction<Object> transaction(); @Override Future<LogSegmentMetadata> updateLastRecord(LogSegmentMetadata segment,
                                                       LogRecordWithDLSN record); @Override Future<LogSegmentMetadata> changeSequenceNumber(LogSegmentMetadata segment,
                                                           long logSegmentSeqNo); @Override Future<LogSegmentMetadata> setLogSegmentActive(LogSegmentMetadata segment); @Override Future<LogSegmentMetadata> setLogSegmentTruncated(LogSegmentMetadata segment); @Override LogSegmentMetadata setLogSegmentTruncated(Transaction<Object> txn, LogSegmentMetadata segment); @Override Future<LogSegmentMetadata> setLogSegmentPartiallyTruncated(LogSegmentMetadata segment, DLSN minActiveDLSN); @Override LogSegmentMetadata setLogSegmentPartiallyTruncated(Transaction<Object> txn,
                                                              LogSegmentMetadata segment,
                                                              DLSN minActiveDLSN);  }
@Test(timeout = 60000) public void testChangeSequenceNumber() throws Exception { String ledgerPath = "/testChangeSequenceNumber"; zkc.get().create(ledgerPath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); Map<Long, LogSegmentMetadata> completedLogSegments = new HashMap<Long, LogSegmentMetadata>(); for (int i = 1; i <= 5; i++) { LogSegmentMetadata segment = DLMTestUtil.completedLogSegment(ledgerPath, i, (i - 1) * 100, i * 100 - 1, 100, i, 100, 0); completedLogSegments.put(((long)i), segment); LOG.info("Create completed segment {} : {}", segment.getZkPath(), segment); segment.write(zkc); } long inprogressSeqNo = 3; LogSegmentMetadata segment = DLMTestUtil.inprogressLogSegment(ledgerPath, inprogressSeqNo, 5 * 100, inprogressSeqNo); LOG.info("Create inprogress segment {} : {}", segment.getZkPath(), segment); segment.write(zkc); Map<Long, LogSegmentMetadata> segmentList = readLogSegments(ledgerPath); assertEquals(5, segmentList.size()); MetadataUpdater dryrunUpdater = new DryrunLogSegmentMetadataStoreUpdater(conf, metadataStore); FutureUtils.result(dryrunUpdater.changeSequenceNumber(segment, 6L)); segmentList = readLogSegments(ledgerPath); assertEquals(5, segmentList.size()); MetadataUpdater updater = LogSegmentMetadataStoreUpdater.createMetadataUpdater(conf, metadataStore); FutureUtils.result(updater.changeSequenceNumber(segment, 6L)); segmentList = readLogSegments(ledgerPath); assertEquals(6, segmentList.size()); for (int i = 1; i <= 5; i++) { LogSegmentMetadata s = segmentList.get((long)i); assertNotNull(s); assertEquals(completedLogSegments.get((long)i), s); } LogSegmentMetadata segmentChanged = segmentList.get(6L); assertNotNull(segmentChanged); assertEquals(6L, segmentChanged.getLogSegmentSequenceNumber()); assertTrue(segmentChanged.isInProgress()); assertEquals(5 * 100, segmentChanged.getFirstTxId()); assertEquals(3L, segmentChanged.getLedgerId()); }