DistributedLogAdmin extends DistributedLogTool { public static void fixInprogressSegmentWithLowerSequenceNumber(final com.twitter.distributedlog.DistributedLogManagerFactory factory, final MetadataUpdater metadataUpdater, final String streamName, final boolean verbose, final boolean interactive) throws IOException { DistributedLogManager dlm = factory.createDistributedLogManagerWithSharedClients(streamName); try { List<LogSegmentMetadata> segments = dlm.getLogSegments(); if (verbose) { System.out.println("LogSegments for " + streamName + " : "); for (LogSegmentMetadata segment : segments) { System.out.println(segment.getLogSegmentSequenceNumber() + "\t: " + segment); } } LOG.info("Get log segments for {} : {}", streamName, segments); long maxCompletedLogSegmentSequenceNumber = -1L; LogSegmentMetadata inprogressSegment = null; for (LogSegmentMetadata segment : segments) { if (!segment.isInProgress()) { maxCompletedLogSegmentSequenceNumber = Math.max(maxCompletedLogSegmentSequenceNumber, segment.getLogSegmentSequenceNumber()); } else { if (null != inprogressSegment) { throw new DLIllegalStateException("Multiple inprogress segments found for stream " + streamName + " : " + segments); } inprogressSegment = segment; } } if (null == inprogressSegment || inprogressSegment.getLogSegmentSequenceNumber() > maxCompletedLogSegmentSequenceNumber) { return; } final long newLogSegmentSequenceNumber = maxCompletedLogSegmentSequenceNumber + 1; if (interactive && !IOUtils.confirmPrompt("Confirm to fix (Y/N), Ctrl+C to break : ")) { return; } final LogSegmentMetadata newSegment = FutureUtils.result(metadataUpdater.changeSequenceNumber(inprogressSegment, newLogSegmentSequenceNumber)); LOG.info("Fixed {} : {} -> {} ", new Object[] { streamName, inprogressSegment, newSegment }); if (verbose) { System.out.println("Fixed " + streamName + " : " + inprogressSegment.getZNodeName() + " -> " + newSegment.getZNodeName()); System.out.println("\t old: " + inprogressSegment); System.out.println("\t new: " + newSegment); System.out.println(); } } finally { dlm.close(); } } DistributedLogAdmin(); static void fixInprogressSegmentWithLowerSequenceNumber(final com.twitter.distributedlog.DistributedLogManagerFactory factory,
                                                                   final MetadataUpdater metadataUpdater,
                                                                   final String streamName,
                                                                   final boolean verbose,
                                                                   final boolean interactive); static void checkAndRepairDLNamespace(final URI uri,
                                                 final com.twitter.distributedlog.DistributedLogManagerFactory factory,
                                                 final MetadataUpdater metadataUpdater,
                                                 final ExecutorService executorService,
                                                 final BookKeeperClient bkc,
                                                 final String digestpw,
                                                 final boolean verbose,
                                                 final boolean interactive); static void checkAndRepairDLNamespace(final URI uri,
                                                 final com.twitter.distributedlog.DistributedLogManagerFactory factory,
                                                 final MetadataUpdater metadataUpdater,
                                                 final ExecutorService executorService,
                                                 final BookKeeperClient bkc,
                                                 final String digestpw,
                                                 final boolean verbose,
                                                 final boolean interactive,
                                                 final int concurrency);  }
@DistributedLogAnnotations.FlakyTest @Ignore @Test(timeout = 60000) @SuppressWarnings("deprecation") public void testChangeSequenceNumber() throws Exception { DistributedLogConfiguration confLocal = new DistributedLogConfiguration(); confLocal.addConfiguration(conf); confLocal.setLogSegmentSequenceNumberValidationEnabled(false); URI uri = createDLMURI("/change-sequence-number"); zooKeeperClient.get().create(uri.getPath(), new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); com.twitter.distributedlog.DistributedLogManagerFactory factory = new com.twitter.distributedlog.DistributedLogManagerFactory(confLocal, uri); String streamName = "change-sequence-number"; DistributedLogManager dlm = factory.createDistributedLogManagerWithSharedClients(streamName); DLMTestUtil.generateCompletedLogSegments(dlm, confLocal, 4, 10); DLMTestUtil.injectLogSegmentWithGivenLogSegmentSeqNo(dlm, confLocal, 5, 41, false, 10, true); dlm.close(); DistributedLogManager readDLM = factory.createDistributedLogManagerWithSharedClients(streamName); AsyncLogReader reader = readDLM.getAsyncLogReader(DLSN.InitialDLSN); long expectedTxId = 1L; for (int i = 0; i < 4 * 10; i++) { LogRecord record = Await.result(reader.readNext()); assertNotNull(record); DLMTestUtil.verifyLogRecord(record); assertEquals(expectedTxId, record.getTransactionId()); expectedTxId++; } dlm = factory.createDistributedLogManagerWithSharedClients(streamName); DLMTestUtil.injectLogSegmentWithGivenLogSegmentSeqNo(dlm, confLocal, 3L, 5 * 10 + 1, true, 10, false); TimeUnit.SECONDS.sleep(2); DLSN dlsn = readDLM.getLastDLSN(); assertTrue(dlsn.compareTo(new DLSN(5, Long.MIN_VALUE, Long.MIN_VALUE)) < 0); assertTrue(dlsn.compareTo(new DLSN(4, -1, Long.MIN_VALUE)) > 0); Future<LogRecordWithDLSN> readFuture = reader.readNext(); try { Await.result(readFuture, Duration.fromMilliseconds(1000)); fail("Should fail reading next when there is a corrupted log segment"); } catch (TimeoutException te) { } DistributedLogAdmin.fixInprogressSegmentWithLowerSequenceNumber(factory, new DryrunLogSegmentMetadataStoreUpdater(confLocal, getLogSegmentMetadataStore(factory)), streamName, false, false); TimeUnit.SECONDS.sleep(2); dlsn = readDLM.getLastDLSN(); assertTrue(dlsn.compareTo(new DLSN(5, Long.MIN_VALUE, Long.MIN_VALUE)) < 0); assertTrue(dlsn.compareTo(new DLSN(4, -1, Long.MIN_VALUE)) > 0); try { Await.result(readFuture, Duration.fromMilliseconds(1000)); fail("Should fail reading next when there is a corrupted log segment"); } catch (TimeoutException te) { } DistributedLogAdmin.fixInprogressSegmentWithLowerSequenceNumber(factory, LogSegmentMetadataStoreUpdater.createMetadataUpdater(confLocal, getLogSegmentMetadataStore(factory)), streamName, false, false); TimeUnit.SECONDS.sleep(2); expectedTxId = 51L; LogRecord record = Await.result(readFuture); assertNotNull(record); DLMTestUtil.verifyLogRecord(record); assertEquals(expectedTxId, record.getTransactionId()); expectedTxId++; for (int i = 1; i < 10; i++) { record = Await.result(reader.readNext()); assertNotNull(record); DLMTestUtil.verifyLogRecord(record); assertEquals(expectedTxId, record.getTransactionId()); expectedTxId++; } dlsn = readDLM.getLastDLSN(); LOG.info("LastDLSN after fix inprogress segment : {}", dlsn); assertTrue(dlsn.compareTo(new DLSN(7, Long.MIN_VALUE, Long.MIN_VALUE)) < 0); assertTrue(dlsn.compareTo(new DLSN(6, -1, Long.MIN_VALUE)) > 0); Utils.close(reader); readDLM.close(); dlm.close(); factory.close(); }