From 7029fc50672d91b47b67ede99d2ba72a6a584305 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 30 Aug 2017 09:58:37 -0700 Subject: [PATCH 1/2] Add user-defined properties to cursor position --- .../bookkeeper/mledger/ManagedCursor.java | 44 +- .../bookkeeper/mledger/ManagedLedgerInfo.java | 1 + .../mledger/impl/ManagedCursorImpl.java | 228 +-- .../impl/ManagedLedgerFactoryImpl.java | 9 + .../mledger/impl/NonDurableCursorImpl.java | 6 +- .../mledger/proto/MLDataFormats.java | 1296 +++++++++++++++-- .../src/main/proto/MLDataFormats.proto | 14 + .../impl/ManagedCursorContainerTest.java | 18 + .../impl/ManagedCursorPropertiesTest.java | 106 ++ .../service/persistent/PersistentTopic.java | 1 + .../data/PersistentTopicInternalStats.java | 2 + 11 files changed, 1507 insertions(+), 218 deletions(-) create mode 100644 managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorPropertiesTest.java 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 ebff5836d6749..03c9cab994bfb 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 @@ -19,6 +19,7 @@ package org.apache.bookkeeper.mledger; import java.util.List; +import java.util.Map; import java.util.Set; import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback; @@ -56,6 +57,11 @@ public enum IndividualDeletedEntries { */ public String getName(); + /** + * Return any properties that were associated with the last stored position + */ + public Map getProperties(); + /** * Read entries from the ManagedLedger, up to the specified number. The returned list can be smaller. * @@ -179,10 +185,25 @@ public void asyncGetNthEntry(int N, IndividualDeletedEntries deletedEntries, Rea * * @param position * the last position that have been successfully consumed + * * @throws ManagedLedgerException */ public void markDelete(Position position) throws InterruptedException, ManagedLedgerException; + /** + * This signals that the reader is done with all the entries up to "position" (included). This can potentially + * trigger a ledger deletion, if all the other cursors are done too with the underlying ledger. + * + * @param position + * the last position that have been successfully consumed + * @param properties + * additional user-defined properties that can be associated with a particular cursor position + * + * @throws ManagedLedgerException + */ + public void markDelete(Position position, Map properties) + throws InterruptedException, ManagedLedgerException; + /** * Asynchronous mark delete * @@ -196,6 +217,21 @@ public void asyncGetNthEntry(int N, IndividualDeletedEntries deletedEntries, Rea */ public void asyncMarkDelete(Position position, MarkDeleteCallback callback, Object ctx); + /** + * Asynchronous mark delete + * + * @see #markDelete(Position) + * @param position + * the last position that have been successfully consumed + * @param properties + * additional user-defined properties that can be associated with a particular cursor position + * @param callback + * callback object + * @param ctx + * opaque context + */ + public void asyncMarkDelete(Position position, Map properties, MarkDeleteCallback callback, Object ctx); + /** * Delete a single message *

@@ -372,7 +408,7 @@ public List replayEntries(Set positions) * callback object returning the list of entries * @param ctx * opaque context - * @return skipped positions + * @return skipped positions * set of positions which are already deleted/acknowledged and skipped while replaying them */ public Set asyncReplayEntries(Set positions, ReadEntriesCallback callback, Object ctx); @@ -404,19 +440,19 @@ public List replayEntries(Set positions) /** * Activate cursor: EntryCacheManager caches entries only for activated-cursors - * + * */ public void setActive(); /** * Deactivate cursor - * + * */ public void setInactive(); /** * Checks if cursor is active or not. - * + * * @return */ public boolean isActive(); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerInfo.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerInfo.java index 3a010cf159db6..e6e9f15c619ec 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerInfo.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerInfo.java @@ -52,6 +52,7 @@ public static class CursorInfo { // Last snapshot of the mark-delete position public PositionInfo markDelete; public List individualDeletedMessages; + public Map properties; } public static class PositionInfo { 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 52bad592b8bd9..04c6bbe1af70c 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 @@ -25,6 +25,7 @@ import java.util.ArrayDeque; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -61,6 +62,7 @@ import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback; import org.apache.bookkeeper.mledger.impl.MetaStore.Stat; import org.apache.bookkeeper.mledger.proto.MLDataFormats; +import org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty; import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo; import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo; import org.apache.bookkeeper.mledger.util.Pair; @@ -71,6 +73,7 @@ import com.google.common.base.Predicate; import com.google.common.collect.Collections2; import com.google.common.collect.Lists; +import com.google.common.collect.Maps; import com.google.common.collect.Range; import com.google.common.collect.RangeSet; import com.google.common.collect.Sets; @@ -87,9 +90,10 @@ public class ManagedCursorImpl implements ManagedCursor { protected volatile PositionImpl markDeletePosition; protected volatile PositionImpl readPosition; + private volatile PendingMarkDeleteEntry lastMarkDeleteEntry; - protected static final AtomicReferenceFieldUpdater WAITING_READ_OP_UPDATER = - AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, OpReadEntry.class, "waitingReadOp"); + protected static final AtomicReferenceFieldUpdater WAITING_READ_OP_UPDATER = AtomicReferenceFieldUpdater + .newUpdater(ManagedCursorImpl.class, OpReadEntry.class, "waitingReadOp"); @SuppressWarnings("unused") private volatile OpReadEntry waitingReadOp = null; @@ -99,8 +103,8 @@ public class ManagedCursorImpl implements ManagedCursor { .newUpdater(ManagedCursorImpl.class, "resetCursorInProgress"); @SuppressWarnings("unused") private volatile int resetCursorInProgress = FALSE; - private static final AtomicIntegerFieldUpdater PENDING_READ_OPS_UPDATER = - AtomicIntegerFieldUpdater.newUpdater(ManagedCursorImpl.class, "pendingReadOps"); + private static final AtomicIntegerFieldUpdater PENDING_READ_OPS_UPDATER = AtomicIntegerFieldUpdater + .newUpdater(ManagedCursorImpl.class, "pendingReadOps"); @SuppressWarnings("unused") private volatile int pendingReadOps = 0; @@ -123,22 +127,25 @@ class PendingMarkDeleteEntry { final PositionImpl newPosition; final MarkDeleteCallback callback; final Object ctx; + final Map properties; // If the callbackGroup is set, it means this mark-delete request was done on behalf of a group of request (just // persist the last one in the chain). In this case we need to trigger the callbacks for every request in the // group. List callbackGroup; - public PendingMarkDeleteEntry(PositionImpl newPosition, MarkDeleteCallback callback, Object ctx) { + public PendingMarkDeleteEntry(PositionImpl newPosition, Map properties, + MarkDeleteCallback callback, Object ctx) { this.newPosition = PositionImpl.get(newPosition); + this.properties = properties; this.callback = callback; this.ctx = ctx; } } private final ArrayDeque pendingMarkDeleteOps = new ArrayDeque<>(); - private static final AtomicIntegerFieldUpdater PENDING_MARK_DELETED_SUBMITTED_COUNT_UPDATER = - AtomicIntegerFieldUpdater.newUpdater(ManagedCursorImpl.class, "pendingMarkDeletedSubmittedCount"); + private static final AtomicIntegerFieldUpdater PENDING_MARK_DELETED_SUBMITTED_COUNT_UPDATER = AtomicIntegerFieldUpdater + .newUpdater(ManagedCursorImpl.class, "pendingMarkDeletedSubmittedCount"); @SuppressWarnings("unused") private volatile int pendingMarkDeletedSubmittedCount = 0; private long lastLedgerSwitchTimestamp; @@ -151,8 +158,8 @@ enum State { Closed // The managed cursor has been closed } - private static final AtomicReferenceFieldUpdater STATE_UPDATER = - AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, State.class, "state"); + private static final AtomicReferenceFieldUpdater STATE_UPDATER = AtomicReferenceFieldUpdater + .newUpdater(ManagedCursorImpl.class, State.class, "state"); private volatile State state = null; public interface VoidCallback { @@ -181,6 +188,11 @@ public interface VoidCallback { } } + @Override + public Map getProperties() { + return lastMarkDeleteEntry.properties; + } + /** * Performs the initial recovery, reading the mark-deleted position from the ledger and then calling initialize to * have a new opened ledger @@ -202,7 +214,18 @@ public void operationComplete(ManagedCursorInfo info, Stat stat) { if (info.getIndividualDeletedMessagesCount() > 0) { recoverIndividualDeletedMessages(info.getIndividualDeletedMessagesList()); } - recoveredCursor(recoveredPosition); + + Map recoveredProperties = Collections.emptyMap(); + if (info.getPropertiesCount() > 0) { + // Recover properties map + recoveredProperties = Maps.newHashMap(); + for (int i = 0; i < info.getPropertiesCount(); i++) { + LongProperty property = info.getProperties(i); + recoveredProperties.put(property.getName(), property.getValue()); + } + } + + recoveredCursor(recoveredPosition, recoveredProperties); callback.operationComplete(); } else { // Need to proceed and read the last entry in the specified ledger to find out the last position @@ -269,11 +292,21 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac return; } + Map recoveredProperties = Collections.emptyMap(); + if (positionInfo.getPropertiesCount() > 0) { + // Recover properties map + recoveredProperties = Maps.newHashMap(); + for (int i = 0; i < positionInfo.getPropertiesCount(); i++) { + LongProperty property = positionInfo.getProperties(i); + recoveredProperties.put(property.getName(), property.getValue()); + } + } + PositionImpl position = new PositionImpl(positionInfo); if (positionInfo.getIndividualDeletedMessagesCount() > 0) { recoverIndividualDeletedMessages(positionInfo.getIndividualDeletedMessagesList()); } - recoveredCursor(position); + recoveredCursor(position, recoveredProperties); callback.operationComplete(); }, null); }, null); @@ -283,19 +316,15 @@ private void recoverIndividualDeletedMessages(List i lock.writeLock().lock(); try { individualDeletedMessages.clear(); - individualDeletedMessagesList - .forEach(messageRange -> individualDeletedMessages.add( - Range.openClosed( - new PositionImpl(messageRange.getLowerEndpoint()), - new PositionImpl(messageRange.getUpperEndpoint()) - ) - )); + individualDeletedMessagesList.forEach(messageRange -> individualDeletedMessages + .add(Range.openClosed(new PositionImpl(messageRange.getLowerEndpoint()), + new PositionImpl(messageRange.getUpperEndpoint())))); } finally { lock.writeLock().unlock(); } } - private void recoveredCursor(PositionImpl position) { + private void recoveredCursor(PositionImpl position, Map properties) { // 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 if (!ledger.ledgerExists(position.getLedgerId())) { @@ -307,11 +336,12 @@ private void recoveredCursor(PositionImpl position) { messagesConsumedCounter = -getNumberOfEntries(Range.openClosed(position, ledger.getLastPosition())); markDeletePosition = position; readPosition = ledger.getNextValidPosition(position); + lastMarkDeleteEntry = new PendingMarkDeleteEntry(markDeletePosition, properties, null, null); STATE_UPDATER.set(this, State.NoLedger); } void initialize(PositionImpl position, final VoidCallback callback) { - recoveredCursor(position); + recoveredCursor(position, Collections.emptyMap()); if (log.isDebugEnabled()) { log.debug("[{}] Consumer {} cursor initialized with counters: consumed {} mdPos {} rdPos {}", ledger.getName(), name, messagesConsumedCounter, markDeletePosition, readPosition); @@ -601,8 +631,8 @@ public long getNumberOfEntriesSinceFirstNotAckedMessage() { public long getNumberOfEntriesInBacklog() { if (log.isDebugEnabled()) { log.debug("[{}] Consumer {} cursor ml-entries: {} -- deleted-counter: {} other counters: mdPos {} rdPos {}", - ledger.getName(), name, ManagedLedgerImpl.ENTRIES_ADDED_COUNTER_UPDATER.get(ledger), messagesConsumedCounter, - markDeletePosition, readPosition); + ledger.getName(), name, ManagedLedgerImpl.ENTRIES_ADDED_COUNTER_UPDATER.get(ledger), + messagesConsumedCounter, markDeletePosition, readPosition); } long backlog = ManagedLedgerImpl.ENTRIES_ADDED_COUNTER_UPDATER.get(ledger) - messagesConsumedCounter; if (backlog < 0) { @@ -649,7 +679,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; @@ -728,6 +758,8 @@ public void operationComplete() { Range.closedOpen(markDeletePosition, newMarkDeletePosition)); } markDeletePosition = newMarkDeletePosition; + lastMarkDeleteEntry = new PendingMarkDeleteEntry(newMarkDeletePosition, Collections.emptyMap(), + null, null); individualDeletedMessages.clear(); PositionImpl oldReadPosition = readPosition; @@ -767,7 +799,7 @@ public void operationFailed(ManagedLedgerException exception) { }; - internalAsyncMarkDelete(newPosition, new MarkDeleteCallback() { + internalAsyncMarkDelete(newPosition, Collections.emptyMap(), new MarkDeleteCallback() { @Override public void markDeleteComplete(Object ctx) { finalCallback.operationComplete(); @@ -868,15 +900,14 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { } /** - * Async replays given positions: - * a. before reading it filters out already-acked messages - * b. reads remaining entries async and gives it to given ReadEntriesCallback - * c. returns all already-acked messages which are not replayed so, those messages can be removed by - * caller(Dispatcher)'s replay-list and it won't try to replay it again + * Async replays given positions: a. before reading it filters out already-acked messages b. reads remaining entries + * async and gives it to given ReadEntriesCallback c. returns all already-acked messages which are not replayed so, + * those messages can be removed by caller(Dispatcher)'s replay-list and it won't try to replay it again * */ @Override - public Set asyncReplayEntries(final Set positions, ReadEntriesCallback callback, Object ctx) { + public Set asyncReplayEntries(final Set positions, + ReadEntriesCallback callback, Object ctx) { List entries = Lists.newArrayListWithExpectedSize(positions.size()); if (positions.isEmpty()) { callback.readEntriesComplete(entries, ctx); @@ -886,8 +917,10 @@ public Set asyncReplayEntries(final Set Set alreadyAcknowledgedPositions = Sets.newHashSet(); lock.readLock().lock(); try { - positions.stream().filter(position -> individualDeletedMessages.contains((PositionImpl) position) - || ((PositionImpl) position).compareTo(markDeletePosition) < 0).forEach(alreadyAcknowledgedPositions::add); + positions.stream() + .filter(position -> individualDeletedMessages.contains((PositionImpl) position) + || ((PositionImpl) position).compareTo(markDeletePosition) < 0) + .forEach(alreadyAcknowledgedPositions::add); } finally { lock.readLock().unlock(); } @@ -927,8 +960,7 @@ public synchronized void readEntryFailed(ManagedLedgerException mle, Object ctx) } }; - positions.stream() - .filter(position -> !alreadyAcknowledgedPositions.contains(position)) + positions.stream().filter(position -> !alreadyAcknowledgedPositions.contains(position)) .forEach(p -> ledger.asyncReadEntry((PositionImpl) p, cb, ctx)); return alreadyAcknowledgedPositions; @@ -968,6 +1000,12 @@ private long getNumberOfEntries(Range range) { @Override public void markDelete(Position position) throws InterruptedException, ManagedLedgerException { + markDelete(position, Collections.emptyMap()); + } + + @Override + public void markDelete(Position position, Map properties) + throws InterruptedException, ManagedLedgerException { checkNotNull(position); checkArgument(position instanceof PositionImpl); @@ -978,7 +1016,7 @@ class Result { final Result result = new Result(); final CountDownLatch counter = new CountDownLatch(1); - asyncMarkDelete(position, new MarkDeleteCallback() { + asyncMarkDelete(position, properties, new MarkDeleteCallback() { @Override public void markDeleteComplete(Object ctx) { counter.countDown(); @@ -1221,6 +1259,12 @@ PositionImpl setAcknowledgedPosition(PositionImpl newMarkDeletePosition) { @Override public void asyncMarkDelete(final Position position, final MarkDeleteCallback callback, final Object ctx) { + asyncMarkDelete(position, Collections.emptyMap(), callback, ctx); + } + + @Override + public void asyncMarkDelete(final Position position, Map properties, + final MarkDeleteCallback callback, final Object ctx) { checkNotNull(position); checkArgument(position instanceof PositionImpl); @@ -1257,17 +1301,18 @@ public void asyncMarkDelete(final Position position, final MarkDeleteCallback ca // Apply rate limiting to mark-delete operations if (markDeleteLimiter != null && !markDeleteLimiter.tryAcquire()) { + lastMarkDeleteEntry = new PendingMarkDeleteEntry(newPosition, properties, null, null); callback.markDeleteComplete(ctx); return; } - internalAsyncMarkDelete(newPosition, callback, ctx); + internalAsyncMarkDelete(newPosition, properties, callback, ctx); } - protected void internalAsyncMarkDelete(final PositionImpl newPosition, final MarkDeleteCallback callback, - final Object ctx) { + protected void internalAsyncMarkDelete(final PositionImpl newPosition, Map properties, + final MarkDeleteCallback callback, final Object ctx) { ledger.mbean.addMarkDeleteOp(); - PendingMarkDeleteEntry mdEntry = new PendingMarkDeleteEntry(newPosition, callback, ctx); + PendingMarkDeleteEntry mdEntry = new PendingMarkDeleteEntry(newPosition, properties, callback, ctx); // We cannot write to the ledger during the switch, need to wait until the new metadata ledger is available synchronized (pendingMarkDeleteOps) { @@ -1309,7 +1354,9 @@ void internalMarkDelete(final PendingMarkDeleteEntry mdEntry) { // ledger is postponed to when the counter goes to 0. PENDING_MARK_DELETED_SUBMITTED_COUNT_UPDATER.incrementAndGet(this); - persistPosition(cursorLedger, mdEntry.newPosition, new VoidCallback() { + lastMarkDeleteEntry = mdEntry; + + persistPosition(cursorLedger, mdEntry, new VoidCallback() { @Override public void operationComplete() { if (log.isDebugEnabled()) { @@ -1498,12 +1545,13 @@ public void asyncDelete(Position pos, final AsyncCallbacks.DeleteCallback callba // Apply rate limiting to mark-delete operations if (markDeleteLimiter != null && !markDeleteLimiter.tryAcquire()) { + lastMarkDeleteEntry = new PendingMarkDeleteEntry(newMarkDeletePosition, Collections.emptyMap(), null, null); callback.deleteComplete(ctx); return; } try { - internalAsyncMarkDelete(newMarkDeletePosition, new MarkDeleteCallback() { + internalAsyncMarkDelete(newMarkDeletePosition, Collections.emptyMap(), new MarkDeleteCallback() { @Override public void markDeleteComplete(Object ctx) { callback.deleteComplete(ctx); @@ -1663,7 +1711,7 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) { } } - private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl position, + private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl position, Map properties, MetaStoreCallback callback) { // 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. @@ -1673,9 +1721,10 @@ private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl positio .setMarkDeleteEntryId(position.getEntryId()); // info.addAllIndividualDeletedMessages(buildIndividualDeletedMessageRanges()); + info.addAllProperties(buildPropertiesMap(properties)); if (log.isDebugEnabled()) { - log.debug("[{}][{}] Closing cursor at md-position: {}", ledger.getName(), name, markDeletePosition); + log.debug("[{}][{}] Closing cursor at md-position: {}", ledger.getName(), name, position); } ledger.getStore().asyncUpdateCursorInfo(ledger.getName(), name, info.build(), cursorLedgerStat, @@ -1701,25 +1750,26 @@ public void asyncClose(final AsyncCallbacks.CloseCallback callback, final Object return; } - persistPositionMetaStore(-1, markDeletePosition, new MetaStoreCallback() { - @Override - public void operationComplete(Void result, Stat stat) { - log.info("[{}][{}] Closed cursor at md-position={}", ledger.getName(), name, - markDeletePosition); + persistPositionMetaStore(-1, lastMarkDeleteEntry.newPosition, lastMarkDeleteEntry.properties, + new MetaStoreCallback() { + @Override + public void operationComplete(Void result, Stat stat) { + log.info("[{}][{}] Closed cursor at md-position={}", ledger.getName(), name, + markDeletePosition); - // At this point the position had already been safely stored in the cursor z-node - callback.closeComplete(ctx); + // At this point the position had already been safely stored in the cursor z-node + callback.closeComplete(ctx); - asyncDeleteLedger(cursorLedger); - } + asyncDeleteLedger(cursorLedger); + } - @Override - public void operationFailed(MetaStoreException e) { - log.warn("[{}][{}] Failed to update cursor info when closing: {}", ledger.getName(), name, - e.getMessage()); - callback.closeFailed(e, ctx); - } - }); + @Override + public void operationFailed(MetaStoreException e) { + log.warn("[{}][{}] Failed to update cursor info when closing: {}", ledger.getName(), name, + e.getMessage()); + callback.closeFailed(e, ctx); + } + }); } /** @@ -1811,13 +1861,13 @@ void createNewMetadataLedger(final VoidCallback callback) { } // Created the ledger, now write the last position // content - final PositionImpl position = (PositionImpl) getMarkDeletedPosition(); - persistPosition(lh, position, new VoidCallback() { + PendingMarkDeleteEntry mdEntry = lastMarkDeleteEntry; + persistPosition(lh, mdEntry, new VoidCallback() { @Override public void operationComplete() { if (log.isDebugEnabled()) { - log.debug("[{}] Persisted position {} for cursor {}", ledger.getName(), position, - name); + log.debug("[{}] Persisted position {} for cursor {}", ledger.getName(), + mdEntry.newPosition, name); } switchToNewLedger(lh, new VoidCallback() { @Override @@ -1842,8 +1892,8 @@ public void operationFailed(ManagedLedgerException exception) { @Override public void operationFailed(ManagedLedgerException exception) { - log.warn("[{}] Failed to persist position {} for cursor {}", ledger.getName(), position, - name); + log.warn("[{}] Failed to persist position {} for cursor {}", ledger.getName(), + mdEntry.newPosition, name); ledger.mbean.startCursorLedgerDeleteOp(); bookkeeper.asyncDeleteLedger(lh.getId(), new DeleteCallback() { @@ -1859,6 +1909,20 @@ public void deleteComplete(int rc, Object ctx) { }, null); } + private List buildPropertiesMap(Map properties) { + if (properties.isEmpty()) { + return Collections.emptyList(); + } + + List longProperties = Lists.newArrayList(); + properties.forEach((name, value) -> { + LongProperty lp = LongProperty.newBuilder().setName(name).setValue(value).build(); + longProperties.add(lp); + }); + + return longProperties; + } + private List buildIndividualDeletedMessageRanges() { lock.readLock().lock(); try { @@ -1866,10 +1930,10 @@ private List buildIndividualDeletedMessageRanges() { return Collections.emptyList(); } - MLDataFormats.NestedPositionInfo.Builder nestedPositionBuilder = MLDataFormats.NestedPositionInfo.newBuilder(); + MLDataFormats.NestedPositionInfo.Builder nestedPositionBuilder = MLDataFormats.NestedPositionInfo + .newBuilder(); MLDataFormats.MessageRange.Builder messageRangeBuilder = MLDataFormats.MessageRange.newBuilder(); - return individualDeletedMessages.asRanges().stream() - .limit(config.getMaxUnackedRangesToPersist()) + return individualDeletedMessages.asRanges().stream().limit(config.getMaxUnackedRangesToPersist()) .map(positionRange -> { PositionImpl p = positionRange.lowerEndpoint(); nestedPositionBuilder.setLedgerId(p.getLedgerId()); @@ -1880,18 +1944,20 @@ private List buildIndividualDeletedMessageRanges() { nestedPositionBuilder.setEntryId(p.getEntryId()); messageRangeBuilder.setUpperEndpoint(nestedPositionBuilder.build()); return messageRangeBuilder.build(); - }) - .collect(Collectors.toList()); + }).collect(Collectors.toList()); } finally { lock.readLock().unlock(); } } - void persistPosition(final LedgerHandle lh, final PositionImpl position, final VoidCallback callback) { - PositionInfo pi = PositionInfo.newBuilder() - .setLedgerId(position.getLedgerId()) + void persistPosition(final LedgerHandle lh, PendingMarkDeleteEntry mdEntry, final VoidCallback callback) { + PositionImpl position = mdEntry.newPosition; + PositionInfo pi = PositionInfo.newBuilder().setLedgerId(position.getLedgerId()) .setEntryId(position.getEntryId()) - .addAllIndividualDeletedMessages(buildIndividualDeletedMessageRanges()).build(); + .addAllIndividualDeletedMessages(buildIndividualDeletedMessageRanges()) + .addAllProperties(buildPropertiesMap(mdEntry.properties)).build(); + + if (log.isDebugEnabled()) { log.debug("[{}] Cursor {} Appending to ledger={} position={}", ledger.getName(), name, lh.getId(), position); @@ -1901,14 +1967,13 @@ void persistPosition(final LedgerHandle lh, final PositionImpl position, final V 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()); + log.debug("[{}] Updated cursor {} position {} in meta-ledger {}", ledger.getName(), name, position, + lh1.getId()); } if (shouldCloseLedger(lh1)) { if (log.isDebugEnabled()) { - log.debug("[{}] Need to create new metadata ledger for consumer {}", ledger.getName(), - name); + log.debug("[{}] Need to create new metadata ledger for consumer {}", ledger.getName(), name); } startCreatingNewMetadataLedger(); } @@ -1943,11 +2008,12 @@ void switchToNewLedger(final LedgerHandle lh, final VoidCallback callback) { if (log.isDebugEnabled()) { log.debug("[{}] Switching cursor {} to ledger {}", ledger.getName(), name, lh.getId()); } - persistPositionMetaStore(lh.getId(), markDeletePosition, new MetaStoreCallback() { + persistPositionMetaStore(lh.getId(), lastMarkDeleteEntry.newPosition, lastMarkDeleteEntry.properties, + new MetaStoreCallback() { @Override public void operationComplete(Void result, Stat stat) { - log.info("[{}] Updated cursor {} with ledger id {} md-position={} rd-position={}", - ledger.getName(), name, lh.getId(), markDeletePosition, readPosition); + log.info("[{}] Updated cursor {} with ledger id {} md-position={} rd-position={}", ledger.getName(), + name, lh.getId(), markDeletePosition, readPosition); final LedgerHandle oldLedger = cursorLedger; cursorLedger = lh; cursorLedgerStat = stat; diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java index 2a1f305d35dbe..4ff9077704038 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java @@ -58,6 +58,7 @@ import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback; import org.apache.bookkeeper.mledger.impl.MetaStore.Stat; import org.apache.bookkeeper.mledger.proto.MLDataFormats; +import org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty; import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo; import org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange; import org.apache.bookkeeper.mledger.util.Futures; @@ -397,6 +398,14 @@ public void operationComplete(ManagedCursorInfo pbCursorInfo, Stat stat) { cursorInfo.markDelete.entryId = pbCursorInfo.getMarkDeleteEntryId(); } + if (pbCursorInfo.getPropertiesCount() > 0) { + cursorInfo.properties = Maps.newTreeMap(); + for (int i = 0; i < pbCursorInfo.getPropertiesCount(); i++) { + LongProperty property = pbCursorInfo.getProperties(i); + cursorInfo.properties.put(property.getName(), property.getValue()); + } + } + if (pbCursorInfo.getIndividualDeletedMessagesCount() > 0) { cursorInfo.individualDeletedMessages = new ArrayList<>(); for (int i = 0; i < pbCursorInfo diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java index 74a208e4a5cd8..8cd4020aca09b 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java @@ -18,6 +18,8 @@ */ package org.apache.bookkeeper.mledger.impl; +import java.util.Map; + import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.mledger.AsyncCallbacks.CloseCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCursorCallback; @@ -80,8 +82,8 @@ void recover(final VoidCallback callback) { } @Override - protected void internalAsyncMarkDelete(final PositionImpl newPosition, final MarkDeleteCallback callback, - final Object ctx) { + protected void internalAsyncMarkDelete(final PositionImpl newPosition, Map properties, + final MarkDeleteCallback callback, final Object ctx) { // Bypass persistence of mark-delete position and individually deleted messages info callback.markDeleteComplete(ctx); } 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 541b83093f600..3768c193f71fc 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 @@ -1266,6 +1266,16 @@ public interface PositionInfoOrBuilder getIndividualDeletedMessagesOrBuilderList(); org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder getIndividualDeletedMessagesOrBuilder( int index); + + // repeated .LongProperty properties = 4; + java.util.List + getPropertiesList(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty getProperties(int index); + int getPropertiesCount(); + java.util.List + getPropertiesOrBuilderList(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongPropertyOrBuilder getPropertiesOrBuilder( + int index); } public static final class PositionInfo extends com.google.protobuf.GeneratedMessage @@ -1337,10 +1347,32 @@ public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRangeOrBuilder g return individualDeletedMessages_.get(index); } + // repeated .LongProperty properties = 4; + public static final int PROPERTIES_FIELD_NUMBER = 4; + private java.util.List properties_; + public java.util.List getPropertiesList() { + return properties_; + } + public java.util.List + getPropertiesOrBuilderList() { + return properties_; + } + public int getPropertiesCount() { + return properties_.size(); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty getProperties(int index) { + return properties_.get(index); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongPropertyOrBuilder getPropertiesOrBuilder( + int index) { + return properties_.get(index); + } + private void initFields() { ledgerId_ = 0L; entryId_ = 0L; individualDeletedMessages_ = java.util.Collections.emptyList(); + properties_ = java.util.Collections.emptyList(); } private byte memoizedIsInitialized = -1; public final boolean isInitialized() { @@ -1361,6 +1393,12 @@ public final boolean isInitialized() { return false; } } + for (int i = 0; i < getPropertiesCount(); i++) { + if (!getProperties(i).isInitialized()) { + memoizedIsInitialized = 0; + return false; + } + } memoizedIsInitialized = 1; return true; } @@ -1377,6 +1415,9 @@ public void writeTo(com.google.protobuf.CodedOutputStream output) for (int i = 0; i < individualDeletedMessages_.size(); i++) { output.writeMessage(3, individualDeletedMessages_.get(i)); } + for (int i = 0; i < properties_.size(); i++) { + output.writeMessage(4, properties_.get(i)); + } getUnknownFields().writeTo(output); } @@ -1398,6 +1439,10 @@ public int getSerializedSize() { size += com.google.protobuf.CodedOutputStream .computeMessageSize(3, individualDeletedMessages_.get(i)); } + for (int i = 0; i < properties_.size(); i++) { + size += com.google.protobuf.CodedOutputStream + .computeMessageSize(4, properties_.get(i)); + } size += getUnknownFields().getSerializedSize(); memoizedSerializedSize = size; return size; @@ -1515,6 +1560,7 @@ private Builder(BuilderParent parent) { private void maybeForceBuilderInitialization() { if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) { getIndividualDeletedMessagesFieldBuilder(); + getPropertiesFieldBuilder(); } } private static Builder create() { @@ -1533,6 +1579,12 @@ public Builder clear() { } else { individualDeletedMessagesBuilder_.clear(); } + if (propertiesBuilder_ == null) { + properties_ = java.util.Collections.emptyList(); + bitField0_ = (bitField0_ & ~0x00000008); + } else { + propertiesBuilder_.clear(); + } return this; } @@ -1588,6 +1640,15 @@ public org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo buildParti } else { result.individualDeletedMessages_ = individualDeletedMessagesBuilder_.build(); } + if (propertiesBuilder_ == null) { + if (((bitField0_ & 0x00000008) == 0x00000008)) { + properties_ = java.util.Collections.unmodifiableList(properties_); + bitField0_ = (bitField0_ & ~0x00000008); + } + result.properties_ = properties_; + } else { + result.properties_ = propertiesBuilder_.build(); + } result.bitField0_ = to_bitField0_; onBuilt(); return result; @@ -1636,6 +1697,32 @@ public Builder mergeFrom(org.apache.bookkeeper.mledger.proto.MLDataFormats.Posit } } } + if (propertiesBuilder_ == null) { + if (!other.properties_.isEmpty()) { + if (properties_.isEmpty()) { + properties_ = other.properties_; + bitField0_ = (bitField0_ & ~0x00000008); + } else { + ensurePropertiesIsMutable(); + properties_.addAll(other.properties_); + } + onChanged(); + } + } else { + if (!other.properties_.isEmpty()) { + if (propertiesBuilder_.isEmpty()) { + propertiesBuilder_.dispose(); + propertiesBuilder_ = null; + properties_ = other.properties_; + bitField0_ = (bitField0_ & ~0x00000008); + propertiesBuilder_ = + com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders ? + getPropertiesFieldBuilder() : null; + } else { + propertiesBuilder_.addAllMessages(other.properties_); + } + } + } this.mergeUnknownFields(other.getUnknownFields()); return this; } @@ -1655,6 +1742,12 @@ public final boolean isInitialized() { return false; } } + for (int i = 0; i < getPropertiesCount(); i++) { + if (!getProperties(i).isInitialized()) { + + return false; + } + } return true; } @@ -1697,6 +1790,12 @@ public Builder mergeFrom( addIndividualDeletedMessages(subBuilder.buildPartial()); break; } + case 34: { + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder subBuilder = org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.newBuilder(); + input.readMessage(subBuilder, extensionRegistry); + addProperties(subBuilder.buildPartial()); + break; + } } } } @@ -1931,6 +2030,192 @@ public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder ad return individualDeletedMessagesBuilder_; } + // repeated .LongProperty properties = 4; + private java.util.List properties_ = + java.util.Collections.emptyList(); + private void ensurePropertiesIsMutable() { + if (!((bitField0_ & 0x00000008) == 0x00000008)) { + properties_ = new java.util.ArrayList(properties_); + bitField0_ |= 0x00000008; + } + } + + private com.google.protobuf.RepeatedFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongPropertyOrBuilder> propertiesBuilder_; + + public java.util.List getPropertiesList() { + if (propertiesBuilder_ == null) { + return java.util.Collections.unmodifiableList(properties_); + } else { + return propertiesBuilder_.getMessageList(); + } + } + public int getPropertiesCount() { + if (propertiesBuilder_ == null) { + return properties_.size(); + } else { + return propertiesBuilder_.getCount(); + } + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty getProperties(int index) { + if (propertiesBuilder_ == null) { + return properties_.get(index); + } else { + return propertiesBuilder_.getMessage(index); + } + } + public Builder setProperties( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty value) { + if (propertiesBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + ensurePropertiesIsMutable(); + properties_.set(index, value); + onChanged(); + } else { + propertiesBuilder_.setMessage(index, value); + } + return this; + } + public Builder setProperties( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder builderForValue) { + if (propertiesBuilder_ == null) { + ensurePropertiesIsMutable(); + properties_.set(index, builderForValue.build()); + onChanged(); + } else { + propertiesBuilder_.setMessage(index, builderForValue.build()); + } + return this; + } + public Builder addProperties(org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty value) { + if (propertiesBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + ensurePropertiesIsMutable(); + properties_.add(value); + onChanged(); + } else { + propertiesBuilder_.addMessage(value); + } + return this; + } + public Builder addProperties( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty value) { + if (propertiesBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + ensurePropertiesIsMutable(); + properties_.add(index, value); + onChanged(); + } else { + propertiesBuilder_.addMessage(index, value); + } + return this; + } + public Builder addProperties( + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder builderForValue) { + if (propertiesBuilder_ == null) { + ensurePropertiesIsMutable(); + properties_.add(builderForValue.build()); + onChanged(); + } else { + propertiesBuilder_.addMessage(builderForValue.build()); + } + return this; + } + public Builder addProperties( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder builderForValue) { + if (propertiesBuilder_ == null) { + ensurePropertiesIsMutable(); + properties_.add(index, builderForValue.build()); + onChanged(); + } else { + propertiesBuilder_.addMessage(index, builderForValue.build()); + } + return this; + } + public Builder addAllProperties( + java.lang.Iterable values) { + if (propertiesBuilder_ == null) { + ensurePropertiesIsMutable(); + super.addAll(values, properties_); + onChanged(); + } else { + propertiesBuilder_.addAllMessages(values); + } + return this; + } + public Builder clearProperties() { + if (propertiesBuilder_ == null) { + properties_ = java.util.Collections.emptyList(); + bitField0_ = (bitField0_ & ~0x00000008); + onChanged(); + } else { + propertiesBuilder_.clear(); + } + return this; + } + public Builder removeProperties(int index) { + if (propertiesBuilder_ == null) { + ensurePropertiesIsMutable(); + properties_.remove(index); + onChanged(); + } else { + propertiesBuilder_.remove(index); + } + return this; + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder getPropertiesBuilder( + int index) { + return getPropertiesFieldBuilder().getBuilder(index); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongPropertyOrBuilder getPropertiesOrBuilder( + int index) { + if (propertiesBuilder_ == null) { + return properties_.get(index); } else { + return propertiesBuilder_.getMessageOrBuilder(index); + } + } + public java.util.List + getPropertiesOrBuilderList() { + if (propertiesBuilder_ != null) { + return propertiesBuilder_.getMessageOrBuilderList(); + } else { + return java.util.Collections.unmodifiableList(properties_); + } + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder addPropertiesBuilder() { + return getPropertiesFieldBuilder().addBuilder( + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.getDefaultInstance()); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder addPropertiesBuilder( + int index) { + return getPropertiesFieldBuilder().addBuilder( + index, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.getDefaultInstance()); + } + public java.util.List + getPropertiesBuilderList() { + return getPropertiesFieldBuilder().getBuilderList(); + } + private com.google.protobuf.RepeatedFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongPropertyOrBuilder> + getPropertiesFieldBuilder() { + if (propertiesBuilder_ == null) { + propertiesBuilder_ = new com.google.protobuf.RepeatedFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongPropertyOrBuilder>( + properties_, + ((bitField0_ & 0x00000008) == 0x00000008), + getParentForChildren(), + isClean()); + properties_ = null; + } + return propertiesBuilder_; + } + // @@protoc_insertion_point(builder_scope:PositionInfo) } @@ -2952,131 +3237,104 @@ public org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfoOrBui // @@protoc_insertion_point(class_scope:MessageRange) } - public interface ManagedCursorInfoOrBuilder + public interface LongPropertyOrBuilder 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(); + // required string name = 1; + boolean hasName(); + String getName(); - // 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); + // required int64 value = 2; + boolean hasValue(); + long getValue(); } - public static final class ManagedCursorInfo extends + public static final class LongProperty extends com.google.protobuf.GeneratedMessage - implements ManagedCursorInfoOrBuilder { - // Use ManagedCursorInfo.newBuilder() to construct. - private ManagedCursorInfo(Builder builder) { + implements LongPropertyOrBuilder { + // Use LongProperty.newBuilder() to construct. + private LongProperty(Builder builder) { super(builder); } - private ManagedCursorInfo(boolean noInit) {} + private LongProperty(boolean noInit) {} - private static final ManagedCursorInfo defaultInstance; - public static ManagedCursorInfo getDefaultInstance() { + private static final LongProperty defaultInstance; + public static LongProperty getDefaultInstance() { return defaultInstance; } - public ManagedCursorInfo getDefaultInstanceForType() { + public LongProperty 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_LongProperty_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_LongProperty_fieldAccessorTable; } private int bitField0_; - // required int64 cursorsLedgerId = 1; - public static final int CURSORSLEDGERID_FIELD_NUMBER = 1; - private long cursorsLedgerId_; - public boolean hasCursorsLedgerId() { + // required string name = 1; + public static final int NAME_FIELD_NUMBER = 1; + private java.lang.Object name_; + public boolean hasName() { return ((bitField0_ & 0x00000001) == 0x00000001); } - public long getCursorsLedgerId() { - return cursorsLedgerId_; + public String getName() { + java.lang.Object ref = name_; + if (ref instanceof String) { + return (String) ref; + } else { + com.google.protobuf.ByteString bs = + (com.google.protobuf.ByteString) ref; + String s = bs.toStringUtf8(); + if (com.google.protobuf.Internal.isValidUtf8(bs)) { + name_ = s; + } + return s; + } + } + private com.google.protobuf.ByteString getNameBytes() { + java.lang.Object ref = name_; + if (ref instanceof String) { + com.google.protobuf.ByteString b = + com.google.protobuf.ByteString.copyFromUtf8((String) ref); + name_ = b; + return b; + } else { + return (com.google.protobuf.ByteString) ref; + } } - // optional int64 markDeleteLedgerId = 2; - public static final int MARKDELETELEDGERID_FIELD_NUMBER = 2; - private long markDeleteLedgerId_; - public boolean hasMarkDeleteLedgerId() { + // required int64 value = 2; + public static final int VALUE_FIELD_NUMBER = 2; + private long value_; + public boolean hasValue() { 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); + public long getValue() { + return value_; } private void initFields() { - cursorsLedgerId_ = 0L; - markDeleteLedgerId_ = 0L; - markDeleteEntryId_ = 0L; - individualDeletedMessages_ = java.util.Collections.emptyList(); + name_ = ""; + value_ = 0L; } private byte memoizedIsInitialized = -1; public final boolean isInitialized() { byte isInitialized = memoizedIsInitialized; if (isInitialized != -1) return isInitialized == 1; - if (!hasCursorsLedgerId()) { + if (!hasName()) { memoizedIsInitialized = 0; return false; } - for (int i = 0; i < getIndividualDeletedMessagesCount(); i++) { - if (!getIndividualDeletedMessages(i).isInitialized()) { - memoizedIsInitialized = 0; - return false; - } + if (!hasValue()) { + memoizedIsInitialized = 0; + return false; } memoizedIsInitialized = 1; return true; @@ -3086,16 +3344,10 @@ public void writeTo(com.google.protobuf.CodedOutputStream output) throws java.io.IOException { getSerializedSize(); if (((bitField0_ & 0x00000001) == 0x00000001)) { - output.writeInt64(1, cursorsLedgerId_); + output.writeBytes(1, getNameBytes()); } 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)); + output.writeInt64(2, value_); } getUnknownFields().writeTo(output); } @@ -3108,19 +3360,11 @@ public int getSerializedSize() { size = 0; if (((bitField0_ & 0x00000001) == 0x00000001)) { size += com.google.protobuf.CodedOutputStream - .computeInt64Size(1, cursorsLedgerId_); + .computeBytesSize(1, getNameBytes()); } 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)); + .computeInt64Size(2, value_); } size += getUnknownFields().getSerializedSize(); memoizedSerializedSize = size; @@ -3134,41 +3378,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.LongProperty 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.LongProperty 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.LongProperty 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.LongProperty 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.LongProperty 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.LongProperty 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.LongProperty parseDelimitedFrom(java.io.InputStream input) throws java.io.IOException { Builder builder = newBuilder(); if (builder.mergeDelimitedFrom(input)) { @@ -3177,7 +3421,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.LongProperty parseDelimitedFrom( java.io.InputStream input, com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { @@ -3188,12 +3432,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.LongProperty 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.LongProperty parseFrom( com.google.protobuf.CodedInputStream input, com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { @@ -3203,7 +3447,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.LongProperty prototype) { return newBuilder().mergeFrom(prototype); } public Builder toBuilder() { return newBuilder(this); } @@ -3216,15 +3460,549 @@ 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.LongPropertyOrBuilder { 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_LongProperty_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_LongProperty_fieldAccessorTable; + } + + // Construct using org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.newBuilder() + private Builder() { + maybeForceBuilderInitialization(); + } + + private Builder(BuilderParent parent) { + super(parent); + maybeForceBuilderInitialization(); + } + private void maybeForceBuilderInitialization() { + if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) { + } + } + private static Builder create() { + return new Builder(); + } + + public Builder clear() { + super.clear(); + name_ = ""; + bitField0_ = (bitField0_ & ~0x00000001); + value_ = 0L; + 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.LongProperty.getDescriptor(); + } + + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty getDefaultInstanceForType() { + return org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.getDefaultInstance(); + } + + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty build() { + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty result = buildPartial(); + if (!result.isInitialized()) { + throw newUninitializedMessageException(result); + } + return result; + } + + private org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty buildParsed() + throws com.google.protobuf.InvalidProtocolBufferException { + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty result = buildPartial(); + if (!result.isInitialized()) { + throw newUninitializedMessageException( + result).asInvalidProtocolBufferException(); + } + return result; + } + + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty buildPartial() { + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty result = new org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty(this); + int from_bitField0_ = bitField0_; + int to_bitField0_ = 0; + if (((from_bitField0_ & 0x00000001) == 0x00000001)) { + to_bitField0_ |= 0x00000001; + } + result.name_ = name_; + if (((from_bitField0_ & 0x00000002) == 0x00000002)) { + to_bitField0_ |= 0x00000002; + } + result.value_ = value_; + result.bitField0_ = to_bitField0_; + onBuilt(); + return result; + } + + public Builder mergeFrom(com.google.protobuf.Message other) { + if (other instanceof org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty) { + return mergeFrom((org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty)other); + } else { + super.mergeFrom(other); + return this; + } + } + + public Builder mergeFrom(org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty other) { + if (other == org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.getDefaultInstance()) return this; + if (other.hasName()) { + setName(other.getName()); + } + if (other.hasValue()) { + setValue(other.getValue()); + } + this.mergeUnknownFields(other.getUnknownFields()); + return this; + } + + public final boolean isInitialized() { + if (!hasName()) { + + return false; + } + if (!hasValue()) { + + 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: { + bitField0_ |= 0x00000001; + name_ = input.readBytes(); + break; + } + case 16: { + bitField0_ |= 0x00000002; + value_ = input.readInt64(); + break; + } + } + } + } + + private int bitField0_; + + // required string name = 1; + private java.lang.Object name_ = ""; + public boolean hasName() { + return ((bitField0_ & 0x00000001) == 0x00000001); + } + public String getName() { + java.lang.Object ref = name_; + if (!(ref instanceof String)) { + String s = ((com.google.protobuf.ByteString) ref).toStringUtf8(); + name_ = s; + return s; + } else { + return (String) ref; + } + } + public Builder setName(String value) { + if (value == null) { + throw new NullPointerException(); + } + bitField0_ |= 0x00000001; + name_ = value; + onChanged(); + return this; + } + public Builder clearName() { + bitField0_ = (bitField0_ & ~0x00000001); + name_ = getDefaultInstance().getName(); + onChanged(); + return this; + } + void setName(com.google.protobuf.ByteString value) { + bitField0_ |= 0x00000001; + name_ = value; + onChanged(); + } + + // required int64 value = 2; + private long value_ ; + public boolean hasValue() { + return ((bitField0_ & 0x00000002) == 0x00000002); + } + public long getValue() { + return value_; + } + public Builder setValue(long value) { + bitField0_ |= 0x00000002; + value_ = value; + onChanged(); + return this; + } + public Builder clearValue() { + bitField0_ = (bitField0_ & ~0x00000002); + value_ = 0L; + onChanged(); + return this; + } + + // @@protoc_insertion_point(builder_scope:LongProperty) + } + + static { + defaultInstance = new LongProperty(true); + defaultInstance.initFields(); + } + + // @@protoc_insertion_point(class_scope:LongProperty) + } + + 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); + + // repeated .LongProperty properties = 5; + java.util.List + getPropertiesList(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty getProperties(int index); + int getPropertiesCount(); + java.util.List + getPropertiesOrBuilderList(); + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongPropertyOrBuilder getPropertiesOrBuilder( + 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); + } + + // repeated .LongProperty properties = 5; + public static final int PROPERTIES_FIELD_NUMBER = 5; + private java.util.List properties_; + public java.util.List getPropertiesList() { + return properties_; + } + public java.util.List + getPropertiesOrBuilderList() { + return properties_; + } + public int getPropertiesCount() { + return properties_.size(); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty getProperties(int index) { + return properties_.get(index); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongPropertyOrBuilder getPropertiesOrBuilder( + int index) { + return properties_.get(index); + } + + private void initFields() { + cursorsLedgerId_ = 0L; + markDeleteLedgerId_ = 0L; + markDeleteEntryId_ = 0L; + individualDeletedMessages_ = java.util.Collections.emptyList(); + properties_ = 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; + } + } + for (int i = 0; i < getPropertiesCount(); i++) { + if (!getProperties(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)); + } + for (int i = 0; i < properties_.size(); i++) { + output.writeMessage(5, properties_.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)); + } + for (int i = 0; i < properties_.size(); i++) { + size += com.google.protobuf.CodedOutputStream + .computeMessageSize(5, properties_.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() @@ -3239,6 +4017,7 @@ private Builder(BuilderParent parent) { private void maybeForceBuilderInitialization() { if (com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders) { getIndividualDeletedMessagesFieldBuilder(); + getPropertiesFieldBuilder(); } } private static Builder create() { @@ -3259,6 +4038,12 @@ public Builder clear() { } else { individualDeletedMessagesBuilder_.clear(); } + if (propertiesBuilder_ == null) { + properties_ = java.util.Collections.emptyList(); + bitField0_ = (bitField0_ & ~0x00000010); + } else { + propertiesBuilder_.clear(); + } return this; } @@ -3318,6 +4103,15 @@ public org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo build } else { result.individualDeletedMessages_ = individualDeletedMessagesBuilder_.build(); } + if (propertiesBuilder_ == null) { + if (((bitField0_ & 0x00000010) == 0x00000010)) { + properties_ = java.util.Collections.unmodifiableList(properties_); + bitField0_ = (bitField0_ & ~0x00000010); + } + result.properties_ = properties_; + } else { + result.properties_ = propertiesBuilder_.build(); + } result.bitField0_ = to_bitField0_; onBuilt(); return result; @@ -3369,6 +4163,32 @@ public Builder mergeFrom(org.apache.bookkeeper.mledger.proto.MLDataFormats.Manag } } } + if (propertiesBuilder_ == null) { + if (!other.properties_.isEmpty()) { + if (properties_.isEmpty()) { + properties_ = other.properties_; + bitField0_ = (bitField0_ & ~0x00000010); + } else { + ensurePropertiesIsMutable(); + properties_.addAll(other.properties_); + } + onChanged(); + } + } else { + if (!other.properties_.isEmpty()) { + if (propertiesBuilder_.isEmpty()) { + propertiesBuilder_.dispose(); + propertiesBuilder_ = null; + properties_ = other.properties_; + bitField0_ = (bitField0_ & ~0x00000010); + propertiesBuilder_ = + com.google.protobuf.GeneratedMessage.alwaysUseFieldBuilders ? + getPropertiesFieldBuilder() : null; + } else { + propertiesBuilder_.addAllMessages(other.properties_); + } + } + } this.mergeUnknownFields(other.getUnknownFields()); return this; } @@ -3384,6 +4204,12 @@ public final boolean isInitialized() { return false; } } + for (int i = 0; i < getPropertiesCount(); i++) { + if (!getProperties(i).isInitialized()) { + + return false; + } + } return true; } @@ -3431,6 +4257,12 @@ public Builder mergeFrom( addIndividualDeletedMessages(subBuilder.buildPartial()); break; } + case 42: { + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder subBuilder = org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.newBuilder(); + input.readMessage(subBuilder, extensionRegistry); + addProperties(subBuilder.buildPartial()); + break; + } } } } @@ -3686,6 +4518,192 @@ public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder ad return individualDeletedMessagesBuilder_; } + // repeated .LongProperty properties = 5; + private java.util.List properties_ = + java.util.Collections.emptyList(); + private void ensurePropertiesIsMutable() { + if (!((bitField0_ & 0x00000010) == 0x00000010)) { + properties_ = new java.util.ArrayList(properties_); + bitField0_ |= 0x00000010; + } + } + + private com.google.protobuf.RepeatedFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongPropertyOrBuilder> propertiesBuilder_; + + public java.util.List getPropertiesList() { + if (propertiesBuilder_ == null) { + return java.util.Collections.unmodifiableList(properties_); + } else { + return propertiesBuilder_.getMessageList(); + } + } + public int getPropertiesCount() { + if (propertiesBuilder_ == null) { + return properties_.size(); + } else { + return propertiesBuilder_.getCount(); + } + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty getProperties(int index) { + if (propertiesBuilder_ == null) { + return properties_.get(index); + } else { + return propertiesBuilder_.getMessage(index); + } + } + public Builder setProperties( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty value) { + if (propertiesBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + ensurePropertiesIsMutable(); + properties_.set(index, value); + onChanged(); + } else { + propertiesBuilder_.setMessage(index, value); + } + return this; + } + public Builder setProperties( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder builderForValue) { + if (propertiesBuilder_ == null) { + ensurePropertiesIsMutable(); + properties_.set(index, builderForValue.build()); + onChanged(); + } else { + propertiesBuilder_.setMessage(index, builderForValue.build()); + } + return this; + } + public Builder addProperties(org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty value) { + if (propertiesBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + ensurePropertiesIsMutable(); + properties_.add(value); + onChanged(); + } else { + propertiesBuilder_.addMessage(value); + } + return this; + } + public Builder addProperties( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty value) { + if (propertiesBuilder_ == null) { + if (value == null) { + throw new NullPointerException(); + } + ensurePropertiesIsMutable(); + properties_.add(index, value); + onChanged(); + } else { + propertiesBuilder_.addMessage(index, value); + } + return this; + } + public Builder addProperties( + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder builderForValue) { + if (propertiesBuilder_ == null) { + ensurePropertiesIsMutable(); + properties_.add(builderForValue.build()); + onChanged(); + } else { + propertiesBuilder_.addMessage(builderForValue.build()); + } + return this; + } + public Builder addProperties( + int index, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder builderForValue) { + if (propertiesBuilder_ == null) { + ensurePropertiesIsMutable(); + properties_.add(index, builderForValue.build()); + onChanged(); + } else { + propertiesBuilder_.addMessage(index, builderForValue.build()); + } + return this; + } + public Builder addAllProperties( + java.lang.Iterable values) { + if (propertiesBuilder_ == null) { + ensurePropertiesIsMutable(); + super.addAll(values, properties_); + onChanged(); + } else { + propertiesBuilder_.addAllMessages(values); + } + return this; + } + public Builder clearProperties() { + if (propertiesBuilder_ == null) { + properties_ = java.util.Collections.emptyList(); + bitField0_ = (bitField0_ & ~0x00000010); + onChanged(); + } else { + propertiesBuilder_.clear(); + } + return this; + } + public Builder removeProperties(int index) { + if (propertiesBuilder_ == null) { + ensurePropertiesIsMutable(); + properties_.remove(index); + onChanged(); + } else { + propertiesBuilder_.remove(index); + } + return this; + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder getPropertiesBuilder( + int index) { + return getPropertiesFieldBuilder().getBuilder(index); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongPropertyOrBuilder getPropertiesOrBuilder( + int index) { + if (propertiesBuilder_ == null) { + return properties_.get(index); } else { + return propertiesBuilder_.getMessageOrBuilder(index); + } + } + public java.util.List + getPropertiesOrBuilderList() { + if (propertiesBuilder_ != null) { + return propertiesBuilder_.getMessageOrBuilderList(); + } else { + return java.util.Collections.unmodifiableList(properties_); + } + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder addPropertiesBuilder() { + return getPropertiesFieldBuilder().addBuilder( + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.getDefaultInstance()); + } + public org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder addPropertiesBuilder( + int index) { + return getPropertiesFieldBuilder().addBuilder( + index, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.getDefaultInstance()); + } + public java.util.List + getPropertiesBuilderList() { + return getPropertiesFieldBuilder().getBuilderList(); + } + private com.google.protobuf.RepeatedFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongPropertyOrBuilder> + getPropertiesFieldBuilder() { + if (propertiesBuilder_ == null) { + propertiesBuilder_ = new com.google.protobuf.RepeatedFieldBuilder< + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder, org.apache.bookkeeper.mledger.proto.MLDataFormats.LongPropertyOrBuilder>( + properties_, + ((bitField0_ & 0x00000010) == 0x00000010), + getParentForChildren(), + isClean()); + properties_ = null; + } + return propertiesBuilder_; + } + // @@protoc_insertion_point(builder_scope:ManagedCursorInfo) } @@ -3722,6 +4740,11 @@ public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder ad private static com.google.protobuf.GeneratedMessage.FieldAccessorTable internal_static_MessageRange_fieldAccessorTable; + private static com.google.protobuf.Descriptors.Descriptor + internal_static_LongProperty_descriptor; + private static + com.google.protobuf.GeneratedMessage.FieldAccessorTable + internal_static_LongProperty_fieldAccessorTable; private static com.google.protobuf.Descriptors.Descriptor internal_static_ManagedCursorInfo_descriptor; private static @@ -3742,18 +4765,21 @@ public org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange.Builder ad "tedPosition\030\002 \001(\0132\023.NestedPositionInfo\032P" + "\n\nLedgerInfo\022\020\n\010ledgerId\030\001 \002(\003\022\017\n\007entrie" + "s\030\002 \001(\003\022\014\n\004size\030\003 \001(\003\022\021\n\ttimestamp\030\004 \001(\003" + - "\"c\n\014PositionInfo\022\020\n\010ledgerId\030\001 \002(\003\022\017\n\007en" + - "tryId\030\002 \002(\003\0220\n\031individualDeletedMessages" + - "\030\003 \003(\0132\r.MessageRange\"7\n\022NestedPositionI" + - "nfo\022\020\n\010ledgerId\030\001 \002(\003\022\017\n\007entryId\030\002 \002(\003\"f", - "\n\014MessageRange\022*\n\rlowerEndpoint\030\001 \002(\0132\023." + - "NestedPositionInfo\022*\n\rupperEndpoint\030\002 \002(" + - "\0132\023.NestedPositionInfo\"\225\001\n\021ManagedCursor" + - "Info\022\027\n\017cursorsLedgerId\030\001 \002(\003\022\032\n\022markDel" + - "eteLedgerId\030\002 \001(\003\022\031\n\021markDeleteEntryId\030\003" + - " \001(\003\0220\n\031individualDeletedMessages\030\004 \003(\0132" + - "\r.MessageRangeB\'\n#org.apache.bookkeeper." + - "mledger.protoH\001" + "\"\206\001\n\014PositionInfo\022\020\n\010ledgerId\030\001 \002(\003\022\017\n\007e" + + "ntryId\030\002 \002(\003\0220\n\031individualDeletedMessage" + + "s\030\003 \003(\0132\r.MessageRange\022!\n\nproperties\030\004 \003" + + "(\0132\r.LongProperty\"7\n\022NestedPositionInfo\022", + "\020\n\010ledgerId\030\001 \002(\003\022\017\n\007entryId\030\002 \002(\003\"f\n\014Me" + + "ssageRange\022*\n\rlowerEndpoint\030\001 \002(\0132\023.Nest" + + "edPositionInfo\022*\n\rupperEndpoint\030\002 \002(\0132\023." + + "NestedPositionInfo\"+\n\014LongProperty\022\014\n\004na" + + "me\030\001 \002(\t\022\r\n\005value\030\002 \002(\003\"\270\001\n\021ManagedCurso" + + "rInfo\022\027\n\017cursorsLedgerId\030\001 \002(\003\022\032\n\022markDe" + + "leteLedgerId\030\002 \001(\003\022\031\n\021markDeleteEntryId\030" + + "\003 \001(\003\0220\n\031individualDeletedMessages\030\004 \003(\013" + + "2\r.MessageRange\022!\n\nproperties\030\005 \003(\0132\r.Lo" + + "ngPropertyB\'\n#org.apache.bookkeeper.mled", + "ger.protoH\001" }; com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner = new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() { @@ -3781,7 +4807,7 @@ 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", "IndividualDeletedMessages", }, + new java.lang.String[] { "LedgerId", "EntryId", "IndividualDeletedMessages", "Properties", }, org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo.class, org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo.Builder.class); internal_static_NestedPositionInfo_descriptor = @@ -3800,12 +4826,20 @@ public com.google.protobuf.ExtensionRegistry assignDescriptors( 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 = + internal_static_LongProperty_descriptor = getDescriptor().getMessageTypes().get(4); + internal_static_LongProperty_fieldAccessorTable = new + com.google.protobuf.GeneratedMessage.FieldAccessorTable( + internal_static_LongProperty_descriptor, + new java.lang.String[] { "Name", "Value", }, + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.class, + org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty.Builder.class); + internal_static_ManagedCursorInfo_descriptor = + getDescriptor().getMessageTypes().get(5); internal_static_ManagedCursorInfo_fieldAccessorTable = new com.google.protobuf.GeneratedMessage.FieldAccessorTable( internal_static_ManagedCursorInfo_descriptor, - new java.lang.String[] { "CursorsLedgerId", "MarkDeleteLedgerId", "MarkDeleteEntryId", "IndividualDeletedMessages", }, + new java.lang.String[] { "CursorsLedgerId", "MarkDeleteLedgerId", "MarkDeleteEntryId", "IndividualDeletedMessages", "Properties", }, 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 37c4ba2a22beb..059b8238a36a3 100644 --- a/managed-ledger/src/main/proto/MLDataFormats.proto +++ b/managed-ledger/src/main/proto/MLDataFormats.proto @@ -42,6 +42,10 @@ message PositionInfo { required int64 ledgerId = 1; required int64 entryId = 2; repeated MessageRange individualDeletedMessages = 3; + + // Additional custom properties associated with + // the current cursor position + repeated LongProperty properties = 4; } message NestedPositionInfo { @@ -54,6 +58,12 @@ message MessageRange { required NestedPositionInfo upperEndpoint = 2; } +// Generic string and long tuple +message LongProperty { + required string name = 1; + required int64 value = 2; +} + message ManagedCursorInfo { // If the ledger id is -1, then the mark-delete position is // the one from the (ledgerId, entryId) snapshot below @@ -63,4 +73,8 @@ message ManagedCursorInfo { optional int64 markDeleteLedgerId = 2; optional int64 markDeleteEntryId = 3; repeated MessageRange individualDeletedMessages = 4; + + // Additional custom properties associated with + // the current cursor position + repeated LongProperty properties = 5; } 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 5062a1ae6e511..e04cb08900bee 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 @@ -23,7 +23,9 @@ import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; +import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Set; import org.apache.bookkeeper.mledger.AsyncCallbacks; @@ -58,6 +60,11 @@ public MockManagedCursor(ManagedCursorContainer container, String name, Position this.position = position; } + @Override + public Map getProperties() { + return Collections.emptyMap(); + } + @Override public boolean isDurable() { return true; @@ -90,6 +97,11 @@ public long getNumberOfEntriesInBacklog() { @Override public void markDelete(Position position) throws ManagedLedgerException { + markDelete(position, Collections.emptyMap()); + } + + @Override + public void markDelete(Position position, Map properties) throws ManagedLedgerException { this.position = position; container.cursorUpdated(this, (PositionImpl) position); } @@ -99,6 +111,12 @@ public void asyncMarkDelete(Position position, MarkDeleteCallback callback, Obje fail(); } + @Override + public void asyncMarkDelete(Position position, Map properties, MarkDeleteCallback callback, + Object ctx) { + fail(); + } + @Override public Position getMarkDeletedPosition() { return position; diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorPropertiesTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorPropertiesTest.java new file mode 100644 index 0000000000000..1e2400d8371d7 --- /dev/null +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorPropertiesTest.java @@ -0,0 +1,106 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.bookkeeper.mledger.impl; + +import static org.testng.Assert.assertEquals; + +import java.util.Collections; +import java.util.Map; +import java.util.TreeMap; + +import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.bookkeeper.mledger.ManagedLedger; +import org.apache.bookkeeper.mledger.ManagedLedgerConfig; +import org.apache.bookkeeper.mledger.ManagedLedgerFactory; +import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig; +import org.apache.bookkeeper.mledger.Position; +import org.apache.bookkeeper.test.MockedBookKeeperTestCase; +import org.testng.annotations.Test; + +public class ManagedCursorPropertiesTest extends MockedBookKeeperTestCase { + + @Test(timeOut = 20000) + void testPropertiesClose() throws Exception { + ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig()); + ManagedCursor c1 = ledger.openCursor("c1"); + + assertEquals(c1.getProperties(), Collections.emptyMap()); + + ledger.addEntry("entry-1".getBytes()); + ledger.addEntry("entry-2".getBytes()); + Position p3 = ledger.addEntry("entry-3".getBytes()); + + Map properties = new TreeMap<>(); + properties.put("a", 1L); + properties.put("b", 2L); + properties.put("c", 3L); + c1.markDelete(p3, properties); + + assertEquals(c1.getProperties(), properties); + + Map properties2 = new TreeMap<>(); + properties2.put("a", 4L); + properties2.put("b", 5L); + properties2.put("c", 6L); + c1.markDelete(p3, properties2); + + assertEquals(c1.getProperties(), properties2); + + ledger.close(); + + // Reopen the managed ledger + ledger = factory.open("my_test_ledger", new ManagedLedgerConfig()); + c1 = ledger.openCursor("c1"); + + assertEquals(c1.getMarkDeletedPosition(), p3); + assertEquals(c1.getProperties(), properties2); + } + + @Test(timeOut = 20000) + void testPropertiesRecoveryAfterCrash() throws Exception { + ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig()); + ManagedCursor c1 = ledger.openCursor("c1"); + + assertEquals(c1.getProperties(), Collections.emptyMap()); + + ledger.addEntry("entry-1".getBytes()); + ledger.addEntry("entry-2".getBytes()); + Position p3 = ledger.addEntry("entry-3".getBytes()); + + Map properties = new TreeMap<>(); + properties.put("a", 1L); + properties.put("b", 2L); + properties.put("c", 3L); + c1.markDelete(p3, properties); + + // Create a new factory to force a managed ledger close and recovery + ManagedLedgerFactoryConfig conf = new ManagedLedgerFactoryConfig(); + ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(bkc, zkc, conf); + + // Reopen the managed ledger + ledger = factory2.open("my_test_ledger", new ManagedLedgerConfig()); + c1 = ledger.openCursor("c1"); + + assertEquals(c1.getMarkDeletedPosition(), p3); + assertEquals(c1.getProperties(), properties); + + factory2.shutdown(); + } + +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 3394c010c0559..6aa4ed25c1b02 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -1176,6 +1176,7 @@ public PersistentTopicInternalStats getInternalStats() { cs.individuallyDeletedMessages = cursor.getIndividuallyDeletedMessages(); cs.lastLedgerSwitchTimestamp = DATE_FORMAT.format(Instant.ofEpochMilli(cursor.getLastLedgerSwitchTimestamp())); cs.state = cursor.getState(); + cs.properties = cursor.getProperties(); stats.cursors.put(cursor.getName(), cs); }); return stats; diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/PersistentTopicInternalStats.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/PersistentTopicInternalStats.java index cf84f3c0e230b..6c9b939e0dad5 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/PersistentTopicInternalStats.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/PersistentTopicInternalStats.java @@ -59,6 +59,8 @@ public static class CursorStats { public String individuallyDeletedMessages; public String lastLedgerSwitchTimestamp; public String state; + + public Map properties; } } From 3f718821b5390a79a259a3997a5fd35982176c2c Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Thu, 7 Sep 2017 16:00:04 -0700 Subject: [PATCH 2/2] Renamed PendingMarkDeleteEntry into MarkDeleteEntry --- .../mledger/impl/ManagedCursorImpl.java | 34 +++++++++---------- 1 file changed, 17 insertions(+), 17 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 04c6bbe1af70c..151f4a59d8d66 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 @@ -90,7 +90,7 @@ public class ManagedCursorImpl implements ManagedCursor { protected volatile PositionImpl markDeletePosition; protected volatile PositionImpl readPosition; - private volatile PendingMarkDeleteEntry lastMarkDeleteEntry; + private volatile MarkDeleteEntry lastMarkDeleteEntry; protected static final AtomicReferenceFieldUpdater WAITING_READ_OP_UPDATER = AtomicReferenceFieldUpdater .newUpdater(ManagedCursorImpl.class, OpReadEntry.class, "waitingReadOp"); @@ -123,7 +123,7 @@ public class ManagedCursorImpl implements ManagedCursor { private final RateLimiter markDeleteLimiter; - class PendingMarkDeleteEntry { + class MarkDeleteEntry { final PositionImpl newPosition; final MarkDeleteCallback callback; final Object ctx; @@ -132,9 +132,9 @@ class PendingMarkDeleteEntry { // If the callbackGroup is set, it means this mark-delete request was done on behalf of a group of request (just // persist the last one in the chain). In this case we need to trigger the callbacks for every request in the // group. - List callbackGroup; + List callbackGroup; - public PendingMarkDeleteEntry(PositionImpl newPosition, Map properties, + public MarkDeleteEntry(PositionImpl newPosition, Map properties, MarkDeleteCallback callback, Object ctx) { this.newPosition = PositionImpl.get(newPosition); this.properties = properties; @@ -143,7 +143,7 @@ public PendingMarkDeleteEntry(PositionImpl newPosition, Map proper } } - 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"); @SuppressWarnings("unused") @@ -336,7 +336,7 @@ private void recoveredCursor(PositionImpl position, Map properties messagesConsumedCounter = -getNumberOfEntries(Range.openClosed(position, ledger.getLastPosition())); markDeletePosition = position; readPosition = ledger.getNextValidPosition(position); - lastMarkDeleteEntry = new PendingMarkDeleteEntry(markDeletePosition, properties, null, null); + lastMarkDeleteEntry = new MarkDeleteEntry(markDeletePosition, properties, null, null); STATE_UPDATER.set(this, State.NoLedger); } @@ -758,7 +758,7 @@ public void operationComplete() { Range.closedOpen(markDeletePosition, newMarkDeletePosition)); } markDeletePosition = newMarkDeletePosition; - lastMarkDeleteEntry = new PendingMarkDeleteEntry(newMarkDeletePosition, Collections.emptyMap(), + lastMarkDeleteEntry = new MarkDeleteEntry(newMarkDeletePosition, Collections.emptyMap(), null, null); individualDeletedMessages.clear(); @@ -1301,7 +1301,7 @@ public void asyncMarkDelete(final Position position, Map propertie // Apply rate limiting to mark-delete operations if (markDeleteLimiter != null && !markDeleteLimiter.tryAcquire()) { - lastMarkDeleteEntry = new PendingMarkDeleteEntry(newPosition, properties, null, null); + lastMarkDeleteEntry = new MarkDeleteEntry(newPosition, properties, null, null); callback.markDeleteComplete(ctx); return; } @@ -1312,7 +1312,7 @@ protected void internalAsyncMarkDelete(final PositionImpl newPosition, Map buildIndividualDeletedMessageRanges() { } } - void persistPosition(final LedgerHandle lh, PendingMarkDeleteEntry mdEntry, final VoidCallback callback) { + void persistPosition(final LedgerHandle lh, MarkDeleteEntry mdEntry, final VoidCallback callback) { PositionImpl position = mdEntry.newPosition; PositionInfo pi = PositionInfo.newBuilder().setLedgerId(position.getLedgerId()) .setEntryId(position.getEntryId())