ZKLogSegmentMetadataStore implements LogSegmentMetadataStore, Watcher, Children2Callback { @Override public void updateLogSegment(Transaction<Object> txn, LogSegmentMetadata segment) { byte[] finalisedData = segment.getFinalisedData().getBytes(UTF_8); Op updateOp = Op.setData(segment.getZkPath(), finalisedData, -1); txn.addOp(DefaultZKOp.of(updateOp)); } ZKLogSegmentMetadataStore(DistributedLogConfiguration conf,
                                     ZooKeeperClient zkc,
                                     OrderedScheduler scheduler); @Override void storeMaxLogSegmentSequenceNumber(Transaction<Object> txn,
                                                 String path,
                                                 Versioned<Long> lssn,
                                                 Transaction.OpListener<Version> listener); @Override void storeMaxTxnId(Transaction<Object> txn,
                              String path,
                              Versioned<Long> transactionId,
                              Transaction.OpListener<Version> listener); @Override Transaction<Object> transaction(); @Override void createLogSegment(Transaction<Object> txn, LogSegmentMetadata segment); @Override void deleteLogSegment(Transaction<Object> txn, LogSegmentMetadata segment); @Override void updateLogSegment(Transaction<Object> txn, LogSegmentMetadata segment); @Override void process(WatchedEvent event); @Override Future<LogSegmentMetadata> getLogSegment(String logSegmentPath); @Override Future<List<String>> getLogSegmentNames(String logSegmentsPath); @Override @SuppressWarnings("unchecked") void processResult(int rc, String path, Object ctx, List<String> children, Stat stat); @Override void registerLogSegmentListener(String logSegmentsPath,
                                           LogSegmentNamesListener listener); @Override void unregisterLogSegmentListener(String logSegmentsPath,
                                             LogSegmentNamesListener listener); @Override void close();  }
@Test(timeout = 60000) public void testUpdateLogSegment() throws Exception { LogSegmentMetadata segment = createLogSegment(1L, 99L); Transaction<Object> createTxn = lsmStore.transaction(); lsmStore.createLogSegment(createTxn, segment); FutureUtils.result(createTxn.execute()); assertNotNull("LogSegment " + segment + " should be created", zkc.get().exists(segment.getZkPath(), false)); LogSegmentMetadata modifiedSegment = createLogSegment(1L, 999L); Transaction<Object> updateTxn = lsmStore.transaction(); lsmStore.updateLogSegment(updateTxn, modifiedSegment); FutureUtils.result(updateTxn.execute()); LogSegmentMetadata readSegment = FutureUtils.result(LogSegmentMetadata.read(zkc, segment.getZkPath(), true)); assertEquals("Last entry id should be changed from 99L to 999L", 999L, readSegment.getLastEntryId()); }