pdolif commented on code in PR #24623:
URL: https://github.com/apache/pulsar/pull/24623#discussion_r2293832748
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/Entry.java:
##########
@@ -67,6 +67,17 @@ public interface Entry {
*/
boolean release();
+ /**
+ * Managed Ledger implementations of EntryImpl should implement this
method to return the read count handler
+ * associated with the entry.
+ * This handler is used to track how many times the entry has been read
and to manage
+ * the eviction of entries from the broker cache based on their expected
read count.
+ * @return
+ */
+ default EntryReadCountHandler getReadCountHandler() {
+ return null;
+ }
Review Comment:
Is there a specific reason to have the default implementation?
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactory.java:
##########
@@ -220,17 +220,43 @@ void asyncDelete(String name,
CompletableFuture<ManagedLedgerConfig> mlConfigFut
EntryCacheManager getEntryCacheManager();
/**
- * update cache evictionTimeThreshold.
- *
- * @param cacheEvictionTimeThresholdNanos time threshold for eviction.
+ * update cache evictionTimeThreshold dynamically. Similar as
+ * {@link
ManagedLedgerFactoryConfig#setCacheEvictionTimeThresholdMillis(long)}
+ * but the value is in nanos. This inconsistency is kept for backwards
compatibility.
+ * @param cacheEvictionTimeThresholdNanos time threshold in nanos for
eviction.
*/
void updateCacheEvictionTimeThreshold(long
cacheEvictionTimeThresholdNanos);
/**
+ * time threshold for eviction. Similar as
+ * {@link
ManagedLedgerFactoryConfig#setCacheEvictionTimeThresholdMillis(long)}
Review Comment:
```suggestion
* {@link
ManagedLedgerFactoryConfig#getCacheEvictionTimeThresholdMillis()}
```
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheManagerEvictionHandler.java:
##########
@@ -64,4 +64,8 @@ public Pair<Integer, Long> evictEntries(long sizeToFree) {
manager.entriesRemoved(evictedSize, evictedEntries);
return evicted;
}
+
+ public Pair<Integer, Long> getNonEvictableSize() {
Review Comment:
Should @ VisibleForTesting be added here?
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerImpl.java:
##########
@@ -0,0 +1,403 @@
+/*
+ * 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 java.util.Objects.requireNonNull;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.locks.StampedLock;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.commons.lang3.tuple.Pair;
+
+/**
+ * Implementation of {@link ManagedCursorContainer} that contains cursors for
a ManagedLedger.
+ * This also implements {@link ActiveManagedCursorContainer} to contain active
cursors.
+ * <p>
+ * This data structure maintains a heap and a map of cursors. The map is used
to relate a cursor name with
+ * an entry index in the heap. The heap data structure sorts cursors in a
binary tree which is represented
+ * in a single array. More details about heap implementations:
+ * <a
href="https://en.wikipedia.org/wiki/Heap_(data_structure)#Implementation">here</a>
+ * <p>
+ * The heap is updated and kept sorted when a cursor is updated.
+ */
+public class ManagedCursorContainerImpl implements ManagedCursorContainer,
ActiveManagedCursorContainer {
+
+ /**
+ * This field is incremented everytime the cursor information is updated.
+ */
+ private long version;
+
+ private static class Item {
+ final ManagedCursor cursor;
+ Position position;
+ int idx;
+
+ Item(ManagedCursor cursor, Position position, int idx) {
+ this.cursor = cursor;
+ this.position = position;
+ this.idx = idx;
+ }
+ }
+
+ public ManagedCursorContainerImpl() {}
+
+ // Used to keep track of slowest cursor.
+ private final ArrayList<Item> heap = new ArrayList<>();
+
+ // Maps a cursor to its position in the heap
+ private final ConcurrentMap<String, Item> cursors = new
ConcurrentSkipListMap<>();
+
+ private final StampedLock rwLock = new StampedLock();
+
+ private int durableCursorCount;
+
+
+ /**
+ * Add a cursor to the container. The cursor will be optionally tracked
for the slowest reader when
+ * a position is passed as the second argument. It is expected that the
position is updated with
+ * {@link #cursorUpdated(ManagedCursor, Position)} method when the
position changes.
+ *
+ * @param cursor cursor to add
+ * @param position position of the cursor to use for ordering, pass null
if the cursor's position shouldn't be
+ * tracked for the slowest reader.
+ */
+ @Override
+ public void add(ManagedCursor cursor, Position position) {
+ long stamp = rwLock.writeLock();
+ try {
+ Item item = new Item(cursor, position, position != null ?
heap.size() : -1);
+ cursors.put(cursor.getName(), item);
+ if (position != null) {
+ heap.add(item);
+ if (heap.size() > 1) {
+ siftUp(item);
+ }
+ }
+ if (cursor.isDurable()) {
+ durableCursorCount++;
+ }
+ version = DataVersion.getNextVersion(version);
+ } finally {
+ rwLock.unlockWrite(stamp);
+ }
+ }
+
+ @Override
+ public ManagedCursor get(String name) {
+ long stamp = rwLock.readLock();
+ try {
+ Item item = cursors.get(name);
+ return item != null ? item.cursor : null;
+ } finally {
+ rwLock.unlockRead(stamp);
+ }
+ }
+
+ @Override
+ public boolean removeCursor(String name) {
+ long stamp = rwLock.writeLock();
+ try {
+ Item item = cursors.remove(name);
+ if (item != null) {
+ if (item.idx >= 0) {
+ if (heap.size() == 1) {
+ heap.clear();
+ } else {
+ // Move the item to the right end of the heap to be
removed
+ Item lastItem = heap.get(heap.size() - 1);
+ swap(item, lastItem);
+ heap.remove(item.idx);
+ // Update the heap
+ siftDown(lastItem);
+ }
+ }
+ if (item.cursor.isDurable()) {
+ durableCursorCount--;
+ }
+ version = DataVersion.getNextVersion(version);
+ return true;
+ } else {
+ return false;
+ }
+ } finally {
+ rwLock.unlockWrite(stamp);
+ }
+ }
+
+
+ @Override
+ public Pair<Position, Position> cursorUpdated(ManagedCursor cursor,
Position newPosition) {
+ requireNonNull(cursor);
+
+ long stamp = rwLock.writeLock();
+ try {
+ Item item = cursors.get(cursor.getName());
+ if (item == null || item.idx == -1) {
+ return null;
+ }
+
+ Position previousSlowestConsumer = heap.get(0).position;
+ item.position = newPosition;
+ version = DataVersion.getNextVersion(version);
+
+ if (heap.size() == 1) {
+ return Pair.of(previousSlowestConsumer, item.position);
+ }
+
+ // When the cursor moves forward, we need to push it toward the
+ // bottom of the tree and push it up if a reset was done
+ if (item.idx == 0 ||
getParent(item).position.compareTo(item.position) <= 0) {
+ siftDown(item);
+ } else {
+ siftUp(item);
+ }
+ Position newSlowestConsumer = heap.get(0).position;
+ return Pair.of(previousSlowestConsumer, newSlowestConsumer);
+ } finally {
+ rwLock.unlockWrite(stamp);
+ }
+ }
+
+ @Override
+ public void updateCursor(ManagedCursor cursor, Position newPosition) {
+ cursorUpdated(cursor, newPosition);
+ }
+
+ /**
+ * Get the slowest reader position for the cursors that are ordered.
+ *
+ * @return the slowest reader position
+ */
+ @Override
+ public Position getSlowestCursorPosition() {
+ long stamp = rwLock.readLock();
+ try {
+ return heap.isEmpty() ? null : heap.get(0).position;
+ } finally {
+ rwLock.unlockRead(stamp);
+ }
+ }
+
+ @Override
+ public ManagedCursor getSlowestCursor() {
+ long stamp = rwLock.readLock();
+ try {
+ return heap.isEmpty() ? null : heap.get(0).cursor;
+ } finally {
+ rwLock.unlockRead(stamp);
+ }
+ }
+
+ /**
+ * @return Returns the CursorInfo for the cursor with the oldest position,
+ * or null if there aren't any tracked cursors
+ */
+ @Override
+ public CursorInfo getCursorWithOldestPosition() {
+ long stamp = rwLock.readLock();
+ try {
+ if (heap.isEmpty()) {
+ return null;
+ } else {
+ Item item = heap.get(0);
+ return new CursorInfo(item.cursor, item.position, version);
+ }
+ } finally {
+ rwLock.unlockRead(stamp);
+ }
+ }
+
+ /**
+ * Check whether there are any cursors.
+ * @return true is there are no cursors and false if there are
+ */
+ @Override
+ public boolean isEmpty() {
+ long stamp = rwLock.tryOptimisticRead();
+ boolean isEmpty = cursors.isEmpty();
+ if (!rwLock.validate(stamp)) {
+ // Fallback to read lock
+ stamp = rwLock.readLock();
+ try {
+ isEmpty = cursors.isEmpty();
+ } finally {
+ rwLock.unlockRead(stamp);
+ }
+ }
+
+ return isEmpty;
+ }
+
+ @Override
+ public int size() {
+ long stamp = rwLock.tryOptimisticRead();
+ int size = cursors.size();
+ if (!rwLock.validate(stamp)) {
+ // Fallback to read lock
+ stamp = rwLock.readLock();
+ try {
+ size = cursors.size();
+ } finally {
+ rwLock.unlockRead(stamp);
+ }
+ }
+ return size;
+ }
+
+ /**
+ * Check whether that are any durable cursors.
Review Comment:
```suggestion
* Check whether there are any durable cursors.
```
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainer.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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 org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.commons.lang3.tuple.Pair;
+
+/**
+ * Contains cursors for a ManagedLedger that are actively being used.
+ * <p>
+ * The goal is to be able to use the container to be used for cache eviction
where the tracking of the slowest cursor
Review Comment:
```suggestion
* The goal is to be able to use the container for cache eviction where the
tracking of the slowest cursor
```
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerImpl.java:
##########
@@ -0,0 +1,403 @@
+/*
+ * 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 java.util.Objects.requireNonNull;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.locks.StampedLock;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.commons.lang3.tuple.Pair;
+
+/**
+ * Implementation of {@link ManagedCursorContainer} that contains cursors for
a ManagedLedger.
+ * This also implements {@link ActiveManagedCursorContainer} to contain active
cursors.
+ * <p>
+ * This data structure maintains a heap and a map of cursors. The map is used
to relate a cursor name with
+ * an entry index in the heap. The heap data structure sorts cursors in a
binary tree which is represented
+ * in a single array. More details about heap implementations:
+ * <a
href="https://en.wikipedia.org/wiki/Heap_(data_structure)#Implementation">here</a>
+ * <p>
+ * The heap is updated and kept sorted when a cursor is updated.
+ */
+public class ManagedCursorContainerImpl implements ManagedCursorContainer,
ActiveManagedCursorContainer {
+
+ /**
+ * This field is incremented everytime the cursor information is updated.
+ */
+ private long version;
+
+ private static class Item {
+ final ManagedCursor cursor;
+ Position position;
+ int idx;
+
+ Item(ManagedCursor cursor, Position position, int idx) {
+ this.cursor = cursor;
+ this.position = position;
+ this.idx = idx;
+ }
+ }
+
+ public ManagedCursorContainerImpl() {}
+
+ // Used to keep track of slowest cursor.
+ private final ArrayList<Item> heap = new ArrayList<>();
+
+ // Maps a cursor to its position in the heap
+ private final ConcurrentMap<String, Item> cursors = new
ConcurrentSkipListMap<>();
+
+ private final StampedLock rwLock = new StampedLock();
+
+ private int durableCursorCount;
+
+
+ /**
+ * Add a cursor to the container. The cursor will be optionally tracked
for the slowest reader when
+ * a position is passed as the second argument. It is expected that the
position is updated with
+ * {@link #cursorUpdated(ManagedCursor, Position)} method when the
position changes.
+ *
+ * @param cursor cursor to add
+ * @param position position of the cursor to use for ordering, pass null
if the cursor's position shouldn't be
+ * tracked for the slowest reader.
+ */
+ @Override
+ public void add(ManagedCursor cursor, Position position) {
+ long stamp = rwLock.writeLock();
+ try {
+ Item item = new Item(cursor, position, position != null ?
heap.size() : -1);
+ cursors.put(cursor.getName(), item);
+ if (position != null) {
+ heap.add(item);
+ if (heap.size() > 1) {
+ siftUp(item);
+ }
+ }
+ if (cursor.isDurable()) {
+ durableCursorCount++;
+ }
+ version = DataVersion.getNextVersion(version);
+ } finally {
+ rwLock.unlockWrite(stamp);
+ }
+ }
+
+ @Override
+ public ManagedCursor get(String name) {
+ long stamp = rwLock.readLock();
+ try {
+ Item item = cursors.get(name);
+ return item != null ? item.cursor : null;
+ } finally {
+ rwLock.unlockRead(stamp);
+ }
+ }
+
+ @Override
+ public boolean removeCursor(String name) {
+ long stamp = rwLock.writeLock();
+ try {
+ Item item = cursors.remove(name);
+ if (item != null) {
+ if (item.idx >= 0) {
+ if (heap.size() == 1) {
+ heap.clear();
+ } else {
+ // Move the item to the right end of the heap to be
removed
+ Item lastItem = heap.get(heap.size() - 1);
+ swap(item, lastItem);
+ heap.remove(item.idx);
+ // Update the heap
+ siftDown(lastItem);
+ }
+ }
+ if (item.cursor.isDurable()) {
+ durableCursorCount--;
+ }
+ version = DataVersion.getNextVersion(version);
+ return true;
+ } else {
+ return false;
+ }
+ } finally {
+ rwLock.unlockWrite(stamp);
+ }
+ }
+
+
+ @Override
+ public Pair<Position, Position> cursorUpdated(ManagedCursor cursor,
Position newPosition) {
+ requireNonNull(cursor);
+
+ long stamp = rwLock.writeLock();
+ try {
+ Item item = cursors.get(cursor.getName());
+ if (item == null || item.idx == -1) {
+ return null;
+ }
+
+ Position previousSlowestConsumer = heap.get(0).position;
+ item.position = newPosition;
+ version = DataVersion.getNextVersion(version);
+
+ if (heap.size() == 1) {
+ return Pair.of(previousSlowestConsumer, item.position);
+ }
+
+ // When the cursor moves forward, we need to push it toward the
+ // bottom of the tree and push it up if a reset was done
+ if (item.idx == 0 ||
getParent(item).position.compareTo(item.position) <= 0) {
+ siftDown(item);
+ } else {
+ siftUp(item);
+ }
+ Position newSlowestConsumer = heap.get(0).position;
+ return Pair.of(previousSlowestConsumer, newSlowestConsumer);
+ } finally {
+ rwLock.unlockWrite(stamp);
+ }
+ }
+
+ @Override
+ public void updateCursor(ManagedCursor cursor, Position newPosition) {
+ cursorUpdated(cursor, newPosition);
+ }
+
+ /**
+ * Get the slowest reader position for the cursors that are ordered.
+ *
+ * @return the slowest reader position
+ */
+ @Override
+ public Position getSlowestCursorPosition() {
+ long stamp = rwLock.readLock();
+ try {
+ return heap.isEmpty() ? null : heap.get(0).position;
+ } finally {
+ rwLock.unlockRead(stamp);
+ }
+ }
+
+ @Override
+ public ManagedCursor getSlowestCursor() {
+ long stamp = rwLock.readLock();
+ try {
+ return heap.isEmpty() ? null : heap.get(0).cursor;
+ } finally {
+ rwLock.unlockRead(stamp);
+ }
+ }
+
+ /**
+ * @return Returns the CursorInfo for the cursor with the oldest position,
+ * or null if there aren't any tracked cursors
+ */
+ @Override
+ public CursorInfo getCursorWithOldestPosition() {
+ long stamp = rwLock.readLock();
+ try {
+ if (heap.isEmpty()) {
+ return null;
+ } else {
+ Item item = heap.get(0);
+ return new CursorInfo(item.cursor, item.position, version);
+ }
+ } finally {
+ rwLock.unlockRead(stamp);
+ }
+ }
+
+ /**
+ * Check whether there are any cursors.
Review Comment:
```suggestion
* Check whether there are any cursors.
```
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainer.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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 org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.commons.lang3.tuple.Pair;
+
+/**
+ * Contains cursors for a ManagedLedger that are actively being used.
+ * <p>
+ * The goal is to be able to use the container to be used for cache eviction
where the tracking of the slowest cursor
+ * is important for determining which cache entries can be evicted.
+ * <p>
+ * The usage depends on the cache eviction configuration:
+ *
+ * <p>When cacheEvictionByMarkDeletedPosition is enabled, the slowest cursor
is the one with the oldest mark deleted
+ * position. Otherwise, it is the one with the oldest read position.
+ *
+ * <p>When cacheEvictionByExpectedReadCount is enabled, the slowest cursor is
not necessarily tracked actively for
+ * every update and the position of the slowest cursor is not returned
immediately. The implementation optimized
+ * for cacheEvictionByExpectedReadCount will throw {@link
UnsupportedOperationException} for calls to
+ * {@link #cursorUpdated(ManagedCursor, Position)} method. The {@link
#updateCursor(ManagedCursor, Position)} method
+ * is used instead to update the cursor position.
+ */
+public interface ActiveManagedCursorContainer extends Iterable<ManagedCursor> {
+ /**
+ * Adds a cursor to the container with the specified position.
+ *
+ * @param cursor The cursor to add
+ * @param position The position of the cursor, if null, the cursor won't
be tracked in the slowest cursor
+ * tracking.
+ */
+ void add(ManagedCursor cursor, Position position);
+
+ /**
+ * Gets a cursor by its name.
+ *
+ * @param name the name of the cursor
+ * @return the ManagedCursor if found, otherwise null
+ */
+ ManagedCursor get(String name);
+
+ /**
+ * Removes a cursor from the container by its name.
+ *
+ * @param name the name of the cursor to remove
+ * @return true if the cursor was removed, false if it was not found
+ */
+ boolean removeCursor(String name);
+
+ /**
+ * Signal that a cursor position has been updated and that the container
must re-order the cursor heap
+ * tracking the slowest cursor and return the previous position of the
slowest cursor and the possibly updated
+ * position of the slowest cursor.
+ *
+ * @param cursor the cursor to update the position for
+ * @param newPosition the updated position for the cursor
+ * @return a pair of positions, representing the previous slowest cursor
and the new slowest cursor (after the
+ * update).
+ */
+ Pair<Position, Position> cursorUpdated(ManagedCursor cursor, Position
newPosition);
+
+ /**
+ * Updates the cursor position without immediately returning the positions
for the slowest cursor.
+ * Compared to {@link #cursorUpdated(ManagedCursor, Position)}, this
method does require the implementation
+ * to immediately re-order the cursor heap tracking the slowest cursor
allowing to optimize the performance
+ * to batch updates in re-ordering.
+ *
+ * @param cursor The cursor to update
+ * @param newPosition The new position to set for the cursor
+ */
+ void updateCursor(ManagedCursor cursor, Position newPosition);
+
+ /**
+ * Gets the position of the slowest cursor.
+ *
+ * @return the position of the slowest cursor
+ */
+ Position getSlowestCursorPosition();
+
+ /**
+ * Checks if the container is empty.
+ *
+ * @return true if the container has no cursors, false otherwise
+ */
+ boolean isEmpty();
+
+ /**
+ * Gets the number of cursors in the container.
+ *
+ * @return the number of cursors
+ */
+ int size();
+
+ /**
+ * Returns the number of cursors that are at the same position or before
the given cursor.
+ * This is currently used in the cacheEvictionByExpectedReadCount cache
eviction algorithm implementation
+ * to estimate how many reads are expected to be performed by cursors for
a given position.
+ * @param cursor the cursor for which to count the number of cursors at
the same position or before
+ * @return the number of cursors at the same position or before the given
cursor, includes the cursor itself
+ */
+ default int getNumberOfCursorsAtSamePositionOrBefore(ManagedCursor cursor)
{
+ throw new UnsupportedOperationException("This method is not supported
by this implementation");
+ }
+
+ /**
+ * Returns true if added entries should be cached. The implementation s
Review Comment:
There is something missing at the end here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
