From ec53a0683a5d79944571e8b431dbcc9469bbfea7 Mon Sep 17 00:00:00 2001 From: Sebastian Schepens Date: Mon, 6 Feb 2017 17:20:24 -0300 Subject: [PATCH 1/2] persist and recover individual deleted messages --- conf/broker.conf | 2 + .../bookkeeper/mledger/ManagedCursor.java | 12 +- .../mledger/ManagedLedgerConfig.java | 18 + .../mledger/impl/ManagedCursorImpl.java | 173 +- .../mledger/impl/MetaStoreImplZookeeper.java | 255 ++- .../bookkeeper/mledger/impl/OpFindNewest.java | 13 +- .../bookkeeper/mledger/impl/PositionImpl.java | 9 + .../mledger/proto/MLDataFormats.java | 1819 ++++++++++++++++- .../src/main/proto/MLDataFormats.proto | 12 + .../impl/ManagedCursorContainerTest.java | 33 +- .../pulsar/broker/ServiceConfiguration.java | 13 + .../pulsar/broker/service/BrokerService.java | 1 + 12 files changed, 2029 insertions(+), 331 deletions(-) diff --git a/conf/broker.conf b/conf/broker.conf index 9a00ba1137076..95fc1637325bf 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -198,6 +198,8 @@ managedLedgerCursorMaxEntriesPerLedger=50000 # Max time before triggering a rollover on a cursor ledger managedLedgerCursorRolloverTimeInSeconds=14400 +managedLedgerMaxUnackedRangesToPersist=1000 + ### --- Load balancer --- ### diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java index cc538ff90d384..39357f3c9603a 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java @@ -15,9 +15,7 @@ */ package org.apache.bookkeeper.mledger; -import java.util.List; -import java.util.Set; - +import com.google.common.annotations.Beta; import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.FindEntryCallback; @@ -26,8 +24,10 @@ import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.SkipEntriesCallback; -import com.google.common.annotations.Beta; -import com.google.common.base.Predicate; + +import java.util.List; +import java.util.Set; +import java.util.function.Predicate; /** * A ManangedCursor is a persisted cursor inside a ManagedLedger. @@ -326,7 +326,7 @@ public void asyncSkipEntries(int numEntriesToSkip, IndividualDeletedEntries dele * opaque context */ public void asyncFindNewestMatching(FindPositionConstraint constraint, Predicate condition, - FindEntryCallback callback, Object ctx); + FindEntryCallback callback, Object ctx); /** * reset the cursor to specified position to enable replay of messages diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java index a2de073acada6..232d062ebe711 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java @@ -31,6 +31,7 @@ @Beta public class ManagedLedgerConfig { + private int maxUnackedRangesToPersist = 1000; private int maxEntriesPerLedger = 50000; private int maxSizePerLedgerMb = 100; private int minimumRolloverTimeMs = 0; @@ -347,4 +348,21 @@ public ManagedLedgerConfig setRetentionSizeInMB(long retentionSizeInMB) { public long getRetentionSizeInMB() { return retentionSizeInMB; } + + /** + * @return max unacked message ranges that will be persisted and recovered. + * + */ + public int getMaxUnackedRangesToPersist() { + return maxUnackedRangesToPersist; + } + + /** + * @param maxUnackedRangesToPersist + * max unacked message ranges that will be persisted and receverd. + */ + public ManagedLedgerConfig setMaxUnackedRangesToPersist(int maxUnackedRangesToPersist) { + this.maxUnackedRangesToPersist = maxUnackedRangesToPersist; + return this; + } } diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java index 799bf1251bf8a..00b676ef98ac8 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java @@ -20,6 +20,7 @@ import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun; import java.util.ArrayDeque; +import java.util.HashSet; import java.util.List; import java.util.Set; import java.util.concurrent.CountDownLatch; @@ -28,11 +29,18 @@ import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; -import org.apache.bookkeeper.client.AsyncCallback.AddCallback; +import com.google.common.base.Objects; +import com.google.common.collect.Collections2; +import com.google.common.collect.Lists; +import com.google.common.collect.Range; +import com.google.common.collect.RangeSet; +import com.google.common.collect.Sets; +import com.google.common.collect.TreeRangeSet; +import com.google.common.util.concurrent.RateLimiter; +import com.google.protobuf.InvalidProtocolBufferException; import org.apache.bookkeeper.client.AsyncCallback.CloseCallback; import org.apache.bookkeeper.client.AsyncCallback.DeleteCallback; import org.apache.bookkeeper.client.BKException; -import org.apache.bookkeeper.client.BKException.Code; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.LedgerEntry; import org.apache.bookkeeper.client.LedgerHandle; @@ -52,22 +60,16 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.PositionBound; import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback; import org.apache.bookkeeper.mledger.impl.MetaStore.Version; +import org.apache.bookkeeper.mledger.proto.MLDataFormats; import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo; import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo; import org.apache.bookkeeper.mledger.util.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.base.Objects; -import com.google.common.base.Predicate; -import com.google.common.collect.Collections2; -import com.google.common.collect.Lists; -import com.google.common.collect.Range; -import com.google.common.collect.RangeSet; -import com.google.common.collect.Sets; -import com.google.common.collect.TreeRangeSet; -import com.google.common.util.concurrent.RateLimiter; -import com.google.protobuf.InvalidProtocolBufferException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Predicate; +import java.util.stream.Collectors; public class ManagedCursorImpl implements ManagedCursor { @@ -124,7 +126,7 @@ public PendingMarkDeleteEntry(PositionImpl newPosition, MarkDeleteCallback callb } } - private final ArrayDeque pendingMarkDeleteOps = new ArrayDeque(); + private final ArrayDeque pendingMarkDeleteOps = new ArrayDeque<>(); private static final AtomicIntegerFieldUpdater PENDING_MARK_DELETED_SUBMITTED_COUNT_UPDATER = AtomicIntegerFieldUpdater.newUpdater(ManagedCursorImpl.class, "pendingMarkDeletedSubmittedCount"); private volatile int pendingMarkDeletedSubmittedCount = 0; @@ -136,16 +138,16 @@ enum State { Open, // Metadata ledger is ready SwitchingLedger, // The metadata ledger is being switched Closed // The managed cursor has been closed - }; + } private static final AtomicReferenceFieldUpdater STATE_UPDATER = AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, State.class, "state"); private volatile State state = null; public interface VoidCallback { - public void operationComplete(); + void operationComplete(); - public void operationFailed(ManagedLedgerException exception); + void operationFailed(ManagedLedgerException exception); } ManagedCursorImpl(BookKeeper bookkeeper, ManagedLedgerConfig config, ManagedLedgerImpl ledger, String cursorName) { @@ -186,6 +188,9 @@ public void operationComplete(ManagedCursorInfo info, Version version) { // closed and the last mark-delete position is stored in the ManagedCursorInfo itself.s PositionImpl recoveredPosition = new PositionImpl(info.getMarkDeleteLedgerId(), info.getMarkDeleteEntryId()); + if (info.getIndividualDeletedMessagesCount() > 0) { + recoverIndividualDeletedMessages(info.getIndividualDeletedMessagesList()); + } recoveredCursor(recoveredPosition); callback.operationComplete(); } else { @@ -255,12 +260,31 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac } PositionImpl position = new PositionImpl(positionInfo); + if (positionInfo.getIndividualDeletedMessagesCount() > 0) { + recoverIndividualDeletedMessages(positionInfo.getIndividualDeletedMessagesList()); + } recoveredCursor(position); callback.operationComplete(); }, null); }, null); } + private void recoverIndividualDeletedMessages(List individualDeletedMessagesList) { + lock.writeLock().lock(); + try { + individualDeletedMessages.clear(); + individualDeletedMessagesList + .forEach(messageRange -> individualDeletedMessages.add( + Range.openClosed( + new PositionImpl(messageRange.getLowerEndpoint()), + new PositionImpl(messageRange.getUpperEndpoint()) + ) + )); + } finally { + lock.writeLock().unlock(); + } + } + private void recoveredCursor(PositionImpl position) { // if the position was at a ledger that didn't exist (since it will be deleted if it was previously empty), // we need to move to the next existing ledger @@ -611,7 +635,7 @@ public void findEntryFailed(ManagedLedgerException exception, Object ctx) { @Override public void asyncFindNewestMatching(FindPositionConstraint constraint, Predicate condition, - FindEntryCallback callback, Object ctx) { + FindEntryCallback callback, Object ctx) { OpFindNewest op; PositionImpl startPosition = null; long max = 0; @@ -836,10 +860,8 @@ public Set asyncReplayEntries(final Set Set alreadyAcknowledgedPositions = Sets.newHashSet(); lock.readLock().lock(); try { - positions.stream().filter(position -> { - return individualDeletedMessages.contains((PositionImpl) position) - || ((PositionImpl) position).compareTo(markDeletePosition) < 0; - }).forEach(pos -> alreadyAcknowledgedPositions.add(pos)); + positions.stream().filter(position -> individualDeletedMessages.contains((PositionImpl) position) + || ((PositionImpl) position).compareTo(markDeletePosition) < 0).forEach(alreadyAcknowledgedPositions::add); } finally { lock.readLock().unlock(); } @@ -871,7 +893,7 @@ public synchronized void readEntryFailed(ManagedLedgerException mle, Object ctx) log.warn("[{}][{}] Error while replaying entries", ledger.getName(), name, mle); if (exception.compareAndSet(null, mle)) { // release the entries just once, any further read success will release the entry straight away - entries.forEach(e -> e.release()); + entries.forEach(Entry::release); } if (--pendingCallbacks == 0) { callback.readEntriesFailed(exception.get(), ctx); @@ -879,9 +901,9 @@ public synchronized void readEntryFailed(ManagedLedgerException mle, Object ctx) } }; - positions.stream().filter(position -> !alreadyAcknowledgedPositions.contains(position)).forEach(p -> { - ledger.asyncReadEntry((PositionImpl) p, cb, ctx); - }); + positions.stream() + .filter(position -> !alreadyAcknowledgedPositions.contains(position)) + .forEach(p -> ledger.asyncReadEntry((PositionImpl) p, cb, ctx)); return alreadyAcknowledgedPositions; } @@ -1631,9 +1653,14 @@ public void asyncClose(final AsyncCallbacks.CloseCallback callback, final Object // When closing we store the last mark-delete position in the z-node itself, so we won't need the cursor ledger, // hence we write it as -1. The cursor ledger is deleted once the z-node write is confirmed. - ManagedCursorInfo info = ManagedCursorInfo.newBuilder().setCursorsLedgerId(-1) - .setMarkDeleteLedgerId(markDeletePosition.getLedgerId()) - .setMarkDeleteEntryId(markDeletePosition.getEntryId()).build(); + ManagedCursorInfo.Builder builder = ManagedCursorInfo.newBuilder() + .setCursorsLedgerId(-1) + .setMarkDeleteLedgerId(markDeletePosition.getLedgerId()) + .setMarkDeleteEntryId(markDeletePosition.getEntryId()); + if (!individualDeletedMessages.isEmpty()) { + builder.addAllIndividualDeletedMessages(buildIndividualDeletedMessageRanges()); + } + ManagedCursorInfo info = builder.build(); if (log.isDebugEnabled()) { log.debug("[{}][{}] Closing cursor at md-position: {}", ledger.getName(), name, markDeletePosition); } @@ -1791,39 +1818,66 @@ public void deleteComplete(int rc, Object ctx) { }, null); } + private List buildIndividualDeletedMessageRanges() { + lock.readLock().lock(); + try { + MLDataFormats.NestedPositionInfo.Builder nestedPositionBuilder = MLDataFormats.NestedPositionInfo.newBuilder(); + MLDataFormats.MessageRange.Builder messageRangeBuilder = MLDataFormats.MessageRange.newBuilder(); + return individualDeletedMessages.asRanges().stream() + .limit(config.getMaxUnackedRangesToPersist()) + .map(positionRange -> { + PositionImpl p = positionRange.lowerEndpoint(); + nestedPositionBuilder.setLedgerId(p.getLedgerId()); + nestedPositionBuilder.setEntryId(p.getEntryId()); + messageRangeBuilder.setLowerEndpoint(nestedPositionBuilder.build()); + p = positionRange.upperEndpoint(); + nestedPositionBuilder.setLedgerId(p.getLedgerId()); + nestedPositionBuilder.setEntryId(p.getEntryId()); + messageRangeBuilder.setUpperEndpoint(nestedPositionBuilder.build()); + return messageRangeBuilder.build(); + }) + .collect(Collectors.toList()); + } finally { + lock.readLock().unlock(); + } + } + void persistPosition(final LedgerHandle lh, final PositionImpl position, final VoidCallback callback) { - PositionInfo pi = position.getPositionInfo(); + PositionInfo.Builder builder = PositionInfo.newBuilder() + .setLedgerId(position.getLedgerId()) + .setEntryId(position.getEntryId()); + if (!individualDeletedMessages.isEmpty()) { + builder.addAllIndividualDeletedMessages(buildIndividualDeletedMessageRanges()); + } + PositionInfo pi = builder.build(); if (log.isDebugEnabled()) { log.debug("[{}] Cursor {} Appending to ledger={} position={}", ledger.getName(), name, lh.getId(), position); } - lh.asyncAddEntry(pi.toByteArray(), new AddCallback() { - @Override - public void addComplete(int rc, LedgerHandle lh, long entryId, Object ctx) { - if (rc == BKException.Code.OK) { - if (log.isDebugEnabled()) { - log.debug("[{}] Updated cursor {} position {} in meta-ledger {}", ledger.getName(), name, - position, lh.getId()); - } + lh.asyncAddEntry(pi.toByteArray(), (rc, lh1, entryId, ctx) -> { + if (rc == BKException.Code.OK) { + if (log.isDebugEnabled()) { + log.debug("[{}] Updated cursor {} position {} in meta-ledger {}", ledger.getName(), name, + position, lh1.getId()); + } - if (shouldCloseLedger(lh)) { - if (log.isDebugEnabled()) { - log.debug("[{}] Need to create new metadata ledger for consumer {}", ledger.getName(), - name); - } - startCreatingNewMetadataLedger(); + if (shouldCloseLedger(lh1)) { + if (log.isDebugEnabled()) { + log.debug("[{}] Need to create new metadata ledger for consumer {}", ledger.getName(), + name); } - - callback.operationComplete(); - } else { - log.warn("[{}] Error updating cursor {} position {} in meta-ledger {}: {}", ledger.getName(), name, - position, lh.getId(), BKException.getMessage(rc)); - // If we've had a write error, the ledger will be automatically closed, we need to create a new one, - // in the meantime the mark-delete will be queued. - STATE_UPDATER.compareAndSet(ManagedCursorImpl.this, State.Open, State.NoLedger); - callback.operationFailed(new ManagedLedgerException(BKException.getMessage(rc))); + startCreatingNewMetadataLedger(); } + + callback.operationComplete(); + } else { + log.warn("[{}] Error updating cursor {} position {} in meta-ledger {}: {}", ledger.getName(), name, + position, lh1.getId(), BKException.getMessage(rc)); + // If we've had a write error, the ledger will be automatically closed, we need to create a new one, + // in the meantime the mark-delete will be queued. + STATE_UPDATER.compareAndSet(ManagedCursorImpl.this, State.Open, State.NoLedger); + callback.operationFailed(new ManagedLedgerException(BKException.getMessage(rc))); } }, null); } @@ -1846,9 +1900,14 @@ void switchToNewLedger(final LedgerHandle lh, final VoidCallback callback) { // Now we have an opened ledger that already has the acknowledged // position written into. At this point we can start using this new // ledger and delete the old one. - ManagedCursorInfo info = ManagedCursorInfo.newBuilder().setCursorsLedgerId(lh.getId()) + ManagedCursorInfo.Builder builder = ManagedCursorInfo.newBuilder() + .setCursorsLedgerId(lh.getId()) .setMarkDeleteLedgerId(markDeletePosition.getLedgerId()) - .setMarkDeleteEntryId(markDeletePosition.getEntryId()).build(); + .setMarkDeleteEntryId(markDeletePosition.getEntryId()); + if (!individualDeletedMessages.isEmpty()) { + builder.addAllIndividualDeletedMessages(buildIndividualDeletedMessageRanges()); + } + ManagedCursorInfo info = builder.build(); if (log.isDebugEnabled()) { log.debug("[{}] Switching cursor {} to ledger {}", ledger.getName(), name, lh.getId()); } @@ -1999,9 +2058,9 @@ void asyncDeleteCursorLedger() { */ private static boolean isBkErrorNotRecoverable(int rc) { switch (rc) { - case Code.NoSuchLedgerExistsException: - case Code.ReadException: - case Code.LedgerRecoveryException: + case BKException.Code.NoSuchLedgerExistsException: + case BKException.Code.ReadException: + case BKException.Code.LedgerRecoveryException: return true; default: diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImplZookeeper.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImplZookeeper.java index c9de9de13d43b..30b1a1b268883 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImplZookeeper.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImplZookeeper.java @@ -21,6 +21,7 @@ import java.util.ArrayList; import java.util.List; +import com.google.protobuf.InvalidProtocolBufferException; import org.apache.bookkeeper.mledger.ManagedLedgerException.BadVersionException; import org.apache.bookkeeper.mledger.ManagedLedgerException.MetaStoreException; import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo; @@ -100,41 +101,37 @@ private ManagedLedgerInfo updateMLInfoTimestamp(ManagedLedgerInfo info) { @Override public void getManagedLedgerInfo(final String ledgerName, final MetaStoreCallback callback) { // Try to get the content or create an empty node - zk.getData(prefix + ledgerName, false, (DataCallback) (rc, path, ctx, readData, stat) -> { - executor.submit(safeRun(() -> { - if (rc == KeeperException.Code.OK.intValue()) { - try { - ManagedLedgerInfo.Builder builder = ManagedLedgerInfo.newBuilder(); - TextFormat.merge(new String(readData, Encoding), builder); - ManagedLedgerInfo info = builder.build(); - info = updateMLInfoTimestamp(info); - callback.operationComplete(info, new ZKVersion(stat.getVersion())); - } catch (ParseException e) { - callback.operationFailed(new MetaStoreException(e)); - } - } else if (rc == KeeperException.Code.NONODE.intValue()) { - log.info("Creating '{}{}'", prefix, ledgerName); - - StringCallback createcb = new StringCallback() { - public void processResult(int rc, String path, Object ctx, String name) { - if (rc == KeeperException.Code.OK.intValue()) { - ManagedLedgerInfo info = ManagedLedgerInfo.getDefaultInstance(); - callback.operationComplete(info, new ZKVersion(0)); - } else { - callback.operationFailed( - new MetaStoreException(KeeperException.create(KeeperException.Code.get(rc)))); - } - } - }; - - ZkUtils.asyncCreateFullPathOptimistic(zk, prefix + ledgerName, new byte[0], Acl, - CreateMode.PERSISTENT, createcb, null); - } else { - callback.operationFailed( - new MetaStoreException(KeeperException.create(KeeperException.Code.get(rc)))); + zk.getData(prefix + ledgerName, false, (rc, path, ctx, readData, stat) -> executor.submit(safeRun(() -> { + if (rc == Code.OK.intValue()) { + try { + ManagedLedgerInfo.Builder builder = ManagedLedgerInfo.newBuilder(); + TextFormat.merge(new String(readData, Encoding), builder); + ManagedLedgerInfo info = builder.build(); + info = updateMLInfoTimestamp(info); + callback.operationComplete(info, new ZKVersion(stat.getVersion())); + } catch (ParseException e) { + callback.operationFailed(new MetaStoreException(e)); } - })); - }, null); + } else if (rc == Code.NONODE.intValue()) { + log.info("Creating '{}{}'", prefix, ledgerName); + + StringCallback createcb = (rc1, path1, ctx1, name) -> { + if (rc1 == Code.OK.intValue()) { + ManagedLedgerInfo info = ManagedLedgerInfo.getDefaultInstance(); + callback.operationComplete(info, new ZKVersion(0)); + } else { + callback.operationFailed( + new MetaStoreException(KeeperException.create(Code.get(rc1)))); + } + }; + + ZkUtils.asyncCreateFullPathOptimistic(zk, prefix + ledgerName, new byte[0], Acl, + CreateMode.PERSISTENT, createcb, null); + } else { + callback.operationFailed( + new MetaStoreException(KeeperException.create(Code.get(rc)))); + } + })), null); } @Override @@ -146,27 +143,23 @@ public void asyncUpdateLedgerIds(String ledgerName, ManagedLedgerInfo mlInfo, Ve log.debug("[{}] Updating metadata version={} with content={}", ledgerName, zkVersion.version, mlInfo); } - zk.setData(prefix + ledgerName, mlInfo.toString().getBytes(Encoding), zkVersion.version, new StatCallback() { - public void processResult(int rc, String path, Object zkCtx, Stat stat) { - executor.submit(safeRun(() -> { - if (log.isDebugEnabled()) { - log.debug("[{}] UpdateLedgersIdsCallback.processResult rc={} newVersion={}", ledgerName, - Code.get(rc), stat != null ? stat.getVersion() : "null"); - } - MetaStoreException status = null; - if (rc == KeeperException.Code.BADVERSION.intValue()) { - // Content has been modified on ZK since our last read - status = new BadVersionException(KeeperException.create(KeeperException.Code.get(rc))); - callback.operationFailed(status); - } else if (rc != KeeperException.Code.OK.intValue()) { - status = new MetaStoreException(KeeperException.create(KeeperException.Code.get(rc))); - callback.operationFailed(status); - } else { - callback.operationComplete(null, new ZKVersion(stat.getVersion())); - } - })); + zk.setData(prefix + ledgerName, mlInfo.toString().getBytes(Encoding), zkVersion.version, (rc, path, zkCtx, stat) -> executor.submit(safeRun(() -> { + if (log.isDebugEnabled()) { + log.debug("[{}] UpdateLedgersIdsCallback.processResult rc={} newVersion={}", ledgerName, + Code.get(rc), stat != null ? stat.getVersion() : "null"); + } + MetaStoreException status = null; + if (rc == Code.BADVERSION.intValue()) { + // Content has been modified on ZK since our last read + status = new BadVersionException(KeeperException.create(Code.get(rc))); + callback.operationFailed(status); + } else if (rc != Code.OK.intValue()) { + status = new MetaStoreException(KeeperException.create(Code.get(rc))); + callback.operationFailed(status); + } else { + callback.operationComplete(null, new ZKVersion(stat.getVersion())); } - }, null); + })), null); } @Override @@ -174,26 +167,22 @@ public void getCursors(final String ledgerName, final MetaStoreCallback children, Stat stat) { - executor.submit(safeRun(() -> { - if (log.isDebugEnabled()) { - log.debug("[{}] getConsumers complete rc={} children={}", ledgerName, Code.get(rc), children); - } - if (rc != KeeperException.Code.OK.intValue()) { - callback.operationFailed( - new MetaStoreException(KeeperException.create(KeeperException.Code.get(rc)))); - return; - } + zk.getChildren(prefix + ledgerName, false, (rc, path, ctx, children, stat) -> executor.submit(safeRun(() -> { + if (log.isDebugEnabled()) { + log.debug("[{}] getConsumers complete rc={} children={}", ledgerName, Code.get(rc), children); + } + if (rc != Code.OK.intValue()) { + callback.operationFailed( + new MetaStoreException(KeeperException.create(Code.get(rc)))); + return; + } - if (log.isDebugEnabled()) { - log.debug("[{}] Get childrend completed version={}", ledgerName, stat.getVersion()); - } - ZKVersion version = new ZKVersion(stat.getVersion()); - callback.operationComplete(children, version); - })); + if (log.isDebugEnabled()) { + log.debug("[{}] Get childrend completed version={}", ledgerName, stat.getVersion()); } - }, null); + ZKVersion version = new ZKVersion(stat.getVersion()); + callback.operationComplete(children, version); + })), null); } @Override @@ -204,22 +193,24 @@ public void asyncGetCursorInfo(String ledgerName, String consumerName, log.debug("Reading from {}", path); } - zk.getData(path, false, (DataCallback) (rc, path1, ctx, data, stat) -> { - executor.submit(safeRun(() -> { - if (rc != KeeperException.Code.OK.intValue()) { - callback.operationFailed( - new MetaStoreException(KeeperException.create(KeeperException.Code.get(rc)))); - } else { + zk.getData(path, false, (rc, path1, ctx, data, stat) -> executor.submit(safeRun(() -> { + if (rc != Code.OK.intValue()) { + callback.operationFailed( + new MetaStoreException(KeeperException.create(Code.get(rc)))); + } else { + try { + callback.operationComplete(ManagedCursorInfo.parseFrom(data), new ZKVersion(stat.getVersion())); + } catch (InvalidProtocolBufferException e) { try { ManagedCursorInfo.Builder info = ManagedCursorInfo.newBuilder(); TextFormat.merge(new String(data, Encoding), info); callback.operationComplete(info.build(), new ZKVersion(stat.getVersion())); - } catch (ParseException e) { + } catch (ParseException e1) { callback.operationFailed(new MetaStoreException(e)); } } - })); - }, null); + } + })), null); if (log.isDebugEnabled()) { log.debug("Reading from {} ok", path); @@ -233,46 +224,42 @@ public void asyncUpdateCursorInfo(final String ledgerName, final String cursorNa info.getCursorsLedgerId(), info.getMarkDeleteLedgerId(), info.getMarkDeleteEntryId()); String path = prefix + ledgerName + "/" + cursorName; - byte[] content = info.toString().getBytes(Encoding); + byte[] content = info.toByteArray(); if (version == null) { if (log.isDebugEnabled()) { log.debug("[{}] Creating consumer {} on meta-data store with {}", ledgerName, cursorName, info); } - zk.create(path, content, Acl, CreateMode.PERSISTENT, (rc, path1, ctx, name) -> { - executor.submit(safeRun(() -> { - if (rc != KeeperException.Code.OK.intValue()) { - log.warn("[{}] Error creating cosumer {} node on meta-data store with {}: ", ledgerName, - cursorName, info, KeeperException.Code.get(rc)); - callback.operationFailed( - new MetaStoreException(KeeperException.create(KeeperException.Code.get(rc)))); - } else { - if (log.isDebugEnabled()) { - log.debug("[{}] Created consumer {} on meta-data store with {}", ledgerName, cursorName, - info); - } - callback.operationComplete(null, new ZKVersion(0)); + zk.create(path, content, Acl, CreateMode.PERSISTENT, (rc, path1, ctx, name) -> executor.submit(safeRun(() -> { + if (rc != Code.OK.intValue()) { + log.warn("[{}] Error creating cosumer {} node on meta-data store with {}: ", ledgerName, + cursorName, info, Code.get(rc)); + callback.operationFailed( + new MetaStoreException(KeeperException.create(Code.get(rc)))); + } else { + if (log.isDebugEnabled()) { + log.debug("[{}] Created consumer {} on meta-data store with {}", ledgerName, cursorName, + info); } - })); - }, null); + callback.operationComplete(null, new ZKVersion(0)); + } + })), null); } else { ZKVersion zkVersion = (ZKVersion) version; if (log.isDebugEnabled()) { log.debug("[{}] Updating consumer {} on meta-data store with {}", ledgerName, cursorName, info); } - zk.setData(path, content, zkVersion.version, (rc, path1, ctx, stat) -> { - executor.submit(safeRun(() -> { - if (rc == KeeperException.Code.BADVERSION.intValue()) { - callback.operationFailed( - new BadVersionException(KeeperException.create(KeeperException.Code.get(rc)))); - } else if (rc != KeeperException.Code.OK.intValue()) { - callback.operationFailed( - new MetaStoreException(KeeperException.create(KeeperException.Code.get(rc)))); - } else { - callback.operationComplete(null, new ZKVersion(stat.getVersion())); - } - })); - }, null); + zk.setData(path, content, zkVersion.version, (rc, path1, ctx, stat) -> executor.submit(safeRun(() -> { + if (rc == Code.BADVERSION.intValue()) { + callback.operationFailed( + new BadVersionException(KeeperException.create(Code.get(rc)))); + } else if (rc != Code.OK.intValue()) { + callback.operationFailed( + new MetaStoreException(KeeperException.create(Code.get(rc)))); + } else { + callback.operationComplete(null, new ZKVersion(stat.getVersion())); + } + })), null); } } @@ -280,37 +267,33 @@ public void asyncUpdateCursorInfo(final String ledgerName, final String cursorNa public void asyncRemoveCursor(final String ledgerName, final String consumerName, final MetaStoreCallback callback) { log.info("[{}] Remove consumer={}", ledgerName, consumerName); - zk.delete(prefix + ledgerName + "/" + consumerName, -1, (rc, path, ctx) -> { - executor.submit(safeRun(() -> { - if (log.isDebugEnabled()) { - log.debug("[{}] [{}] zk delete done. rc={}", ledgerName, consumerName, Code.get(rc)); - } - if (rc == KeeperException.Code.OK.intValue()) { - callback.operationComplete(null, null); - } else { - callback.operationFailed( - new MetaStoreException(KeeperException.create(KeeperException.Code.get(rc)))); - } - })); - }, null); + zk.delete(prefix + ledgerName + "/" + consumerName, -1, (rc, path, ctx) -> executor.submit(safeRun(() -> { + if (log.isDebugEnabled()) { + log.debug("[{}] [{}] zk delete done. rc={}", ledgerName, consumerName, Code.get(rc)); + } + if (rc == Code.OK.intValue()) { + callback.operationComplete(null, null); + } else { + callback.operationFailed( + new MetaStoreException(KeeperException.create(Code.get(rc)))); + } + })), null); } @Override public void removeManagedLedger(String ledgerName, MetaStoreCallback callback) { log.info("[{}] Remove ManagedLedger", ledgerName); - zk.delete(prefix + ledgerName, -1, (rc, path, ctx) -> { - executor.submit(safeRun(() -> { - if (log.isDebugEnabled()) { - log.debug("[{}] zk delete done. rc={}", ledgerName, Code.get(rc)); - } - if (rc == KeeperException.Code.OK.intValue()) { - callback.operationComplete(null, null); - } else { - callback.operationFailed( - new MetaStoreException(KeeperException.create(KeeperException.Code.get(rc)))); - } - })); - }, null); + zk.delete(prefix + ledgerName, -1, (rc, path, ctx) -> executor.submit(safeRun(() -> { + if (log.isDebugEnabled()) { + log.debug("[{}] zk delete done. rc={}", ledgerName, Code.get(rc)); + } + if (rc == Code.OK.intValue()) { + callback.operationComplete(null, null); + } else { + callback.operationFailed( + new MetaStoreException(KeeperException.create(Code.get(rc)))); + } + })), null); } @Override diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java index 2a63241a6179c..9381aaed3044d 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java @@ -15,13 +15,14 @@ */ package org.apache.bookkeeper.mledger.impl; -import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.FindEntryCallback; +import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedLedgerException; import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.PositionBound; -import com.google.common.base.Predicate; + +import java.util.function.Predicate; /** */ @@ -43,7 +44,7 @@ enum State { State state; public OpFindNewest(ManagedCursorImpl cursor, PositionImpl startPosition, Predicate condition, - long numberOfEntries, FindEntryCallback callback, Object ctx) { + long numberOfEntries, FindEntryCallback callback, Object ctx) { this.cursor = cursor; this.startPosition = startPosition; this.callback = callback; @@ -61,7 +62,7 @@ public OpFindNewest(ManagedCursorImpl cursor, PositionImpl startPosition, Predic public void readEntryComplete(Entry entry, Object ctx) { switch (state) { case checkFirst: - if (!condition.apply(entry)) { + if (!condition.test(entry)) { callback.findEntryComplete(null, OpFindNewest.this.ctx); return; } else { @@ -74,7 +75,7 @@ public void readEntryComplete(Entry entry, Object ctx) { } break; case checkLast: - if (condition.apply(entry)) { + if (condition.test(entry)) { callback.findEntryComplete(entry.getPosition(), OpFindNewest.this.ctx); return; } else { @@ -85,7 +86,7 @@ public void readEntryComplete(Entry entry, Object ctx) { } break; case searching: - if (condition.apply(entry)) { + if (condition.test(entry)) { // mid - last lastMatchedPosition = entry.getPosition(); min = mid(); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImpl.java index b58bbe594de9b..f08d13dcd0de0 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImpl.java @@ -17,8 +17,11 @@ import static com.google.common.base.Preconditions.checkNotNull; +import com.google.common.collect.RangeSet; import org.apache.bookkeeper.mledger.Position; +import org.apache.bookkeeper.mledger.proto.MLDataFormats; import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo; +import org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo; import com.google.common.base.Objects; import com.google.common.collect.ComparisonChain; @@ -39,6 +42,12 @@ public PositionImpl(PositionInfo pi) { this.recyclerHandle = null; } + public PositionImpl(NestedPositionInfo npi) { + this.ledgerId = npi.getLedgerId(); + this.entryId = npi.getEntryId(); + this.recyclerHandle = null; + } + public PositionImpl(long ledgerId, long entryId) { this.ledgerId = ledgerId; this.entryId = entryId; diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/proto/MLDataFormats.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/proto/MLDataFormats.java index 26ab6fd11fc18..f19ca650211e7 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/proto/MLDataFormats.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/proto/MLDataFormats.java @@ -1098,6 +1098,16 @@ public interface PositionInfoOrBuilder // required int64 entryId = 2; boolean hasEntryId(); long getEntryId(); + + // repeated .MessageRange individualDeletedMessages = 3; + java.util.List + getIndividualDeletedMessagesList(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange getIndividualDeletedMessages(int index); + int getIndividualDeletedMessagesCount(); + java.util.List + getIndividualDeletedMessagesOrBuilderList(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder getIndividualDeletedMessagesOrBuilder( + int index); } public static final class PositionInfo extends com.google.protobuf.GeneratedMessage @@ -1148,9 +1158,31 @@ public long getEntryId() { return entryId_; } + // repeated .MessageRange individualDeletedMessages = 3; + public static final int INDIVIDUALDELETEDMESSAGES_FIELD_NUMBER = 3; + private java.util.List individualDeletedMessages_; + public java.util.List getIndividualDeletedMessagesList() { + return individualDeletedMessages_; + } + public java.util.List + getIndividualDeletedMessagesOrBuilderList() { + return individualDeletedMessages_; + } + public int getIndividualDeletedMessagesCount() { + return individualDeletedMessages_.size(); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange getIndividualDeletedMessages(int index) { + return individualDeletedMessages_.get(index); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder getIndividualDeletedMessagesOrBuilder( + int index) { + return individualDeletedMessages_.get(index); + } + private void initFields() { ledgerId_ = 0L; entryId_ = 0L; + individualDeletedMessages_ = java.util.Collections.emptyList(); } private byte memoizedIsInitialized = -1; public final boolean isInitialized() { @@ -1165,6 +1197,12 @@ public final boolean isInitialized() { memoizedIsInitialized = 0; return false; } + for (int i = 0; i < getIndividualDeletedMessagesCount(); i++) { + if (!getIndividualDeletedMessages(i).isInitialized()) { + memoizedIsInitialized = 0; + return false; + } + } memoizedIsInitialized = 1; return true; } @@ -1178,6 +1216,9 @@ public void writeTo(com.google.protobuf.CodedOutputStream output) if (((bitField0_ & 0x00000002) == 0x00000002)) { output.writeInt64(2, entryId_); } + for (int i = 0; i < individualDeletedMessages_.size(); i++) { + output.writeMessage(3, individualDeletedMessages_.get(i)); + } getUnknownFields().writeTo(output); } @@ -1195,6 +1236,10 @@ public int getSerializedSize() { size += com.google.protobuf.CodedOutputStream .computeInt64Size(2, entryId_); } + for (int i = 0; i < individualDeletedMessages_.size(); i++) { + size += com.google.protobuf.CodedOutputStream + .computeMessageSize(3, individualDeletedMessages_.get(i)); + } size += getUnknownFields().getSerializedSize(); memoizedSerializedSize = size; return size; @@ -1311,6 +1356,7 @@ private Builder(BuilderParent parent) { } private void maybeForceBuilderInitialization() { if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) { + getIndividualDeletedMessagesFieldBuilder(); } } private static Builder create() { @@ -1323,6 +1369,12 @@ public Builder clear() { bitField0_ = (bitField0_ & ~0x00000001); entryId_ = 0L; bitField0_ = (bitField0_ & ~0x00000002); + if (individualDeletedMessagesBuilder_ == null) { + individualDeletedMessages_ = java.util.Collections.emptyList(); + bitField0_ = (bitField0_ & ~0x00000004); + } else { + individualDeletedMessagesBuilder_.clear(); + } return this; } @@ -1369,6 +1421,15 @@ public org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo buildParti to_bitField0_ |= 0x00000002; } result.entryId_ = entryId_; + if (individualDeletedMessagesBuilder_ == null) { + if (((bitField0_ & 0x00000004) == 0x00000004)) { + individualDeletedMessages_ = java.util.Collections.unmodifiableList(individualDeletedMessages_); + bitField0_ = (bitField0_ & ~0x00000004); + } + result.individualDeletedMessages_ = individualDeletedMessages_; + } else { + result.individualDeletedMessages_ = individualDeletedMessagesBuilder_.build(); + } result.bitField0_ = to_bitField0_; onBuilt(); return result; @@ -1391,6 +1452,32 @@ public Builder mergeFrom(org.apache.bookkeeper.mledger.proto.MLDataFormats.Posit if (other.hasEntryId()) { setEntryId(other.getEntryId()); } + if (individualDeletedMessagesBuilder_ == null) { + if (!other.individualDeletedMessages_.isEmpty()) { + if (individualDeletedMessages_.isEmpty()) { + individualDeletedMessages_ = other.individualDeletedMessages_; + bitField0_ = (bitField0_ & ~0x00000004); + } else { + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.addAll(other.individualDeletedMessages_); + } + onChanged(); + } + } else { + if (!other.individualDeletedMessages_.isEmpty()) { + if (individualDeletedMessagesBuilder_.isEmpty()) { + individualDeletedMessagesBuilder_.dispose(); + individualDeletedMessagesBuilder_ = null; + individualDeletedMessages_ = other.individualDeletedMessages_; + bitField0_ = (bitField0_ & ~0x00000004); + individualDeletedMessagesBuilder_ = + com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders ? + getIndividualDeletedMessagesFieldBuilder() : null; + } else { + individualDeletedMessagesBuilder_.addAllMessages(other.individualDeletedMessages_); + } + } + } this.mergeUnknownFields(other.getUnknownFields()); return this; } @@ -1404,6 +1491,12 @@ public final boolean isInitialized() { return false; } + for (int i = 0; i < getIndividualDeletedMessagesCount(); i++) { + if (!getIndividualDeletedMessages(i).isInitialized()) { + + return false; + } + } return true; } @@ -1440,6 +1533,12 @@ public Builder mergeFrom( entryId_ = input.readInt64(); break; } + case 26: { + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder subBuilder = org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.newBuilder(); + input.readMessage(subBuilder, extensionRegistry); + addIndividualDeletedMessages(subBuilder.buildPartial()); + break; + } } } } @@ -1488,6 +1587,192 @@ public Builder clearEntryId() { return this; } + // repeated .MessageRange individualDeletedMessages = 3; + private java.util.List individualDeletedMessages_ = + java.util.Collections.emptyList(); + private void ensureIndividualDeletedMessagesIsMutable() { + if (!((bitField0_ & 0x00000004) == 0x00000004)) { + individualDeletedMessages_ = new java.util.ArrayList(individualDeletedMessages_); + bitField0_ |= 0x00000004; + } + } + + private com.google.protobuf.RepeatedFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder> individualDeletedMessagesBuilder_; + + public java.util.List getIndividualDeletedMessagesList() { + if (individualDeletedMessagesBuilder_ == null) { + return java.util.Collections.unmodifiableList(individualDeletedMessages_); + } else { + return individualDeletedMessagesBuilder_.getMessageList(); + } + } + public int getIndividualDeletedMessagesCount() { + if (individualDeletedMessagesBuilder_ == null) { + return individualDeletedMessages_.size(); + } else { + return individualDeletedMessagesBuilder_.getCount(); + } + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange getIndividualDeletedMessages(int index) { + if (individualDeletedMessagesBuilder_ == null) { + return individualDeletedMessages_.get(index); + } else { + return individualDeletedMessagesBuilder_.getMessage(index); + } + } + public Builder setIndividualDeletedMessages( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange value) { + if (individualDeletedMessagesBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.set(index, value); + onChanged(); + } else { + individualDeletedMessagesBuilder_.setMessage(index, value); + } + return this; + } + public Builder setIndividualDeletedMessages( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder builderForValue) { + if (individualDeletedMessagesBuilder_ == null) { + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.set(index, builderForValue.build()); + onChanged(); + } else { + individualDeletedMessagesBuilder_.setMessage(index, builderForValue.build()); + } + return this; + } + public Builder addIndividualDeletedMessages(org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange value) { + if (individualDeletedMessagesBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.add(value); + onChanged(); + } else { + individualDeletedMessagesBuilder_.addMessage(value); + } + return this; + } + public Builder addIndividualDeletedMessages( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange value) { + if (individualDeletedMessagesBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.add(index, value); + onChanged(); + } else { + individualDeletedMessagesBuilder_.addMessage(index, value); + } + return this; + } + public Builder addIndividualDeletedMessages( + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder builderForValue) { + if (individualDeletedMessagesBuilder_ == null) { + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.add(builderForValue.build()); + onChanged(); + } else { + individualDeletedMessagesBuilder_.addMessage(builderForValue.build()); + } + return this; + } + public Builder addIndividualDeletedMessages( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder builderForValue) { + if (individualDeletedMessagesBuilder_ == null) { + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.add(index, builderForValue.build()); + onChanged(); + } else { + individualDeletedMessagesBuilder_.addMessage(index, builderForValue.build()); + } + return this; + } + public Builder addAllIndividualDeletedMessages( + java.lang.Iterable values) { + if (individualDeletedMessagesBuilder_ == null) { + ensureIndividualDeletedMessagesIsMutable(); + super.addAll(values, individualDeletedMessages_); + onChanged(); + } else { + individualDeletedMessagesBuilder_.addAllMessages(values); + } + return this; + } + public Builder clearIndividualDeletedMessages() { + if (individualDeletedMessagesBuilder_ == null) { + individualDeletedMessages_ = java.util.Collections.emptyList(); + bitField0_ = (bitField0_ & ~0x00000004); + onChanged(); + } else { + individualDeletedMessagesBuilder_.clear(); + } + return this; + } + public Builder removeIndividualDeletedMessages(int index) { + if (individualDeletedMessagesBuilder_ == null) { + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.remove(index); + onChanged(); + } else { + individualDeletedMessagesBuilder_.remove(index); + } + return this; + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder getIndividualDeletedMessagesBuilder( + int index) { + return getIndividualDeletedMessagesFieldBuilder().getBuilder(index); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder getIndividualDeletedMessagesOrBuilder( + int index) { + if (individualDeletedMessagesBuilder_ == null) { + return individualDeletedMessages_.get(index); } else { + return individualDeletedMessagesBuilder_.getMessageOrBuilder(index); + } + } + public java.util.List + getIndividualDeletedMessagesOrBuilderList() { + if (individualDeletedMessagesBuilder_ != null) { + return individualDeletedMessagesBuilder_.getMessageOrBuilderList(); + } else { + return java.util.Collections.unmodifiableList(individualDeletedMessages_); + } + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder addIndividualDeletedMessagesBuilder() { + return getIndividualDeletedMessagesFieldBuilder().addBuilder( + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.getDefaultInstance()); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder addIndividualDeletedMessagesBuilder( + int index) { + return getIndividualDeletedMessagesFieldBuilder().addBuilder( + index, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.getDefaultInstance()); + } + public java.util.List + getIndividualDeletedMessagesBuilderList() { + return getIndividualDeletedMessagesFieldBuilder().getBuilderList(); + } + private com.google.protobuf.RepeatedFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder> + getIndividualDeletedMessagesFieldBuilder() { + if (individualDeletedMessagesBuilder_ == null) { + individualDeletedMessagesBuilder_ = new com.google.protobuf.RepeatedFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder>( + individualDeletedMessages_, + ((bitField0_ & 0x00000004) == 0x00000004), + getParentForChildren(), + isClean()); + individualDeletedMessages_ = null; + } + return individualDeletedMessagesBuilder_; + } + // @@protoc_insertion_point(builder_scope:PositionInfo) } @@ -1499,91 +1784,80 @@ public Builder clearEntryId() { // @@protoc_insertion_point(class_scope:PositionInfo) } - public interface ManagedCursorInfoOrBuilder + public interface NestedPositionInfoOrBuilder extends com.google.protobuf.MessageOrBuilder { - // required int64 cursorsLedgerId = 1; - boolean hasCursorsLedgerId(); - long getCursorsLedgerId(); - - // optional int64 markDeleteLedgerId = 2; - boolean hasMarkDeleteLedgerId(); - long getMarkDeleteLedgerId(); + // required int64 ledgerId = 1; + boolean hasLedgerId(); + long getLedgerId(); - // optional int64 markDeleteEntryId = 3; - boolean hasMarkDeleteEntryId(); - long getMarkDeleteEntryId(); + // required int64 entryId = 2; + boolean hasEntryId(); + long getEntryId(); } - public static final class ManagedCursorInfo extends + public static final class NestedPositionInfo extends com.google.protobuf.GeneratedMessage - implements ManagedCursorInfoOrBuilder { - // Use ManagedCursorInfo.newBuilder() to construct. - private ManagedCursorInfo(Builder builder) { + implements NestedPositionInfoOrBuilder { + // Use NestedPositionInfo.newBuilder() to construct. + private NestedPositionInfo(Builder builder) { super(builder); } - private ManagedCursorInfo(boolean noInit) {} + private NestedPositionInfo(boolean noInit) {} - private static final ManagedCursorInfo defaultInstance; - public static ManagedCursorInfo getDefaultInstance() { + private static final NestedPositionInfo defaultInstance; + public static NestedPositionInfo getDefaultInstance() { return defaultInstance; } - public ManagedCursorInfo getDefaultInstanceForType() { + public NestedPositionInfo getDefaultInstanceForType() { return defaultInstance; } public static final com.google.protobuf.Descriptors.Descriptor getDescriptor() { - return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_ManagedCursorInfo_descriptor; + return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_NestedPositionInfo_descriptor; } protected com.google.protobuf.GeneratedMessage.FieldAccessorTable internalGetFieldAccessorTable() { - return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_ManagedCursorInfo_fieldAccessorTable; + return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_NestedPositionInfo_fieldAccessorTable; } private int bitField0_; - // required int64 cursorsLedgerId = 1; - public static final int CURSORSLEDGERID_FIELD_NUMBER = 1; - private long cursorsLedgerId_; - public boolean hasCursorsLedgerId() { + // required int64 ledgerId = 1; + public static final int LEDGERID_FIELD_NUMBER = 1; + private long ledgerId_; + public boolean hasLedgerId() { return ((bitField0_ & 0x00000001) == 0x00000001); } - public long getCursorsLedgerId() { - return cursorsLedgerId_; + public long getLedgerId() { + return ledgerId_; } - // optional int64 markDeleteLedgerId = 2; - public static final int MARKDELETELEDGERID_FIELD_NUMBER = 2; - private long markDeleteLedgerId_; - public boolean hasMarkDeleteLedgerId() { + // required int64 entryId = 2; + public static final int ENTRYID_FIELD_NUMBER = 2; + private long entryId_; + public boolean hasEntryId() { return ((bitField0_ & 0x00000002) == 0x00000002); } - public long getMarkDeleteLedgerId() { - return markDeleteLedgerId_; - } - - // optional int64 markDeleteEntryId = 3; - public static final int MARKDELETEENTRYID_FIELD_NUMBER = 3; - private long markDeleteEntryId_; - public boolean hasMarkDeleteEntryId() { - return ((bitField0_ & 0x00000004) == 0x00000004); - } - public long getMarkDeleteEntryId() { - return markDeleteEntryId_; + public long getEntryId() { + return entryId_; } private void initFields() { - cursorsLedgerId_ = 0L; - markDeleteLedgerId_ = 0L; - markDeleteEntryId_ = 0L; + ledgerId_ = 0L; + entryId_ = 0L; } private byte memoizedIsInitialized = -1; public final boolean isInitialized() { byte isInitialized = memoizedIsInitialized; if (isInitialized != -1) return isInitialized == 1; - if (!hasCursorsLedgerId()) { + if (!hasLedgerId()) { + memoizedIsInitialized = 0; + return false; + } + if (!hasEntryId()) { memoizedIsInitialized = 0; return false; } @@ -1595,13 +1869,10 @@ public void writeTo(com.google.protobuf.CodedOutputStream output) throws java.io.IOException { getSerializedSize(); if (((bitField0_ & 0x00000001) == 0x00000001)) { - output.writeInt64(1, cursorsLedgerId_); + output.writeInt64(1, ledgerId_); } if (((bitField0_ & 0x00000002) == 0x00000002)) { - output.writeInt64(2, markDeleteLedgerId_); - } - if (((bitField0_ & 0x00000004) == 0x00000004)) { - output.writeInt64(3, markDeleteEntryId_); + output.writeInt64(2, entryId_); } getUnknownFields().writeTo(output); } @@ -1614,15 +1885,11 @@ public int getSerializedSize() { size = 0; if (((bitField0_ & 0x00000001) == 0x00000001)) { size += com.google.protobuf.CodedOutputStream - .computeInt64Size(1, cursorsLedgerId_); + .computeInt64Size(1, ledgerId_); } if (((bitField0_ & 0x00000002) == 0x00000002)) { size += com.google.protobuf.CodedOutputStream - .computeInt64Size(2, markDeleteLedgerId_); - } - if (((bitField0_ & 0x00000004) == 0x00000004)) { - size += com.google.protobuf.CodedOutputStream - .computeInt64Size(3, markDeleteEntryId_); + .computeInt64Size(2, entryId_); } size += getUnknownFields().getSerializedSize(); memoizedSerializedSize = size; @@ -1636,41 +1903,41 @@ protected java.lang.Object writeReplace() return super.writeReplace(); } - public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom( + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo parseFrom( com.google.protobuf.ByteString data) throws com.google.protobuf.InvalidProtocolBufferException { return newBuilder().mergeFrom(data).buildParsed(); } - public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom( + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo parseFrom( com.google.protobuf.ByteString data, com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws com.google.protobuf.InvalidProtocolBufferException { return newBuilder().mergeFrom(data, extensionRegistry) .buildParsed(); } - public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom(byte[] data) + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo parseFrom(byte[] data) throws com.google.protobuf.InvalidProtocolBufferException { return newBuilder().mergeFrom(data).buildParsed(); } - public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom( + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo parseFrom( byte[] data, com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws com.google.protobuf.InvalidProtocolBufferException { return newBuilder().mergeFrom(data, extensionRegistry) .buildParsed(); } - public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom(java.io.InputStream input) + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo parseFrom(java.io.InputStream input) throws java.io.IOException { return newBuilder().mergeFrom(input).buildParsed(); } - public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom( + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo parseFrom( java.io.InputStream input, com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { return newBuilder().mergeFrom(input, extensionRegistry) .buildParsed(); } - public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseDelimitedFrom(java.io.InputStream input) + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo parseDelimitedFrom(java.io.InputStream input) throws java.io.IOException { Builder builder = newBuilder(); if (builder.mergeDelimitedFrom(input)) { @@ -1679,7 +1946,7 @@ public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInf return null; } } - public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseDelimitedFrom( + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo parseDelimitedFrom( java.io.InputStream input, com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { @@ -1690,12 +1957,12 @@ public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInf return null; } } - public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom( + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo parseFrom( com.google.protobuf.CodedInputStream input) throws java.io.IOException { return newBuilder().mergeFrom(input).buildParsed(); } - public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom( + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo parseFrom( com.google.protobuf.CodedInputStream input, com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { @@ -1705,7 +1972,7 @@ public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInf public static Builder newBuilder() { return Builder.create(); } public Builder newBuilderForType() { return newBuilder(); } - public static Builder newBuilder(org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo prototype) { + public static Builder newBuilder(org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo prototype) { return newBuilder().mergeFrom(prototype); } public Builder toBuilder() { return newBuilder(this); } @@ -1718,18 +1985,18 @@ protected Builder newBuilderForType( } public static final class Builder extends com.google.protobuf.GeneratedMessage.Builder - implements org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfoOrBuilder { + implements org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder { public static final com.google.protobuf.Descriptors.Descriptor getDescriptor() { - return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_ManagedCursorInfo_descriptor; + return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_NestedPositionInfo_descriptor; } protected com.google.protobuf.GeneratedMessage.FieldAccessorTable internalGetFieldAccessorTable() { - return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_ManagedCursorInfo_fieldAccessorTable; + return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_NestedPositionInfo_fieldAccessorTable; } - // Construct using org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo.newBuilder() + // Construct using org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.newBuilder() private Builder() { maybeForceBuilderInitialization(); } @@ -1748,12 +2015,10 @@ private static Builder create() { public Builder clear() { super.clear(); - cursorsLedgerId_ = 0L; + ledgerId_ = 0L; bitField0_ = (bitField0_ & ~0x00000001); - markDeleteLedgerId_ = 0L; + entryId_ = 0L; bitField0_ = (bitField0_ & ~0x00000002); - markDeleteEntryId_ = 0L; - bitField0_ = (bitField0_ & ~0x00000004); return this; } @@ -1763,24 +2028,24 @@ public Builder clone() { public com.google.protobuf.Descriptors.Descriptor getDescriptorForType() { - return org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo.getDescriptor(); + return org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDescriptor(); } - public org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo getDefaultInstanceForType() { - return org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo.getDefaultInstance(); + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo getDefaultInstanceForType() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDefaultInstance(); } - public org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo build() { - org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo result = buildPartial(); + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo build() { + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo result = buildPartial(); if (!result.isInitialized()) { throw newUninitializedMessageException(result); } return result; } - private org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo buildParsed() + private org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo buildParsed() throws com.google.protobuf.InvalidProtocolBufferException { - org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo result = buildPartial(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo result = buildPartial(); if (!result.isInitialized()) { throw newUninitializedMessageException( result).asInvalidProtocolBufferException(); @@ -1788,37 +2053,1128 @@ private org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo buil return result; } - public org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo buildPartial() { - org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo result = new org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo(this); + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo buildPartial() { + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo result = new org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo(this); int from_bitField0_ = bitField0_; int to_bitField0_ = 0; if (((from_bitField0_ & 0x00000001) == 0x00000001)) { to_bitField0_ |= 0x00000001; } - result.cursorsLedgerId_ = cursorsLedgerId_; + result.ledgerId_ = ledgerId_; if (((from_bitField0_ & 0x00000002) == 0x00000002)) { to_bitField0_ |= 0x00000002; } - result.markDeleteLedgerId_ = markDeleteLedgerId_; - if (((from_bitField0_ & 0x00000004) == 0x00000004)) { - to_bitField0_ |= 0x00000004; - } - result.markDeleteEntryId_ = markDeleteEntryId_; + result.entryId_ = entryId_; result.bitField0_ = to_bitField0_; onBuilt(); return result; } public Builder mergeFrom(com.google.protobuf.Message other) { - if (other instanceof org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo) { - return mergeFrom((org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo)other); + if (other instanceof org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo) { + return mergeFrom((org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo)other); } else { super.mergeFrom(other); return this; } } - public Builder mergeFrom(org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo other) { + public Builder mergeFrom(org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo other) { + if (other == org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDefaultInstance()) return this; + if (other.hasLedgerId()) { + setLedgerId(other.getLedgerId()); + } + if (other.hasEntryId()) { + setEntryId(other.getEntryId()); + } + this.mergeUnknownFields(other.getUnknownFields()); + return this; + } + + public final boolean isInitialized() { + if (!hasLedgerId()) { + + return false; + } + if (!hasEntryId()) { + + return false; + } + return true; + } + + public Builder mergeFrom( + com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + com.google.protobuf.UnknownFieldSet.Builder unknownFields = + com.google.protobuf.UnknownFieldSet.newBuilder( + this.getUnknownFields()); + while (true) { + int tag = input.readTag(); + switch (tag) { + case 0: + this.setUnknownFields(unknownFields.build()); + onChanged(); + return this; + default: { + if (!parseUnknownField(input, unknownFields, + extensionRegistry, tag)) { + this.setUnknownFields(unknownFields.build()); + onChanged(); + return this; + } + break; + } + case 8: { + bitField0_ |= 0x00000001; + ledgerId_ = input.readInt64(); + break; + } + case 16: { + bitField0_ |= 0x00000002; + entryId_ = input.readInt64(); + break; + } + } + } + } + + private int bitField0_; + + // required int64 ledgerId = 1; + private long ledgerId_ ; + public boolean hasLedgerId() { + return ((bitField0_ & 0x00000001) == 0x00000001); + } + public long getLedgerId() { + return ledgerId_; + } + public Builder setLedgerId(long value) { + bitField0_ |= 0x00000001; + ledgerId_ = value; + onChanged(); + return this; + } + public Builder clearLedgerId() { + bitField0_ = (bitField0_ & ~0x00000001); + ledgerId_ = 0L; + onChanged(); + return this; + } + + // required int64 entryId = 2; + private long entryId_ ; + public boolean hasEntryId() { + return ((bitField0_ & 0x00000002) == 0x00000002); + } + public long getEntryId() { + return entryId_; + } + public Builder setEntryId(long value) { + bitField0_ |= 0x00000002; + entryId_ = value; + onChanged(); + return this; + } + public Builder clearEntryId() { + bitField0_ = (bitField0_ & ~0x00000002); + entryId_ = 0L; + onChanged(); + return this; + } + + // @@protoc_insertion_point(builder_scope:NestedPositionInfo) + } + + static { + defaultInstance = new NestedPositionInfo(true); + defaultInstance.initFields(); + } + + // @@protoc_insertion_point(class_scope:NestedPositionInfo) + } + + public interface MessageRangeOrBuilder + extends com.google.protobuf.MessageOrBuilder { + + // required .NestedPositionInfo lowerEndpoint = 1; + boolean hasLowerEndpoint(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo getLowerEndpoint(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder getLowerEndpointOrBuilder(); + + // required .NestedPositionInfo upperEndpoint = 2; + boolean hasUpperEndpoint(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo getUpperEndpoint(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder getUpperEndpointOrBuilder(); + } + public static final class MessageRange extends + com.google.protobuf.GeneratedMessage + implements MessageRangeOrBuilder { + // Use MessageRange.newBuilder() to construct. + private MessageRange(Builder builder) { + super(builder); + } + private MessageRange(boolean noInit) {} + + private static final MessageRange defaultInstance; + public static MessageRange getDefaultInstance() { + return defaultInstance; + } + + public MessageRange getDefaultInstanceForType() { + return defaultInstance; + } + + public static final com.google.protobuf.Descriptors.Descriptor + getDescriptor() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_MessageRange_descriptor; + } + + protected com.google.protobuf.GeneratedMessage.FieldAccessorTable + internalGetFieldAccessorTable() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_MessageRange_fieldAccessorTable; + } + + private int bitField0_; + // required .NestedPositionInfo lowerEndpoint = 1; + public static final int LOWERENDPOINT_FIELD_NUMBER = 1; + private org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo lowerEndpoint_; + public boolean hasLowerEndpoint() { + return ((bitField0_ & 0x00000001) == 0x00000001); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo getLowerEndpoint() { + return lowerEndpoint_; + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder getLowerEndpointOrBuilder() { + return lowerEndpoint_; + } + + // required .NestedPositionInfo upperEndpoint = 2; + public static final int UPPERENDPOINT_FIELD_NUMBER = 2; + private org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo upperEndpoint_; + public boolean hasUpperEndpoint() { + return ((bitField0_ & 0x00000002) == 0x00000002); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo getUpperEndpoint() { + return upperEndpoint_; + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder getUpperEndpointOrBuilder() { + return upperEndpoint_; + } + + private void initFields() { + lowerEndpoint_ = org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDefaultInstance(); + upperEndpoint_ = org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDefaultInstance(); + } + private byte memoizedIsInitialized = -1; + public final boolean isInitialized() { + byte isInitialized = memoizedIsInitialized; + if (isInitialized != -1) return isInitialized == 1; + + if (!hasLowerEndpoint()) { + memoizedIsInitialized = 0; + return false; + } + if (!hasUpperEndpoint()) { + memoizedIsInitialized = 0; + return false; + } + if (!getLowerEndpoint().isInitialized()) { + memoizedIsInitialized = 0; + return false; + } + if (!getUpperEndpoint().isInitialized()) { + memoizedIsInitialized = 0; + return false; + } + memoizedIsInitialized = 1; + return true; + } + + public void writeTo(com.google.protobuf.CodedOutputStream output) + throws java.io.IOException { + getSerializedSize(); + if (((bitField0_ & 0x00000001) == 0x00000001)) { + output.writeMessage(1, lowerEndpoint_); + } + if (((bitField0_ & 0x00000002) == 0x00000002)) { + output.writeMessage(2, upperEndpoint_); + } + getUnknownFields().writeTo(output); + } + + private int memoizedSerializedSize = -1; + public int getSerializedSize() { + int size = memoizedSerializedSize; + if (size != -1) return size; + + size = 0; + if (((bitField0_ & 0x00000001) == 0x00000001)) { + size += com.google.protobuf.CodedOutputStream + .computeMessageSize(1, lowerEndpoint_); + } + if (((bitField0_ & 0x00000002) == 0x00000002)) { + size += com.google.protobuf.CodedOutputStream + .computeMessageSize(2, upperEndpoint_); + } + size += getUnknownFields().getSerializedSize(); + memoizedSerializedSize = size; + return size; + } + + private static final long serialVersionUID = 0L; + @java.lang.Override + protected java.lang.Object writeReplace() + throws java.io.ObjectStreamException { + return super.writeReplace(); + } + + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange parseFrom( + com.google.protobuf.ByteString data) + throws com.google.protobuf.InvalidProtocolBufferException { + return newBuilder().mergeFrom(data).buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange parseFrom( + com.google.protobuf.ByteString data, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return newBuilder().mergeFrom(data, extensionRegistry) + .buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange parseFrom(byte[] data) + throws com.google.protobuf.InvalidProtocolBufferException { + return newBuilder().mergeFrom(data).buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange parseFrom( + byte[] data, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return newBuilder().mergeFrom(data, extensionRegistry) + .buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange parseFrom(java.io.InputStream input) + throws java.io.IOException { + return newBuilder().mergeFrom(input).buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange parseFrom( + java.io.InputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + return newBuilder().mergeFrom(input, extensionRegistry) + .buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange parseDelimitedFrom(java.io.InputStream input) + throws java.io.IOException { + Builder builder = newBuilder(); + if (builder.mergeDelimitedFrom(input)) { + return builder.buildParsed(); + } else { + return null; + } + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange parseDelimitedFrom( + java.io.InputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + Builder builder = newBuilder(); + if (builder.mergeDelimitedFrom(input, extensionRegistry)) { + return builder.buildParsed(); + } else { + return null; + } + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange parseFrom( + com.google.protobuf.CodedInputStream input) + throws java.io.IOException { + return newBuilder().mergeFrom(input).buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange parseFrom( + com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + return newBuilder().mergeFrom(input, extensionRegistry) + .buildParsed(); + } + + public static Builder newBuilder() { return Builder.create(); } + public Builder newBuilderForType() { return newBuilder(); } + public static Builder newBuilder(org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange prototype) { + return newBuilder().mergeFrom(prototype); + } + public Builder toBuilder() { return newBuilder(this); } + + @java.lang.Override + protected Builder newBuilderForType( + com.google.protobuf.GeneratedMessage.BuilderParent parent) { + Builder builder = new Builder(parent); + return builder; + } + public static final class Builder extends + com.google.protobuf.GeneratedMessage.Builder + implements org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder { + public static final com.google.protobuf.Descriptors.Descriptor + getDescriptor() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_MessageRange_descriptor; + } + + protected com.google.protobuf.GeneratedMessage.FieldAccessorTable + internalGetFieldAccessorTable() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_MessageRange_fieldAccessorTable; + } + + // Construct using org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.newBuilder() + private Builder() { + maybeForceBuilderInitialization(); + } + + private Builder(BuilderParent parent) { + super(parent); + maybeForceBuilderInitialization(); + } + private void maybeForceBuilderInitialization() { + if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) { + getLowerEndpointFieldBuilder(); + getUpperEndpointFieldBuilder(); + } + } + private static Builder create() { + return new Builder(); + } + + public Builder clear() { + super.clear(); + if (lowerEndpointBuilder_ == null) { + lowerEndpoint_ = org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDefaultInstance(); + } else { + lowerEndpointBuilder_.clear(); + } + bitField0_ = (bitField0_ & ~0x00000001); + if (upperEndpointBuilder_ == null) { + upperEndpoint_ = org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDefaultInstance(); + } else { + upperEndpointBuilder_.clear(); + } + bitField0_ = (bitField0_ & ~0x00000002); + return this; + } + + public Builder clone() { + return create().mergeFrom(buildPartial()); + } + + public com.google.protobuf.Descriptors.Descriptor + getDescriptorForType() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.getDescriptor(); + } + + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange getDefaultInstanceForType() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.getDefaultInstance(); + } + + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange build() { + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange result = buildPartial(); + if (!result.isInitialized()) { + throw newUninitializedMessageException(result); + } + return result; + } + + private org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange buildParsed() + throws com.google.protobuf.InvalidProtocolBufferException { + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange result = buildPartial(); + if (!result.isInitialized()) { + throw newUninitializedMessageException( + result).asInvalidProtocolBufferException(); + } + return result; + } + + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange buildPartial() { + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange result = new org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange(this); + int from_bitField0_ = bitField0_; + int to_bitField0_ = 0; + if (((from_bitField0_ & 0x00000001) == 0x00000001)) { + to_bitField0_ |= 0x00000001; + } + if (lowerEndpointBuilder_ == null) { + result.lowerEndpoint_ = lowerEndpoint_; + } else { + result.lowerEndpoint_ = lowerEndpointBuilder_.build(); + } + if (((from_bitField0_ & 0x00000002) == 0x00000002)) { + to_bitField0_ |= 0x00000002; + } + if (upperEndpointBuilder_ == null) { + result.upperEndpoint_ = upperEndpoint_; + } else { + result.upperEndpoint_ = upperEndpointBuilder_.build(); + } + result.bitField0_ = to_bitField0_; + onBuilt(); + return result; + } + + public Builder mergeFrom(com.google.protobuf.Message other) { + if (other instanceof org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange) { + return mergeFrom((org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange)other); + } else { + super.mergeFrom(other); + return this; + } + } + + public Builder mergeFrom(org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange other) { + if (other == org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.getDefaultInstance()) return this; + if (other.hasLowerEndpoint()) { + mergeLowerEndpoint(other.getLowerEndpoint()); + } + if (other.hasUpperEndpoint()) { + mergeUpperEndpoint(other.getUpperEndpoint()); + } + this.mergeUnknownFields(other.getUnknownFields()); + return this; + } + + public final boolean isInitialized() { + if (!hasLowerEndpoint()) { + + return false; + } + if (!hasUpperEndpoint()) { + + return false; + } + if (!getLowerEndpoint().isInitialized()) { + + return false; + } + if (!getUpperEndpoint().isInitialized()) { + + return false; + } + return true; + } + + public Builder mergeFrom( + com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + com.google.protobuf.UnknownFieldSet.Builder unknownFields = + com.google.protobuf.UnknownFieldSet.newBuilder( + this.getUnknownFields()); + while (true) { + int tag = input.readTag(); + switch (tag) { + case 0: + this.setUnknownFields(unknownFields.build()); + onChanged(); + return this; + default: { + if (!parseUnknownField(input, unknownFields, + extensionRegistry, tag)) { + this.setUnknownFields(unknownFields.build()); + onChanged(); + return this; + } + break; + } + case 10: { + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder subBuilder = org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.newBuilder(); + if (hasLowerEndpoint()) { + subBuilder.mergeFrom(getLowerEndpoint()); + } + input.readMessage(subBuilder, extensionRegistry); + setLowerEndpoint(subBuilder.buildPartial()); + break; + } + case 18: { + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder subBuilder = org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.newBuilder(); + if (hasUpperEndpoint()) { + subBuilder.mergeFrom(getUpperEndpoint()); + } + input.readMessage(subBuilder, extensionRegistry); + setUpperEndpoint(subBuilder.buildPartial()); + break; + } + } + } + } + + private int bitField0_; + + // required .NestedPositionInfo lowerEndpoint = 1; + private org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo lowerEndpoint_ = org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDefaultInstance(); + private com.google.protobuf.SingleFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo, org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder> lowerEndpointBuilder_; + public boolean hasLowerEndpoint() { + return ((bitField0_ & 0x00000001) == 0x00000001); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo getLowerEndpoint() { + if (lowerEndpointBuilder_ == null) { + return lowerEndpoint_; + } else { + return lowerEndpointBuilder_.getMessage(); + } + } + public Builder setLowerEndpoint(org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo value) { + if (lowerEndpointBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + lowerEndpoint_ = value; + onChanged(); + } else { + lowerEndpointBuilder_.setMessage(value); + } + bitField0_ |= 0x00000001; + return this; + } + public Builder setLowerEndpoint( + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder builderForValue) { + if (lowerEndpointBuilder_ == null) { + lowerEndpoint_ = builderForValue.build(); + onChanged(); + } else { + lowerEndpointBuilder_.setMessage(builderForValue.build()); + } + bitField0_ |= 0x00000001; + return this; + } + public Builder mergeLowerEndpoint(org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo value) { + if (lowerEndpointBuilder_ == null) { + if (((bitField0_ & 0x00000001) == 0x00000001) && + lowerEndpoint_ != org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDefaultInstance()) { + lowerEndpoint_ = + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.newBuilder(lowerEndpoint_).mergeFrom(value).buildPartial(); + } else { + lowerEndpoint_ = value; + } + onChanged(); + } else { + lowerEndpointBuilder_.mergeFrom(value); + } + bitField0_ |= 0x00000001; + return this; + } + public Builder clearLowerEndpoint() { + if (lowerEndpointBuilder_ == null) { + lowerEndpoint_ = org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDefaultInstance(); + onChanged(); + } else { + lowerEndpointBuilder_.clear(); + } + bitField0_ = (bitField0_ & ~0x00000001); + return this; + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder getLowerEndpointBuilder() { + bitField0_ |= 0x00000001; + onChanged(); + return getLowerEndpointFieldBuilder().getBuilder(); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder getLowerEndpointOrBuilder() { + if (lowerEndpointBuilder_ != null) { + return lowerEndpointBuilder_.getMessageOrBuilder(); + } else { + return lowerEndpoint_; + } + } + private com.google.protobuf.SingleFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo, org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder> + getLowerEndpointFieldBuilder() { + if (lowerEndpointBuilder_ == null) { + lowerEndpointBuilder_ = new com.google.protobuf.SingleFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo, org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder>( + lowerEndpoint_, + getParentForChildren(), + isClean()); + lowerEndpoint_ = null; + } + return lowerEndpointBuilder_; + } + + // required .NestedPositionInfo upperEndpoint = 2; + private org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo upperEndpoint_ = org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDefaultInstance(); + private com.google.protobuf.SingleFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo, org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder> upperEndpointBuilder_; + public boolean hasUpperEndpoint() { + return ((bitField0_ & 0x00000002) == 0x00000002); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo getUpperEndpoint() { + if (upperEndpointBuilder_ == null) { + return upperEndpoint_; + } else { + return upperEndpointBuilder_.getMessage(); + } + } + public Builder setUpperEndpoint(org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo value) { + if (upperEndpointBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + upperEndpoint_ = value; + onChanged(); + } else { + upperEndpointBuilder_.setMessage(value); + } + bitField0_ |= 0x00000002; + return this; + } + public Builder setUpperEndpoint( + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder builderForValue) { + if (upperEndpointBuilder_ == null) { + upperEndpoint_ = builderForValue.build(); + onChanged(); + } else { + upperEndpointBuilder_.setMessage(builderForValue.build()); + } + bitField0_ |= 0x00000002; + return this; + } + public Builder mergeUpperEndpoint(org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo value) { + if (upperEndpointBuilder_ == null) { + if (((bitField0_ & 0x00000002) == 0x00000002) && + upperEndpoint_ != org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDefaultInstance()) { + upperEndpoint_ = + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.newBuilder(upperEndpoint_).mergeFrom(value).buildPartial(); + } else { + upperEndpoint_ = value; + } + onChanged(); + } else { + upperEndpointBuilder_.mergeFrom(value); + } + bitField0_ |= 0x00000002; + return this; + } + public Builder clearUpperEndpoint() { + if (upperEndpointBuilder_ == null) { + upperEndpoint_ = org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.getDefaultInstance(); + onChanged(); + } else { + upperEndpointBuilder_.clear(); + } + bitField0_ = (bitField0_ & ~0x00000002); + return this; + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder getUpperEndpointBuilder() { + bitField0_ |= 0x00000002; + onChanged(); + return getUpperEndpointFieldBuilder().getBuilder(); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder getUpperEndpointOrBuilder() { + if (upperEndpointBuilder_ != null) { + return upperEndpointBuilder_.getMessageOrBuilder(); + } else { + return upperEndpoint_; + } + } + private com.google.protobuf.SingleFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo, org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder> + getUpperEndpointFieldBuilder() { + if (upperEndpointBuilder_ == null) { + upperEndpointBuilder_ = new com.google.protobuf.SingleFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo, org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBuilder>( + upperEndpoint_, + getParentForChildren(), + isClean()); + upperEndpoint_ = null; + } + return upperEndpointBuilder_; + } + + // @@protoc_insertion_point(builder_scope:MessageRange) + } + + static { + defaultInstance = new MessageRange(true); + defaultInstance.initFields(); + } + + // @@protoc_insertion_point(class_scope:MessageRange) + } + + public interface ManagedCursorInfoOrBuilder + extends com.google.protobuf.MessageOrBuilder { + + // required int64 cursorsLedgerId = 1; + boolean hasCursorsLedgerId(); + long getCursorsLedgerId(); + + // optional int64 markDeleteLedgerId = 2; + boolean hasMarkDeleteLedgerId(); + long getMarkDeleteLedgerId(); + + // optional int64 markDeleteEntryId = 3; + boolean hasMarkDeleteEntryId(); + long getMarkDeleteEntryId(); + + // repeated .MessageRange individualDeletedMessages = 4; + java.util.List + getIndividualDeletedMessagesList(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange getIndividualDeletedMessages(int index); + int getIndividualDeletedMessagesCount(); + java.util.List + getIndividualDeletedMessagesOrBuilderList(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder getIndividualDeletedMessagesOrBuilder( + int index); + } + public static final class ManagedCursorInfo extends + com.google.protobuf.GeneratedMessage + implements ManagedCursorInfoOrBuilder { + // Use ManagedCursorInfo.newBuilder() to construct. + private ManagedCursorInfo(Builder builder) { + super(builder); + } + private ManagedCursorInfo(boolean noInit) {} + + private static final ManagedCursorInfo defaultInstance; + public static ManagedCursorInfo getDefaultInstance() { + return defaultInstance; + } + + public ManagedCursorInfo getDefaultInstanceForType() { + return defaultInstance; + } + + public static final com.google.protobuf.Descriptors.Descriptor + getDescriptor() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_ManagedCursorInfo_descriptor; + } + + protected com.google.protobuf.GeneratedMessage.FieldAccessorTable + internalGetFieldAccessorTable() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_ManagedCursorInfo_fieldAccessorTable; + } + + private int bitField0_; + // required int64 cursorsLedgerId = 1; + public static final int CURSORSLEDGERID_FIELD_NUMBER = 1; + private long cursorsLedgerId_; + public boolean hasCursorsLedgerId() { + return ((bitField0_ & 0x00000001) == 0x00000001); + } + public long getCursorsLedgerId() { + return cursorsLedgerId_; + } + + // optional int64 markDeleteLedgerId = 2; + public static final int MARKDELETELEDGERID_FIELD_NUMBER = 2; + private long markDeleteLedgerId_; + public boolean hasMarkDeleteLedgerId() { + return ((bitField0_ & 0x00000002) == 0x00000002); + } + public long getMarkDeleteLedgerId() { + return markDeleteLedgerId_; + } + + // optional int64 markDeleteEntryId = 3; + public static final int MARKDELETEENTRYID_FIELD_NUMBER = 3; + private long markDeleteEntryId_; + public boolean hasMarkDeleteEntryId() { + return ((bitField0_ & 0x00000004) == 0x00000004); + } + public long getMarkDeleteEntryId() { + return markDeleteEntryId_; + } + + // repeated .MessageRange individualDeletedMessages = 4; + public static final int INDIVIDUALDELETEDMESSAGES_FIELD_NUMBER = 4; + private java.util.List individualDeletedMessages_; + public java.util.List getIndividualDeletedMessagesList() { + return individualDeletedMessages_; + } + public java.util.List + getIndividualDeletedMessagesOrBuilderList() { + return individualDeletedMessages_; + } + public int getIndividualDeletedMessagesCount() { + return individualDeletedMessages_.size(); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange getIndividualDeletedMessages(int index) { + return individualDeletedMessages_.get(index); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder getIndividualDeletedMessagesOrBuilder( + int index) { + return individualDeletedMessages_.get(index); + } + + private void initFields() { + cursorsLedgerId_ = 0L; + markDeleteLedgerId_ = 0L; + markDeleteEntryId_ = 0L; + individualDeletedMessages_ = java.util.Collections.emptyList(); + } + private byte memoizedIsInitialized = -1; + public final boolean isInitialized() { + byte isInitialized = memoizedIsInitialized; + if (isInitialized != -1) return isInitialized == 1; + + if (!hasCursorsLedgerId()) { + memoizedIsInitialized = 0; + return false; + } + for (int i = 0; i < getIndividualDeletedMessagesCount(); i++) { + if (!getIndividualDeletedMessages(i).isInitialized()) { + memoizedIsInitialized = 0; + return false; + } + } + memoizedIsInitialized = 1; + return true; + } + + public void writeTo(com.google.protobuf.CodedOutputStream output) + throws java.io.IOException { + getSerializedSize(); + if (((bitField0_ & 0x00000001) == 0x00000001)) { + output.writeInt64(1, cursorsLedgerId_); + } + if (((bitField0_ & 0x00000002) == 0x00000002)) { + output.writeInt64(2, markDeleteLedgerId_); + } + if (((bitField0_ & 0x00000004) == 0x00000004)) { + output.writeInt64(3, markDeleteEntryId_); + } + for (int i = 0; i < individualDeletedMessages_.size(); i++) { + output.writeMessage(4, individualDeletedMessages_.get(i)); + } + getUnknownFields().writeTo(output); + } + + private int memoizedSerializedSize = -1; + public int getSerializedSize() { + int size = memoizedSerializedSize; + if (size != -1) return size; + + size = 0; + if (((bitField0_ & 0x00000001) == 0x00000001)) { + size += com.google.protobuf.CodedOutputStream + .computeInt64Size(1, cursorsLedgerId_); + } + if (((bitField0_ & 0x00000002) == 0x00000002)) { + size += com.google.protobuf.CodedOutputStream + .computeInt64Size(2, markDeleteLedgerId_); + } + if (((bitField0_ & 0x00000004) == 0x00000004)) { + size += com.google.protobuf.CodedOutputStream + .computeInt64Size(3, markDeleteEntryId_); + } + for (int i = 0; i < individualDeletedMessages_.size(); i++) { + size += com.google.protobuf.CodedOutputStream + .computeMessageSize(4, individualDeletedMessages_.get(i)); + } + size += getUnknownFields().getSerializedSize(); + memoizedSerializedSize = size; + return size; + } + + private static final long serialVersionUID = 0L; + @java.lang.Override + protected java.lang.Object writeReplace() + throws java.io.ObjectStreamException { + return super.writeReplace(); + } + + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom( + com.google.protobuf.ByteString data) + throws com.google.protobuf.InvalidProtocolBufferException { + return newBuilder().mergeFrom(data).buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom( + com.google.protobuf.ByteString data, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return newBuilder().mergeFrom(data, extensionRegistry) + .buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom(byte[] data) + throws com.google.protobuf.InvalidProtocolBufferException { + return newBuilder().mergeFrom(data).buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom( + byte[] data, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return newBuilder().mergeFrom(data, extensionRegistry) + .buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom(java.io.InputStream input) + throws java.io.IOException { + return newBuilder().mergeFrom(input).buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom( + java.io.InputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + return newBuilder().mergeFrom(input, extensionRegistry) + .buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseDelimitedFrom(java.io.InputStream input) + throws java.io.IOException { + Builder builder = newBuilder(); + if (builder.mergeDelimitedFrom(input)) { + return builder.buildParsed(); + } else { + return null; + } + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseDelimitedFrom( + java.io.InputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + Builder builder = newBuilder(); + if (builder.mergeDelimitedFrom(input, extensionRegistry)) { + return builder.buildParsed(); + } else { + return null; + } + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom( + com.google.protobuf.CodedInputStream input) + throws java.io.IOException { + return newBuilder().mergeFrom(input).buildParsed(); + } + public static org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo parseFrom( + com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + return newBuilder().mergeFrom(input, extensionRegistry) + .buildParsed(); + } + + public static Builder newBuilder() { return Builder.create(); } + public Builder newBuilderForType() { return newBuilder(); } + public static Builder newBuilder(org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo prototype) { + return newBuilder().mergeFrom(prototype); + } + public Builder toBuilder() { return newBuilder(this); } + + @java.lang.Override + protected Builder newBuilderForType( + com.google.protobuf.GeneratedMessage.BuilderParent parent) { + Builder builder = new Builder(parent); + return builder; + } + public static final class Builder extends + com.google.protobuf.GeneratedMessage.Builder + implements org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfoOrBuilder { + public static final com.google.protobuf.Descriptors.Descriptor + getDescriptor() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_ManagedCursorInfo_descriptor; + } + + protected com.google.protobuf.GeneratedMessage.FieldAccessorTable + internalGetFieldAccessorTable() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.internal_static_ManagedCursorInfo_fieldAccessorTable; + } + + // Construct using org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo.newBuilder() + private Builder() { + maybeForceBuilderInitialization(); + } + + private Builder(BuilderParent parent) { + super(parent); + maybeForceBuilderInitialization(); + } + private void maybeForceBuilderInitialization() { + if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) { + getIndividualDeletedMessagesFieldBuilder(); + } + } + private static Builder create() { + return new Builder(); + } + + public Builder clear() { + super.clear(); + cursorsLedgerId_ = 0L; + bitField0_ = (bitField0_ & ~0x00000001); + markDeleteLedgerId_ = 0L; + bitField0_ = (bitField0_ & ~0x00000002); + markDeleteEntryId_ = 0L; + bitField0_ = (bitField0_ & ~0x00000004); + if (individualDeletedMessagesBuilder_ == null) { + individualDeletedMessages_ = java.util.Collections.emptyList(); + bitField0_ = (bitField0_ & ~0x00000008); + } else { + individualDeletedMessagesBuilder_.clear(); + } + return this; + } + + public Builder clone() { + return create().mergeFrom(buildPartial()); + } + + public com.google.protobuf.Descriptors.Descriptor + getDescriptorForType() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo.getDescriptor(); + } + + public org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo getDefaultInstanceForType() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo.getDefaultInstance(); + } + + public org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo build() { + org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo result = buildPartial(); + if (!result.isInitialized()) { + throw newUninitializedMessageException(result); + } + return result; + } + + private org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo buildParsed() + throws com.google.protobuf.InvalidProtocolBufferException { + org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo result = buildPartial(); + if (!result.isInitialized()) { + throw newUninitializedMessageException( + result).asInvalidProtocolBufferException(); + } + return result; + } + + public org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo buildPartial() { + org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo result = new org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo(this); + int from_bitField0_ = bitField0_; + int to_bitField0_ = 0; + if (((from_bitField0_ & 0x00000001) == 0x00000001)) { + to_bitField0_ |= 0x00000001; + } + result.cursorsLedgerId_ = cursorsLedgerId_; + if (((from_bitField0_ & 0x00000002) == 0x00000002)) { + to_bitField0_ |= 0x00000002; + } + result.markDeleteLedgerId_ = markDeleteLedgerId_; + if (((from_bitField0_ & 0x00000004) == 0x00000004)) { + to_bitField0_ |= 0x00000004; + } + result.markDeleteEntryId_ = markDeleteEntryId_; + if (individualDeletedMessagesBuilder_ == null) { + if (((bitField0_ & 0x00000008) == 0x00000008)) { + individualDeletedMessages_ = java.util.Collections.unmodifiableList(individualDeletedMessages_); + bitField0_ = (bitField0_ & ~0x00000008); + } + result.individualDeletedMessages_ = individualDeletedMessages_; + } else { + result.individualDeletedMessages_ = individualDeletedMessagesBuilder_.build(); + } + result.bitField0_ = to_bitField0_; + onBuilt(); + return result; + } + + public Builder mergeFrom(com.google.protobuf.Message other) { + if (other instanceof org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo) { + return mergeFrom((org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo)other); + } else { + super.mergeFrom(other); + return this; + } + } + + public Builder mergeFrom(org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo other) { if (other == org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo.getDefaultInstance()) return this; if (other.hasCursorsLedgerId()) { setCursorsLedgerId(other.getCursorsLedgerId()); @@ -1829,6 +3185,32 @@ public Builder mergeFrom(org.apache.bookkeeper.mledger.proto.MLDataFormats.Manag if (other.hasMarkDeleteEntryId()) { setMarkDeleteEntryId(other.getMarkDeleteEntryId()); } + if (individualDeletedMessagesBuilder_ == null) { + if (!other.individualDeletedMessages_.isEmpty()) { + if (individualDeletedMessages_.isEmpty()) { + individualDeletedMessages_ = other.individualDeletedMessages_; + bitField0_ = (bitField0_ & ~0x00000008); + } else { + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.addAll(other.individualDeletedMessages_); + } + onChanged(); + } + } else { + if (!other.individualDeletedMessages_.isEmpty()) { + if (individualDeletedMessagesBuilder_.isEmpty()) { + individualDeletedMessagesBuilder_.dispose(); + individualDeletedMessagesBuilder_ = null; + individualDeletedMessages_ = other.individualDeletedMessages_; + bitField0_ = (bitField0_ & ~0x00000008); + individualDeletedMessagesBuilder_ = + com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders ? + getIndividualDeletedMessagesFieldBuilder() : null; + } else { + individualDeletedMessagesBuilder_.addAllMessages(other.individualDeletedMessages_); + } + } + } this.mergeUnknownFields(other.getUnknownFields()); return this; } @@ -1838,6 +3220,12 @@ public final boolean isInitialized() { return false; } + for (int i = 0; i < getIndividualDeletedMessagesCount(); i++) { + if (!getIndividualDeletedMessages(i).isInitialized()) { + + return false; + } + } return true; } @@ -1879,6 +3267,12 @@ public Builder mergeFrom( markDeleteEntryId_ = input.readInt64(); break; } + case 34: { + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder subBuilder = org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.newBuilder(); + input.readMessage(subBuilder, extensionRegistry); + addIndividualDeletedMessages(subBuilder.buildPartial()); + break; + } } } } @@ -1948,6 +3342,192 @@ public Builder clearMarkDeleteEntryId() { return this; } + // repeated .MessageRange individualDeletedMessages = 4; + private java.util.List individualDeletedMessages_ = + java.util.Collections.emptyList(); + private void ensureIndividualDeletedMessagesIsMutable() { + if (!((bitField0_ & 0x00000008) == 0x00000008)) { + individualDeletedMessages_ = new java.util.ArrayList(individualDeletedMessages_); + bitField0_ |= 0x00000008; + } + } + + private com.google.protobuf.RepeatedFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder> individualDeletedMessagesBuilder_; + + public java.util.List getIndividualDeletedMessagesList() { + if (individualDeletedMessagesBuilder_ == null) { + return java.util.Collections.unmodifiableList(individualDeletedMessages_); + } else { + return individualDeletedMessagesBuilder_.getMessageList(); + } + } + public int getIndividualDeletedMessagesCount() { + if (individualDeletedMessagesBuilder_ == null) { + return individualDeletedMessages_.size(); + } else { + return individualDeletedMessagesBuilder_.getCount(); + } + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange getIndividualDeletedMessages(int index) { + if (individualDeletedMessagesBuilder_ == null) { + return individualDeletedMessages_.get(index); + } else { + return individualDeletedMessagesBuilder_.getMessage(index); + } + } + public Builder setIndividualDeletedMessages( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange value) { + if (individualDeletedMessagesBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.set(index, value); + onChanged(); + } else { + individualDeletedMessagesBuilder_.setMessage(index, value); + } + return this; + } + public Builder setIndividualDeletedMessages( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder builderForValue) { + if (individualDeletedMessagesBuilder_ == null) { + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.set(index, builderForValue.build()); + onChanged(); + } else { + individualDeletedMessagesBuilder_.setMessage(index, builderForValue.build()); + } + return this; + } + public Builder addIndividualDeletedMessages(org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange value) { + if (individualDeletedMessagesBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.add(value); + onChanged(); + } else { + individualDeletedMessagesBuilder_.addMessage(value); + } + return this; + } + public Builder addIndividualDeletedMessages( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange value) { + if (individualDeletedMessagesBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.add(index, value); + onChanged(); + } else { + individualDeletedMessagesBuilder_.addMessage(index, value); + } + return this; + } + public Builder addIndividualDeletedMessages( + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder builderForValue) { + if (individualDeletedMessagesBuilder_ == null) { + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.add(builderForValue.build()); + onChanged(); + } else { + individualDeletedMessagesBuilder_.addMessage(builderForValue.build()); + } + return this; + } + public Builder addIndividualDeletedMessages( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder builderForValue) { + if (individualDeletedMessagesBuilder_ == null) { + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.add(index, builderForValue.build()); + onChanged(); + } else { + individualDeletedMessagesBuilder_.addMessage(index, builderForValue.build()); + } + return this; + } + public Builder addAllIndividualDeletedMessages( + java.lang.Iterable values) { + if (individualDeletedMessagesBuilder_ == null) { + ensureIndividualDeletedMessagesIsMutable(); + super.addAll(values, individualDeletedMessages_); + onChanged(); + } else { + individualDeletedMessagesBuilder_.addAllMessages(values); + } + return this; + } + public Builder clearIndividualDeletedMessages() { + if (individualDeletedMessagesBuilder_ == null) { + individualDeletedMessages_ = java.util.Collections.emptyList(); + bitField0_ = (bitField0_ & ~0x00000008); + onChanged(); + } else { + individualDeletedMessagesBuilder_.clear(); + } + return this; + } + public Builder removeIndividualDeletedMessages(int index) { + if (individualDeletedMessagesBuilder_ == null) { + ensureIndividualDeletedMessagesIsMutable(); + individualDeletedMessages_.remove(index); + onChanged(); + } else { + individualDeletedMessagesBuilder_.remove(index); + } + return this; + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder getIndividualDeletedMessagesBuilder( + int index) { + return getIndividualDeletedMessagesFieldBuilder().getBuilder(index); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder getIndividualDeletedMessagesOrBuilder( + int index) { + if (individualDeletedMessagesBuilder_ == null) { + return individualDeletedMessages_.get(index); } else { + return individualDeletedMessagesBuilder_.getMessageOrBuilder(index); + } + } + public java.util.List + getIndividualDeletedMessagesOrBuilderList() { + if (individualDeletedMessagesBuilder_ != null) { + return individualDeletedMessagesBuilder_.getMessageOrBuilderList(); + } else { + return java.util.Collections.unmodifiableList(individualDeletedMessages_); + } + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder addIndividualDeletedMessagesBuilder() { + return getIndividualDeletedMessagesFieldBuilder().addBuilder( + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.getDefaultInstance()); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder addIndividualDeletedMessagesBuilder( + int index) { + return getIndividualDeletedMessagesFieldBuilder().addBuilder( + index, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.getDefaultInstance()); + } + public java.util.List + getIndividualDeletedMessagesBuilderList() { + return getIndividualDeletedMessagesFieldBuilder().getBuilderList(); + } + private com.google.protobuf.RepeatedFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder> + getIndividualDeletedMessagesFieldBuilder() { + if (individualDeletedMessagesBuilder_ == null) { + individualDeletedMessagesBuilder_ = new com.google.protobuf.RepeatedFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder>( + individualDeletedMessages_, + ((bitField0_ & 0x00000008) == 0x00000008), + getParentForChildren(), + isClean()); + individualDeletedMessages_ = null; + } + return individualDeletedMessagesBuilder_; + } + // @@protoc_insertion_point(builder_scope:ManagedCursorInfo) } @@ -1974,6 +3554,16 @@ public Builder clearMarkDeleteEntryId() { private static com.google.protobuf.GeneratedMessage.FieldAccessorTable internal_static_PositionInfo_fieldAccessorTable; + private static com.google.protobuf.Descriptors.Descriptor + internal_static_NestedPositionInfo_descriptor; + private static + com.google.protobuf.GeneratedMessage.FieldAccessorTable + internal_static_NestedPositionInfo_fieldAccessorTable; + private static com.google.protobuf.Descriptors.Descriptor + internal_static_MessageRange_descriptor; + private static + com.google.protobuf.GeneratedMessage.FieldAccessorTable + internal_static_MessageRange_fieldAccessorTable; private static com.google.protobuf.Descriptors.Descriptor internal_static_ManagedCursorInfo_descriptor; private static @@ -1992,12 +3582,19 @@ public Builder clearMarkDeleteEntryId() { "\021ManagedLedgerInfo\0221\n\nledgerInfo\030\001 \003(\0132\035" + ".ManagedLedgerInfo.LedgerInfo\032P\n\nLedgerI" + "nfo\022\020\n\010ledgerId\030\001 \002(\003\022\017\n\007entries\030\002 \001(\003\022\014" + - "\n\004size\030\003 \001(\003\022\021\n\ttimestamp\030\004 \001(\003\"1\n\014Posit" + + "\n\004size\030\003 \001(\003\022\021\n\ttimestamp\030\004 \001(\003\"c\n\014Posit" + "ionInfo\022\020\n\010ledgerId\030\001 \002(\003\022\017\n\007entryId\030\002 \002" + - "(\003\"c\n\021ManagedCursorInfo\022\027\n\017cursorsLedger" + - "Id\030\001 \002(\003\022\032\n\022markDeleteLedgerId\030\002 \001(\003\022\031\n\021" + - "markDeleteEntryId\030\003 \001(\003B\'\n#org.apache.bo" + - "okkeeper.mledger.protoH\001" + "(\003\0220\n\031individualDeletedMessages\030\003 \003(\0132\r." + + "MessageRange\"7\n\022NestedPositionInfo\022\020\n\010le" + + "dgerId\030\001 \002(\003\022\017\n\007entryId\030\002 \002(\003\"f\n\014Message" + + "Range\022*\n\rlowerEndpoint\030\001 \002(\0132\023.NestedPos", + "itionInfo\022*\n\rupperEndpoint\030\002 \002(\0132\023.Neste" + + "dPositionInfo\"\225\001\n\021ManagedCursorInfo\022\027\n\017c" + + "ursorsLedgerId\030\001 \002(\003\022\032\n\022markDeleteLedger" + + "Id\030\002 \001(\003\022\031\n\021markDeleteEntryId\030\003 \001(\003\0220\n\031i" + + "ndividualDeletedMessages\030\004 \003(\0132\r.Message" + + "RangeB\'\n#org.apache.bookkeeper.mledger.p" + + "rotoH\001" }; com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner = new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() { @@ -2025,15 +3622,31 @@ public com.google.protobuf.ExtensionRegistry assignDescriptors( internal_static_PositionInfo_fieldAccessorTable = new com.google.protobuf.GeneratedMessage.FieldAccessorTable( internal_static_PositionInfo_descriptor, - new java.lang.String[] { "LedgerId", "EntryId", }, + new java.lang.String[] { "LedgerId", "EntryId", "IndividualDeletedMessages", }, org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo.class, org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo.Builder.class); - internal_static_ManagedCursorInfo_descriptor = + internal_static_NestedPositionInfo_descriptor = getDescriptor().getMessageTypes().get(2); + internal_static_NestedPositionInfo_fieldAccessorTable = new + com.google.protobuf.GeneratedMessage.FieldAccessorTable( + internal_static_NestedPositionInfo_descriptor, + new java.lang.String[] { "LedgerId", "EntryId", }, + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.class, + org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo.Builder.class); + internal_static_MessageRange_descriptor = + getDescriptor().getMessageTypes().get(3); + internal_static_MessageRange_fieldAccessorTable = new + com.google.protobuf.GeneratedMessage.FieldAccessorTable( + internal_static_MessageRange_descriptor, + new java.lang.String[] { "LowerEndpoint", "UpperEndpoint", }, + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.class, + org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder.class); + internal_static_ManagedCursorInfo_descriptor = + getDescriptor().getMessageTypes().get(4); internal_static_ManagedCursorInfo_fieldAccessorTable = new com.google.protobuf.GeneratedMessage.FieldAccessorTable( internal_static_ManagedCursorInfo_descriptor, - new java.lang.String[] { "CursorsLedgerId", "MarkDeleteLedgerId", "MarkDeleteEntryId", }, + new java.lang.String[] { "CursorsLedgerId", "MarkDeleteLedgerId", "MarkDeleteEntryId", "IndividualDeletedMessages", }, org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo.class, org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo.Builder.class); return null; diff --git a/managed-ledger/src/main/proto/MLDataFormats.proto b/managed-ledger/src/main/proto/MLDataFormats.proto index 67da6eecc04d4..bd841c678f9f9 100644 --- a/managed-ledger/src/main/proto/MLDataFormats.proto +++ b/managed-ledger/src/main/proto/MLDataFormats.proto @@ -30,6 +30,17 @@ message ManagedLedgerInfo { message PositionInfo { required int64 ledgerId = 1; required int64 entryId = 2; + repeated MessageRange individualDeletedMessages = 3; +} + +message NestedPositionInfo { + required int64 ledgerId = 1; + required int64 entryId = 2; +} + +message MessageRange { + required NestedPositionInfo lowerEndpoint = 1; + required NestedPositionInfo upperEndpoint = 2; } message ManagedCursorInfo { @@ -40,4 +51,5 @@ message ManagedCursorInfo { // Last snapshot of the mark-delete position optional int64 markDeleteLedgerId = 2; optional int64 markDeleteEntryId = 3; + repeated MessageRange individualDeletedMessages = 4; } diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java index a36e65e26d3ad..763b81b91a538 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java @@ -15,30 +15,17 @@ */ package org.apache.bookkeeper.mledger.impl; -import static org.testng.Assert.assertEquals; -import static org.testng.Assert.assertFalse; -import static org.testng.Assert.assertTrue; -import static org.testng.Assert.fail; +import com.google.common.collect.Lists; +import com.google.common.collect.Sets; +import org.apache.bookkeeper.mledger.*; +import org.apache.bookkeeper.mledger.AsyncCallbacks.*; +import org.testng.annotations.Test; import java.util.List; import java.util.Set; +import java.util.function.Predicate; -import org.apache.bookkeeper.mledger.AsyncCallbacks; -import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback; -import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback; -import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback; -import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback; -import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback; -import org.apache.bookkeeper.mledger.AsyncCallbacks.SkipEntriesCallback; -import org.apache.bookkeeper.mledger.Entry; -import org.apache.bookkeeper.mledger.ManagedCursor; -import org.apache.bookkeeper.mledger.ManagedLedgerException; -import org.apache.bookkeeper.mledger.Position; -import org.testng.annotations.Test; - -import com.google.common.base.Predicate; -import com.google.common.collect.Lists; -import com.google.common.collect.Sets; +import static org.testng.Assert.*; @Test public class ManagedCursorContainerTest { @@ -123,7 +110,7 @@ public void close() { } @Override - public void asyncClose(AsyncCallbacks.CloseCallback callback, Object ctx) { + public void asyncClose(CloseCallback callback, Object ctx) { } @Override @@ -160,11 +147,11 @@ public Position findNewestMatching(Predicate condition) @Override public void asyncFindNewestMatching(FindPositionConstraint constraint, Predicate condition, - AsyncCallbacks.FindEntryCallback callback, Object ctx) { + FindEntryCallback callback, Object ctx) { } @Override - public void asyncResetCursor(final Position position, AsyncCallbacks.ResetCursorCallback callback) { + public void asyncResetCursor(final Position position, ResetCursorCallback callback) { } diff --git a/pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java index ab3099362e5d5..7eb32d3f71ad2 100644 --- a/pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java @@ -178,6 +178,11 @@ public class ServiceConfiguration implements PulsarConfiguration{ private int managedLedgerCursorMaxEntriesPerLedger = 50000; // Max time before triggering a rollover on a cursor ledger private int managedLedgerCursorRolloverTimeInSeconds = 14400; + // Max number of entries to append to a ledger before triggering a rollover + // A ledger rollover is triggered on these conditions Either the max + // rollover time has been reached or max entries have been written to the + // ledged and at least min-time has passed + private int managedLedgerMaxUnackedRangesToPersist = 1000; /*** --- Load balancer --- ****/ // Enable load balancer @@ -680,6 +685,14 @@ public void setManagedLedgerCursorRolloverTimeInSeconds(int managedLedgerCursorR this.managedLedgerCursorRolloverTimeInSeconds = managedLedgerCursorRolloverTimeInSeconds; } + public int getManagedLedgerMaxUnackedRangesToPersist() { + return managedLedgerMaxUnackedRangesToPersist; + } + + public void setManagedLedgerMaxUnackedRangesToPersist(int managedLedgerMaxUnackedRangesToPersist) { + this.managedLedgerMaxUnackedRangesToPersist = managedLedgerMaxUnackedRangesToPersist; + } + public boolean isLoadBalancerEnabled() { return loadBalancerEnabled; } diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java index c999790df3958..306f58025f54a 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java @@ -495,6 +495,7 @@ public CompletableFuture getManagedLedgerConfig(Destination config.setThrottleMarkDelete(persistencePolicies.getManagedLedgerMaxMarkDeleteRate()); config.setDigestType(DigestType.CRC32); + config.setMaxUnackedRangesToPersist(serviceConfig.getManagedLedgerMaxUnackedRangesToPersist()); config.setMaxEntriesPerLedger(serviceConfig.getManagedLedgerMaxEntriesPerLedger()); config.setMinimumRolloverTime(serviceConfig.getManagedLedgerMinLedgerRolloverTimeMinutes(), TimeUnit.MINUTES); From 9865970b86be967e9e7f17258679a869032b73c8 Mon Sep 17 00:00:00 2001 From: Sebastian Schepens Date: Thu, 23 Feb 2017 10:55:05 -0300 Subject: [PATCH 2/2] Rearrange lock and synchornized blocks to prevent deadlocks --- .../mledger/impl/ManagedCursorImpl.java | 159 +++++++++--------- 1 file changed, 78 insertions(+), 81 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java index 00b676ef98ac8..d7c723e78573b 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java @@ -20,12 +20,12 @@ import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun; import java.util.ArrayDeque; -import java.util.HashSet; import java.util.List; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.*; +import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; @@ -685,14 +685,12 @@ protected void internalResetCursor(final PositionImpl newPosition, AsyncCallbacks.ResetCursorCallback resetCursorCallback) { log.info("[{}] Initiate reset position to {} on cursor {}", ledger.getName(), newPosition, name); - synchronized (pendingMarkDeleteOps) { - if (!RESET_CURSOR_IN_PROGRESS_UPDATER.compareAndSet(this, FALSE, TRUE)) { - log.error("[{}] reset requested - position [{}], previous reset in progress - cursor {}", - ledger.getName(), newPosition, name); - resetCursorCallback.resetFailed( - new ManagedLedgerException.ConcurrentFindCursorPositionException("reset already in progress"), - newPosition); - } + if (!RESET_CURSOR_IN_PROGRESS_UPDATER.compareAndSet(this, FALSE, TRUE)) { + log.error("[{}] reset requested - position [{}], previous reset in progress - cursor {}", + ledger.getName(), newPosition, name); + resetCursorCallback.resetFailed( + new ManagedLedgerException.ConcurrentFindCursorPositionException("reset already in progress"), + newPosition); } final AsyncCallbacks.ResetCursorCallback callback = resetCursorCallback; @@ -730,10 +728,10 @@ public void operationComplete() { } synchronized (pendingMarkDeleteOps) { pendingMarkDeleteOps.clear(); - if (!RESET_CURSOR_IN_PROGRESS_UPDATER.compareAndSet(ManagedCursorImpl.this, TRUE, FALSE)) { - log.error("[{}] expected reset position [{}], but another reset in progress on cursor {}", - ledger.getName(), newPosition, name); - } + } + if (!RESET_CURSOR_IN_PROGRESS_UPDATER.compareAndSet(ManagedCursorImpl.this, TRUE, FALSE)) { + log.error("[{}] expected reset position [{}], but another reset in progress on cursor {}", + ledger.getName(), newPosition, name); } callback.resetComplete(newPosition); @@ -741,11 +739,9 @@ public void operationComplete() { @Override public void operationFailed(ManagedLedgerException exception) { - synchronized (pendingMarkDeleteOps) { - if (!RESET_CURSOR_IN_PROGRESS_UPDATER.compareAndSet(ManagedCursorImpl.this, TRUE, FALSE)) { - log.error("[{}] expected reset position [{}], but another reset in progress on cursor {}", - ledger.getName(), newPosition, name); - } + if (!RESET_CURSOR_IN_PROGRESS_UPDATER.compareAndSet(ManagedCursorImpl.this, TRUE, FALSE)) { + log.error("[{}] expected reset position [{}], but another reset in progress on cursor {}", + ledger.getName(), newPosition, name); } callback.resetFailed(new ManagedLedgerException.InvalidCursorPositionException( "unable to persist position for cursor reset " + newPosition.toString()), newPosition); @@ -1248,36 +1244,37 @@ private void internalAsyncMarkDelete(final PositionImpl newPosition, final MarkD PendingMarkDeleteEntry mdEntry = new PendingMarkDeleteEntry(newPosition, callback, ctx); // We cannot write to the ledger during the switch, need to wait until the new metadata ledger is available - synchronized (pendingMarkDeleteOps) { - // The state might have changed while we were waiting on the queue mutex - switch (STATE_UPDATER.get(this)) { - case Closed: - callback.markDeleteFailed(new ManagedLedgerException("Cursor was already closed"), ctx); - return; + switch (STATE_UPDATER.get(this)) { + case Closed: + callback.markDeleteFailed(new ManagedLedgerException("Cursor was already closed"), ctx); + return; - case NoLedger: - // We need to create a new ledger to write into - startCreatingNewMetadataLedger(); - // fall through - case SwitchingLedger: + case NoLedger: + // We need to create a new ledger to write into + startCreatingNewMetadataLedger(); + // fall through + case SwitchingLedger: + synchronized (pendingMarkDeleteOps) { pendingMarkDeleteOps.add(mdEntry); - break; + } + break; - case Open: - if (PENDING_READ_OPS_UPDATER.get(this) > 0) { - // Wait until no read operation are pending + case Open: + if (PENDING_READ_OPS_UPDATER.get(this) > 0) { + // Wait until no read operation are pending + synchronized (pendingMarkDeleteOps) { pendingMarkDeleteOps.add(mdEntry); - } else { - // Execute the mark delete immediately - internalMarkDelete(mdEntry); } - break; - - default: - log.error("[{}][{}] Invalid cursor state: {}", ledger.getName(), name, state); - callback.markDeleteFailed(new ManagedLedgerException("Cursor was in invalid state: " + state), ctx); - break; + } else { + // Execute the mark delete immediately + internalMarkDelete(mdEntry); } + break; + + default: + log.error("[{}][{}] Invalid cursor state: {}", ledger.getName(), name, state); + callback.markDeleteFailed(new ManagedLedgerException("Cursor was in invalid state: " + state), ctx); + break; } } @@ -1408,6 +1405,8 @@ public void asyncDelete(Position pos, final AsyncCallbacks.DeleteCallback callba lock.writeLock().lock(); + boolean canMarkDelete = false; + PositionImpl newMarkDeletePosition = null; try { if (log.isDebugEnabled()) { log.debug( @@ -1423,9 +1422,6 @@ public void asyncDelete(Position pos, final AsyncCallbacks.DeleteCallback callba return; } - boolean canMarkDelete = false; - PositionImpl newMarkDeletePosition = null; - if (previousPosition.compareTo(markDeletePosition) == 0 && individualDeletedMessages.isEmpty()) { if (log.isDebugEnabled()) { log.debug("[{}][{}] Immediately mark-delete to position {}", ledger.getName(), name, position); @@ -1462,7 +1458,19 @@ public void asyncDelete(Position pos, final AsyncCallbacks.DeleteCallback callba newMarkDeletePosition = range.upperEndpoint(); } } + } catch (Exception e) { + log.warn("[{}] [{}] Error doing asyncDelete [{}]", ledger.getName(), name, e.getMessage(), e); + if (log.isDebugEnabled()) { + log.debug("[{}] Consumer {} cursor asyncDelete error, counters: consumed {} mdPos {} rdPos {}", + ledger.getName(), name, messagesConsumedCounter, markDeletePosition, readPosition); + } + callback.deleteFailed(new ManagedLedgerException(e), ctx); + return; + } finally { + lock.writeLock().unlock(); + } + try { if (canMarkDelete) { newMarkDeletePosition = setAcknowledgedPosition(newMarkDeletePosition); @@ -1501,8 +1509,6 @@ public void markDeleteFailed(ManagedLedgerException exception, Object ctx) { ledger.getName(), name, messagesConsumedCounter, markDeletePosition, readPosition); } callback.deleteFailed(new ManagedLedgerException(e), ctx); - } finally { - lock.writeLock().unlock(); } } @@ -1720,12 +1726,9 @@ void createNewMetadataLedger() { @Override public void operationComplete() { // We now have a new ledger where we can write - synchronized (pendingMarkDeleteOps) { - flushPendingMarkDeletes(); - - // Resume normal mark-delete operations - STATE_UPDATER.set(ManagedCursorImpl.this, State.Open); - } + flushPendingMarkDeletes(); + // Resume normal mark-delete operations + STATE_UPDATER.set(ManagedCursorImpl.this, State.Open); } @Override @@ -1746,28 +1749,24 @@ public void operationFailed(ManagedLedgerException exception) { } private void flushPendingMarkDeletes() { - if (!pendingMarkDeleteOps.isEmpty()) { - if (RESET_CURSOR_IN_PROGRESS_UPDATER.get(this) == TRUE) { - failPendingMarkDeletes(); - } else { - internalFlushPendingMarkDeletes(); + PendingMarkDeleteEntry lastEntry = null; + synchronized (pendingMarkDeleteOps) { + if (!pendingMarkDeleteOps.isEmpty()) { + if (RESET_CURSOR_IN_PROGRESS_UPDATER.get(this) == TRUE) { + for (PendingMarkDeleteEntry e : pendingMarkDeleteOps) { + e.callback.markDeleteFailed(new ManagedLedgerException("reset cursor in progress"), e.ctx); + } + pendingMarkDeleteOps.clear(); + } else { + lastEntry = pendingMarkDeleteOps.getLast(); + lastEntry.callbackGroup = Lists.newArrayList(pendingMarkDeleteOps); + pendingMarkDeleteOps.clear(); + } } } - } - - private void failPendingMarkDeletes() { - for (PendingMarkDeleteEntry e : pendingMarkDeleteOps) { - e.callback.markDeleteFailed(new ManagedLedgerException("reset cursor in progress"), e.ctx); + if (lastEntry != null) { + internalMarkDelete(lastEntry); } - pendingMarkDeleteOps.clear(); - } - - void internalFlushPendingMarkDeletes() { - PendingMarkDeleteEntry lastEntry = pendingMarkDeleteOps.getLast(); - lastEntry.callbackGroup = Lists.newArrayList(pendingMarkDeleteOps); - pendingMarkDeleteOps.clear(); - - internalMarkDelete(lastEntry); } void createNewMetadataLedger(final VoidCallback callback) { @@ -1995,15 +1994,13 @@ void decrementPendingMarkDeleteCount() { void readOperationCompleted() { if (PENDING_READ_OPS_UPDATER.decrementAndGet(this) == 0) { - synchronized (pendingMarkDeleteOps) { - if (STATE_UPDATER.get(this) == State.Open) { - // Flush the pending writes only if the state is open. - flushPendingMarkDeletes(); - } else if (PENDING_MARK_DELETED_SUBMITTED_COUNT_UPDATER.get(this) != 0) { - log.info( - "[{}] read operation completed and cursor was closed. need to call any queued cursor close", - name); - } + if (STATE_UPDATER.get(this) == State.Open) { + // Flush the pending writes only if the state is open. + flushPendingMarkDeletes(); + } else if (PENDING_MARK_DELETED_SUBMITTED_COUNT_UPDATER.get(this) != 0) { + log.info( + "[{}] read operation completed and cursor was closed. need to call any queued cursor close", + name); } } }