You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/04/27 23:21:02 UTC
[GitHub] sijie closed pull request #1668: BookKeeper ledgers cleaned up
after offload
sijie closed pull request #1668: BookKeeper ledgers cleaned up after offload
URL: https://github.com/apache/incubator-pulsar/pull/1668
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
index f3add7c4fc..566701c1dd 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
@@ -54,6 +54,7 @@
private long retentionTimeMs = 0;
private long retentionSizeInMB = 0;
private boolean autoSkipNonRecoverableData;
+ private long offloadLedgerDeletionLagMs = TimeUnit.HOURS.toMillis(4);
private DigestType digestType = DigestType.CRC32C;
private byte[] password = "".getBytes(Charsets.UTF_8);
@@ -386,6 +387,28 @@ public long getRetentionSizeInMB() {
return retentionSizeInMB;
}
+ /**
+ * When a ledger is offloaded from bookkeeper storage to longterm storage, the bookkeeper ledger
+ * is not deleted immediately. Instead we wait for a grace period before deleting from bookkeeper.
+ * The offloadLedgerDeleteLag sets this grace period.
+ *
+ * @param lagTime period to wait before deleting offloaded ledgers from bookkeeper
+ * @param unit timeunit for lagTime
+ */
+ public ManagedLedgerConfig setOffloadLedgerDeletionLag(int lagTime, TimeUnit unit) {
+ this.offloadLedgerDeletionLagMs = unit.toMillis(lagTime);
+ return this;
+ }
+
+ /**
+ * Number of milliseconds before an offloaded ledger will be deleted from bookkeeper.
+ *
+ * @return the offload ledger deletion lag time in milliseconds
+ */
+ public long getOffloadLedgerDeletionLagMillis() {
+ return offloadLedgerDeletionLagMs;
+ }
+
/**
* Skip reading non-recoverable/unreadable data-ledger under managed-ledger's list. It helps when data-ledgers gets
* corrupted at bookkeeper and managed-cursor is stuck at that ledger.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 59f4ad7b12..1e3a6c0b8f 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -97,6 +97,7 @@
import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
import org.apache.bookkeeper.mledger.impl.MetaStore.Stat;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.OffloadContext;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.NestedPositionInfo;
import org.apache.bookkeeper.mledger.util.CallbackMutex;
@@ -1534,13 +1535,17 @@ void notifyCursors() {
}
private void trimConsumedLedgersInBackground() {
+ trimConsumedLedgersInBackground(Futures.NULL_PROMISE);
+ }
+
+ private void trimConsumedLedgersInBackground(CompletableFuture<?> promise) {
executor.executeOrdered(name, safeRun(() -> {
- internalTrimConsumedLedgers();
+ internalTrimConsumedLedgers(promise);
}));
}
- private void scheduleDeferredTrimming() {
- scheduledExecutor.schedule(safeRun(() -> trimConsumedLedgersInBackground()), 100, TimeUnit.MILLISECONDS);
+ private void scheduleDeferredTrimming(CompletableFuture<?> promise) {
+ scheduledExecutor.schedule(safeRun(() -> trimConsumedLedgersInBackground(promise)), 100, TimeUnit.MILLISECONDS);
}
private boolean hasLedgerRetentionExpired(long ledgerTimestamp) {
@@ -1559,20 +1564,27 @@ private boolean isLedgerRetentionOverSizeQuota() {
&& TOTAL_SIZE_UPDATER.get(this) > ((long) config.getRetentionSizeInMB()) * 1024 * 1024;
}
+ private boolean isOffloadedNeedsDelete(OffloadContext offload) {
+ long elapsedMs = clock.millis() - offload.getTimestamp();
+ return offload.getComplete()
+ && !offload.getBookkeeperDeleted()
+ && elapsedMs > config.getOffloadLedgerDeletionLagMillis();
+ }
+
/**
* Checks whether there are ledger that have been fully consumed and deletes them.
*
* @throws Exception
*/
- void internalTrimConsumedLedgers() {
+ void internalTrimConsumedLedgers(CompletableFuture<?> promise) {
// Ensure only one trimming operation is active
if (!trimmerMutex.tryLock()) {
- scheduleDeferredTrimming();
+ scheduleDeferredTrimming(promise);
return;
}
List<LedgerInfo> ledgersToDelete = Lists.newArrayList();
-
+ List<LedgerInfo> offloadedLedgersToDelete = Lists.newArrayList();
synchronized (this) {
if (log.isDebugEnabled()) {
log.debug("[{}] Start TrimConsumedLedgers. ledgers={} totalSize={}", name, ledgers.keySet(),
@@ -1581,6 +1593,7 @@ void internalTrimConsumedLedgers() {
if (STATE_UPDATER.get(this) == State.Closed) {
log.debug("[{}] Ignoring trimming request since the managed ledger was already closed", name);
trimmerMutex.unlock();
+ promise.completeExceptionally(new ManagedLedgerAlreadyClosedException("Can't trim closed ledger"));
return;
}
@@ -1595,6 +1608,7 @@ void internalTrimConsumedLedgers() {
if (slowestReaderPosition != null) {
slowestReaderLedgerId = slowestReaderPosition.getLedgerId();
} else {
+ promise.completeExceptionally(new ManagedLedgerException("Couldn't find reader position"));
trimmerMutex.unlock();
return;
}
@@ -1616,43 +1630,55 @@ void internalTrimConsumedLedgers() {
name, ls.getLedgerId(), (clock.millis() - ls.getTimestamp()) / 1000.0, expired,
overRetentionQuota, currentLedger.getId());
}
- if (ls.getLedgerId() == currentLedger.getId() || (!expired && !overRetentionQuota)) {
- if (log.isDebugEnabled()) {
- if (!expired) {
- log.debug("[{}] ledger id skipped for deletion as unexpired: {}", name, ls.getLedgerId());
- }
- if (!overRetentionQuota) {
- log.debug("[{}] ledger id: {} skipped for deletion as size: {} under quota: {} MB", name,
- ls.getLedgerId(), TOTAL_SIZE_UPDATER.get(this), config.getRetentionSizeInMB());
- }
- }
+ if (ls.getLedgerId() == currentLedger.getId()) {
+ log.debug("[{}] ledger id skipped for deletion as it is currently being written to",
+ name, ls.getLedgerId());
+ break;
+ } else if (expired) {
+ log.debug("[{}] Ledger {} has expired, ts {}", name, ls.getLedgerId(), ls.getTimestamp());
+ ledgersToDelete.add(ls);
+ } else if (overRetentionQuota) {
+ log.debug("[{}] Ledger {} is over quota", name, ls.getLedgerId());
+ ledgersToDelete.add(ls);
+ } else if (isOffloadedNeedsDelete(ls.getOffloadContext())) {
+ log.debug("[{}] Ledger {} has been offloaded, bookkeeper ledger needs to be deleted",
+ name, ls.getLedgerId());
+ offloadedLedgersToDelete.add(ls);
+ } else {
+ log.debug("[{}] Nothing done for ledger {}. Neither expired, over-quota nor offloaded",
+ name, ls.getLedgerId());
break;
}
-
- ledgersToDelete.add(ls);
- ledgerCache.remove(ls.getLedgerId());
}
- if (ledgersToDelete.isEmpty()) {
+ if (ledgersToDelete.isEmpty() && offloadedLedgersToDelete.isEmpty()) {
trimmerMutex.unlock();
+ promise.complete(null);
return;
}
if (STATE_UPDATER.get(this) == State.CreatingLedger // Give up now and schedule a new trimming
|| !ledgersListMutex.tryLock()) { // Avoid deadlocks with other operations updating the ledgers list
- scheduleDeferredTrimming();
+ scheduleDeferredTrimming(promise);
trimmerMutex.unlock();
return;
}
// Update metadata
for (LedgerInfo ls : ledgersToDelete) {
+ ledgerCache.remove(ls.getLedgerId());
+
ledgers.remove(ls.getLedgerId());
NUMBER_OF_ENTRIES_UPDATER.addAndGet(this, -ls.getEntries());
TOTAL_SIZE_UPDATER.addAndGet(this, -ls.getSize());
entryCache.invalidateAllEntries(ls.getLedgerId());
}
+ for (LedgerInfo ls : offloadedLedgersToDelete) {
+ LedgerInfo.Builder newInfoBuilder = ls.toBuilder();
+ newInfoBuilder.getOffloadContextBuilder().setBookkeeperDeleted(true);
+ ledgers.put(ls.getLedgerId(), newInfoBuilder.build());
+ }
if (log.isDebugEnabled()) {
log.debug("[{}] Updating of ledgers list after trimming", name);
@@ -1671,6 +1697,12 @@ public void operationComplete(Void result, Stat stat) {
log.info("[{}] Removing ledger {} - size: {}", name, ls.getLedgerId(), ls.getSize());
asyncDeleteLedger(ls.getLedgerId(), ls);
}
+ for (LedgerInfo ls : offloadedLedgersToDelete) {
+ log.info("[{}] Deleting offloaded ledger {} from bookkeeper - size: {}",
+ name, ls.getLedgerId(), ls.getSize());
+ asyncDeleteLedgerFromBookKeeper(ls.getLedgerId());
+ }
+ promise.complete(null);
}
@Override
@@ -1678,6 +1710,8 @@ public void operationFailed(MetaStoreException e) {
log.warn("[{}] Failed to update the list of ledgers after trimming", name, e);
ledgersListMutex.unlock();
trimmerMutex.unlock();
+
+ promise.completeExceptionally(e);
}
});
}
@@ -1761,8 +1795,15 @@ public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) {
}
}
- private void asyncDeleteLedger(long ledgerId, LedgerInfo info) {
+ private void asyncDeleteLedgerFromBookKeeper(long ledgerId) {
asyncDeleteLedger(ledgerId, DEFAULT_LEDGER_DELETE_RETRIES);
+ }
+
+ private void asyncDeleteLedger(long ledgerId, LedgerInfo info) {
+ if (!info.getOffloadContext().getBookkeeperDeleted()) {
+ // only delete if it hasn't been previously deleted for offload
+ asyncDeleteLedger(ledgerId, DEFAULT_LEDGER_DELETE_RETRIES);
+ }
if (info.getOffloadContext().hasUidMsb()) {
UUID uuid = new UUID(info.getOffloadContext().getUidMsb(),
@@ -2075,7 +2116,8 @@ public void operationFailed(MetaStoreException e) {
})
.whenComplete((result, exception) -> {
if (exception != null) {
- log.warn("[{}] Failed to prepare ledger {} for offload, uuid {}", name, ledgerId, uuid);
+ log.warn("[{}] Failed to prepare ledger {} for offload, uuid {}",
+ name, ledgerId, uuid, exception);
} else {
log.info("[{}] Metadata prepared for offload of ledger {} with uuid {}", name, ledgerId, uuid);
}
@@ -2091,6 +2133,7 @@ public void operationFailed(MetaStoreException e) {
if (existingUuid.equals(uuid)) {
LedgerInfo.Builder builder = oldInfo.toBuilder();
builder.getOffloadContextBuilder()
+ .setTimestamp(clock.millis())
.setComplete(true);
return builder.build();
} else {
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 1f0fae4fca..73ba1dab56 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
@@ -22,6 +22,14 @@ public static void registerAllExtensions(
// optional bool complete = 3;
boolean hasComplete();
boolean getComplete();
+
+ // optional bool bookkeeperDeleted = 4;
+ boolean hasBookkeeperDeleted();
+ boolean getBookkeeperDeleted();
+
+ // optional int64 timestamp = 5;
+ boolean hasTimestamp();
+ long getTimestamp();
}
public static final class OffloadContext extends
com.google.protobuf.GeneratedMessage
@@ -82,10 +90,32 @@ public boolean getComplete() {
return complete_;
}
+ // optional bool bookkeeperDeleted = 4;
+ public static final int BOOKKEEPERDELETED_FIELD_NUMBER = 4;
+ private boolean bookkeeperDeleted_;
+ public boolean hasBookkeeperDeleted() {
+ return ((bitField0_ & 0x00000008) == 0x00000008);
+ }
+ public boolean getBookkeeperDeleted() {
+ return bookkeeperDeleted_;
+ }
+
+ // optional int64 timestamp = 5;
+ public static final int TIMESTAMP_FIELD_NUMBER = 5;
+ private long timestamp_;
+ public boolean hasTimestamp() {
+ return ((bitField0_ & 0x00000010) == 0x00000010);
+ }
+ public long getTimestamp() {
+ return timestamp_;
+ }
+
private void initFields() {
uidMsb_ = 0L;
uidLsb_ = 0L;
complete_ = false;
+ bookkeeperDeleted_ = false;
+ timestamp_ = 0L;
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -108,6 +138,12 @@ public void writeTo(com.google.protobuf.CodedOutputStream output)
if (((bitField0_ & 0x00000004) == 0x00000004)) {
output.writeBool(3, complete_);
}
+ if (((bitField0_ & 0x00000008) == 0x00000008)) {
+ output.writeBool(4, bookkeeperDeleted_);
+ }
+ if (((bitField0_ & 0x00000010) == 0x00000010)) {
+ output.writeInt64(5, timestamp_);
+ }
getUnknownFields().writeTo(output);
}
@@ -129,6 +165,14 @@ public int getSerializedSize() {
size += com.google.protobuf.CodedOutputStream
.computeBoolSize(3, complete_);
}
+ if (((bitField0_ & 0x00000008) == 0x00000008)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeBoolSize(4, bookkeeperDeleted_);
+ }
+ if (((bitField0_ & 0x00000010) == 0x00000010)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeInt64Size(5, timestamp_);
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -259,6 +303,10 @@ public Builder clear() {
bitField0_ = (bitField0_ & ~0x00000002);
complete_ = false;
bitField0_ = (bitField0_ & ~0x00000004);
+ bookkeeperDeleted_ = false;
+ bitField0_ = (bitField0_ & ~0x00000008);
+ timestamp_ = 0L;
+ bitField0_ = (bitField0_ & ~0x00000010);
return this;
}
@@ -309,6 +357,14 @@ public Builder clone() {
to_bitField0_ |= 0x00000004;
}
result.complete_ = complete_;
+ if (((from_bitField0_ & 0x00000008) == 0x00000008)) {
+ to_bitField0_ |= 0x00000008;
+ }
+ result.bookkeeperDeleted_ = bookkeeperDeleted_;
+ if (((from_bitField0_ & 0x00000010) == 0x00000010)) {
+ to_bitField0_ |= 0x00000010;
+ }
+ result.timestamp_ = timestamp_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -334,6 +390,12 @@ public Builder mergeFrom(org.apache.bookkeeper.mledger.proto.MLDataFormats.Offlo
if (other.hasComplete()) {
setComplete(other.getComplete());
}
+ if (other.hasBookkeeperDeleted()) {
+ setBookkeeperDeleted(other.getBookkeeperDeleted());
+ }
+ if (other.hasTimestamp()) {
+ setTimestamp(other.getTimestamp());
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -380,6 +442,16 @@ public Builder mergeFrom(
complete_ = input.readBool();
break;
}
+ case 32: {
+ bitField0_ |= 0x00000008;
+ bookkeeperDeleted_ = input.readBool();
+ break;
+ }
+ case 40: {
+ bitField0_ |= 0x00000010;
+ timestamp_ = input.readInt64();
+ break;
+ }
}
}
}
@@ -449,6 +521,48 @@ public Builder clearComplete() {
return this;
}
+ // optional bool bookkeeperDeleted = 4;
+ private boolean bookkeeperDeleted_ ;
+ public boolean hasBookkeeperDeleted() {
+ return ((bitField0_ & 0x00000008) == 0x00000008);
+ }
+ public boolean getBookkeeperDeleted() {
+ return bookkeeperDeleted_;
+ }
+ public Builder setBookkeeperDeleted(boolean value) {
+ bitField0_ |= 0x00000008;
+ bookkeeperDeleted_ = value;
+ onChanged();
+ return this;
+ }
+ public Builder clearBookkeeperDeleted() {
+ bitField0_ = (bitField0_ & ~0x00000008);
+ bookkeeperDeleted_ = false;
+ onChanged();
+ return this;
+ }
+
+ // optional int64 timestamp = 5;
+ private long timestamp_ ;
+ public boolean hasTimestamp() {
+ return ((bitField0_ & 0x00000010) == 0x00000010);
+ }
+ public long getTimestamp() {
+ return timestamp_;
+ }
+ public Builder setTimestamp(long value) {
+ bitField0_ |= 0x00000010;
+ timestamp_ = value;
+ onChanged();
+ return this;
+ }
+ public Builder clearTimestamp() {
+ bitField0_ = (bitField0_ & ~0x00000010);
+ timestamp_ = 0L;
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:OffloadContext)
}
@@ -5359,29 +5473,31 @@ public Builder removeProperties(int index) {
descriptor;
static {
java.lang.String[] descriptorData = {
- "\n\"src/main/proto/MLDataFormats.proto\"B\n\016" +
+ "\n\"src/main/proto/MLDataFormats.proto\"p\n\016" +
"OffloadContext\022\016\n\006uidMsb\030\001 \001(\003\022\016\n\006uidLsb" +
- "\030\002 \001(\003\022\020\n\010complete\030\003 \001(\010\"\362\001\n\021ManagedLedg" +
- "erInfo\0221\n\nledgerInfo\030\001 \003(\0132\035.ManagedLedg" +
- "erInfo.LedgerInfo\022/\n\022terminatedPosition\030" +
- "\002 \001(\0132\023.NestedPositionInfo\032y\n\nLedgerInfo" +
- "\022\020\n\010ledgerId\030\001 \002(\003\022\017\n\007entries\030\002 \001(\003\022\014\n\004s" +
- "ize\030\003 \001(\003\022\021\n\ttimestamp\030\004 \001(\003\022\'\n\016offloadC" +
- "ontext\030\005 \001(\0132\017.OffloadContext\"\206\001\n\014Positi" +
- "onInfo\022\020\n\010ledgerId\030\001 \002(\003\022\017\n\007entryId\030\002 \002(",
- "\003\0220\n\031individualDeletedMessages\030\003 \003(\0132\r.M" +
- "essageRange\022!\n\nproperties\030\004 \003(\0132\r.LongPr" +
- "operty\"7\n\022NestedPositionInfo\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.NestedPositionI" +
- "nfo\022*\n\rupperEndpoint\030\002 \002(\0132\023.NestedPosit" +
- "ionInfo\"+\n\014LongProperty\022\014\n\004name\030\001 \002(\t\022\r\n" +
- "\005value\030\002 \002(\003\"\270\001\n\021ManagedCursorInfo\022\027\n\017cu" +
- "rsorsLedgerId\030\001 \002(\003\022\032\n\022markDeleteLedgerI" +
- "d\030\002 \001(\003\022\031\n\021markDeleteEntryId\030\003 \001(\003\0220\n\031in",
- "dividualDeletedMessages\030\004 \003(\0132\r.MessageR" +
- "ange\022!\n\nproperties\030\005 \003(\0132\r.LongPropertyB" +
- "\'\n#org.apache.bookkeeper.mledger.protoH\001"
+ "\030\002 \001(\003\022\020\n\010complete\030\003 \001(\010\022\031\n\021bookkeeperDe" +
+ "leted\030\004 \001(\010\022\021\n\ttimestamp\030\005 \001(\003\"\362\001\n\021Manag" +
+ "edLedgerInfo\0221\n\nledgerInfo\030\001 \003(\0132\035.Manag" +
+ "edLedgerInfo.LedgerInfo\022/\n\022terminatedPos" +
+ "ition\030\002 \001(\0132\023.NestedPositionInfo\032y\n\nLedg" +
+ "erInfo\022\020\n\010ledgerId\030\001 \002(\003\022\017\n\007entries\030\002 \001(" +
+ "\003\022\014\n\004size\030\003 \001(\003\022\021\n\ttimestamp\030\004 \001(\003\022\'\n\016of" +
+ "floadContext\030\005 \001(\0132\017.OffloadContext\"\206\001\n\014",
+ "PositionInfo\022\020\n\010ledgerId\030\001 \002(\003\022\017\n\007entryI" +
+ "d\030\002 \002(\003\0220\n\031individualDeletedMessages\030\003 \003" +
+ "(\0132\r.MessageRange\022!\n\nproperties\030\004 \003(\0132\r." +
+ "LongProperty\"7\n\022NestedPositionInfo\022\020\n\010le" +
+ "dgerId\030\001 \002(\003\022\017\n\007entryId\030\002 \002(\003\"f\n\014Message" +
+ "Range\022*\n\rlowerEndpoint\030\001 \002(\0132\023.NestedPos" +
+ "itionInfo\022*\n\rupperEndpoint\030\002 \002(\0132\023.Neste" +
+ "dPositionInfo\"+\n\014LongProperty\022\014\n\004name\030\001 " +
+ "\002(\t\022\r\n\005value\030\002 \002(\003\"\270\001\n\021ManagedCursorInfo" +
+ "\022\027\n\017cursorsLedgerId\030\001 \002(\003\022\032\n\022markDeleteL",
+ "edgerId\030\002 \001(\003\022\031\n\021markDeleteEntryId\030\003 \001(\003" +
+ "\0220\n\031individualDeletedMessages\030\004 \003(\0132\r.Me" +
+ "ssageRange\022!\n\nproperties\030\005 \003(\0132\r.LongPro" +
+ "pertyB\'\n#org.apache.bookkeeper.mledger.p" +
+ "rotoH\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -5393,7 +5509,7 @@ public Builder removeProperties(int index) {
internal_static_OffloadContext_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_OffloadContext_descriptor,
- new java.lang.String[] { "UidMsb", "UidLsb", "Complete", },
+ new java.lang.String[] { "UidMsb", "UidLsb", "Complete", "BookkeeperDeleted", "Timestamp", },
org.apache.bookkeeper.mledger.proto.MLDataFormats.OffloadContext.class,
org.apache.bookkeeper.mledger.proto.MLDataFormats.OffloadContext.Builder.class);
internal_static_ManagedLedgerInfo_descriptor =
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/Futures.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/Futures.java
index 11801bb342..9fd4a071f2 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/Futures.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/Futures.java
@@ -29,6 +29,8 @@
*/
public class Futures {
+ public static CompletableFuture<Void> NULL_PROMISE = CompletableFuture.completedFuture(null);
+
/**
* Adapts a {@link CloseCallback} to a {@link CompletableFuture}.
*/
diff --git a/managed-ledger/src/main/proto/MLDataFormats.proto b/managed-ledger/src/main/proto/MLDataFormats.proto
index 99e26e3cb4..29f465c9c2 100644
--- a/managed-ledger/src/main/proto/MLDataFormats.proto
+++ b/managed-ledger/src/main/proto/MLDataFormats.proto
@@ -25,6 +25,8 @@ message OffloadContext {
optional int64 uidMsb = 1;
optional int64 uidLsb = 2;
optional bool complete = 3;
+ optional bool bookkeeperDeleted = 4;
+ optional int64 timestamp = 5;
}
message ManagedLedgerInfo {
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index 60d66a1a08..d7cce973bd 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -40,6 +40,7 @@
import java.util.Iterator;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executor;
@@ -1670,7 +1671,7 @@ public void testDeletionAfterRetention() throws Exception {
c1.skipEntries(1, IndividualDeletedEntries.Exclude);
// let retention expire
Thread.sleep(1000);
- ml.internalTrimConsumedLedgers();
+ ml.internalTrimConsumedLedgers(CompletableFuture.completedFuture(null));
assertTrue(ml.getLedgersInfoAsList().size() <= 1);
assertTrue(ml.getTotalSize() <= "shortmessage".getBytes().length);
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadLedgerDeleteTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadLedgerDeleteTest.java
new file mode 100644
index 0000000000..41a7ad0480
--- /dev/null
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadLedgerDeleteTest.java
@@ -0,0 +1,148 @@
+/**
+ * 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.apache.bookkeeper.mledger.impl.OffloadPrefixTest.assertEventuallyTrue;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.util.MockClock;
+import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class OffloadLedgerDeleteTest extends MockedBookKeeperTestCase {
+ private static final Logger log = LoggerFactory.getLogger(OffloadLedgerDeleteTest.class);
+
+ @Test
+ public void testLaggedDelete() throws Exception {
+ OffloadPrefixTest.MockLedgerOffloader offloader = new OffloadPrefixTest.MockLedgerOffloader();
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ MockClock clock = new MockClock();
+ config.setMaxEntriesPerLedger(10);
+ config.setMinimumRolloverTime(0, TimeUnit.SECONDS);
+ config.setRetentionTime(10, TimeUnit.MINUTES);
+ config.setOffloadLedgerDeletionLag(5, TimeUnit.MINUTES);
+ config.setLedgerOffloader(offloader);
+ config.setClock(clock);
+
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl)factory.open("my_test_ledger", config);
+ int i = 0;
+ for (; i < 15; i++) {
+ String content = "entry-" + i;
+ ledger.addEntry(content.getBytes());
+ }
+ Assert.assertEquals(ledger.getLedgersInfoAsList().size(), 2);
+ long firstLedgerId = ledger.getLedgersInfoAsList().get(0).getLedgerId();
+
+ ledger.offloadPrefix(ledger.getLastConfirmedEntry());
+
+ Assert.assertEquals(ledger.getLedgersInfoAsList().size(), 2);
+ Assert.assertEquals(ledger.getLedgersInfoAsList().stream()
+ .filter(e -> e.getOffloadContext().getComplete())
+ .map(e -> e.getLedgerId()).collect(Collectors.toSet()),
+ offloader.offloadedLedgers());
+ Assert.assertTrue(bkc.getLedgers().contains(firstLedgerId));
+
+ clock.advance(2, TimeUnit.MINUTES);
+ CompletableFuture<Void> promise = new CompletableFuture<>();
+ ledger.internalTrimConsumedLedgers(promise);
+ promise.join();
+ Assert.assertTrue(bkc.getLedgers().contains(firstLedgerId));
+
+ clock.advance(5, TimeUnit.MINUTES);
+ CompletableFuture<Void> promise2 = new CompletableFuture<>();
+ ledger.internalTrimConsumedLedgers(promise2);
+ promise2.join();
+
+ // assert bk ledger is deleted
+ assertEventuallyTrue(() -> !bkc.getLedgers().contains(firstLedgerId));
+
+ // ledger still exists in list
+ Assert.assertEquals(ledger.getLedgersInfoAsList().stream()
+ .filter(e -> e.getOffloadContext().getComplete())
+ .map(e -> e.getLedgerId()).collect(Collectors.toSet()),
+ offloader.offloadedLedgers());
+
+ // move past retention, should be deleted from offloaded also
+ clock.advance(5, TimeUnit.MINUTES);
+ CompletableFuture<Void> promise3 = new CompletableFuture<>();
+ ledger.internalTrimConsumedLedgers(promise3);
+ promise3.join();
+
+ Assert.assertEquals(ledger.getLedgersInfoAsList().size(), 1);
+ assertEventuallyTrue(() -> offloader.deletedOffloads().contains(firstLedgerId));
+ }
+
+ @Test
+ public void testLaggedDeleteRetentionSetLower() throws Exception {
+ OffloadPrefixTest.MockLedgerOffloader offloader = new OffloadPrefixTest.MockLedgerOffloader();
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ MockClock clock = new MockClock();
+ config.setMaxEntriesPerLedger(10);
+ config.setMinimumRolloverTime(0, TimeUnit.SECONDS);
+ config.setRetentionTime(5, TimeUnit.MINUTES);
+ config.setOffloadLedgerDeletionLag(10, TimeUnit.MINUTES);
+ config.setLedgerOffloader(offloader);
+ config.setClock(clock);
+
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl)factory.open("my_test_ledger", config);
+ int i = 0;
+ for (; i < 15; i++) {
+ String content = "entry-" + i;
+ ledger.addEntry(content.getBytes());
+ }
+ Assert.assertEquals(ledger.getLedgersInfoAsList().size(), 2);
+ long firstLedgerId = ledger.getLedgersInfoAsList().get(0).getLedgerId();
+
+ ledger.offloadPrefix(ledger.getLastConfirmedEntry());
+
+ Assert.assertEquals(ledger.getLedgersInfoAsList().size(), 2);
+ Assert.assertEquals(ledger.getLedgersInfoAsList().stream()
+ .filter(e -> e.getOffloadContext().getComplete())
+ .map(e -> e.getLedgerId()).collect(Collectors.toSet()),
+ offloader.offloadedLedgers());
+ Assert.assertTrue(bkc.getLedgers().contains(firstLedgerId));
+
+ clock.advance(2, TimeUnit.MINUTES);
+ CompletableFuture<Void> promise = new CompletableFuture<>();
+ ledger.internalTrimConsumedLedgers(promise);
+ promise.join();
+ Assert.assertTrue(bkc.getLedgers().contains(firstLedgerId));
+
+ clock.advance(5, TimeUnit.MINUTES);
+ CompletableFuture<Void> promise2 = new CompletableFuture<>();
+ ledger.internalTrimConsumedLedgers(promise2);
+ promise2.join();
+
+ // ensure it gets deleted from both bookkeeper and offloader
+ assertEventuallyTrue(() -> !bkc.getLedgers().contains(firstLedgerId));
+ assertEventuallyTrue(() -> offloader.deletedOffloads().contains(firstLedgerId));
+ }
+}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java
index 311a82e600..278182cef3 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java
@@ -633,7 +633,7 @@ public void testOffloadDeleteIncomplete() throws Exception {
assertEventuallyTrue(() -> offloader.deletedOffloads().contains(firstLedger));
}
- void assertEventuallyTrue(BooleanSupplier predicate) throws Exception {
+ static void assertEventuallyTrue(BooleanSupplier predicate) throws Exception {
// wait up to 3 seconds
for (int i = 0; i < 30 && !predicate.getAsBoolean(); i++) {
Thread.sleep(100);
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/util/MockClock.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/util/MockClock.java
new file mode 100644
index 0000000000..b74bec3e6d
--- /dev/null
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/util/MockClock.java
@@ -0,0 +1,50 @@
+/**
+ * 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.util;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class MockClock extends Clock {
+ private Instant initialInstant = Clock.systemUTC().instant();
+ private AtomicLong delta = new AtomicLong(0);
+
+ public void advance(long period, TimeUnit unit) {
+ delta.addAndGet(unit.toNanos(period));
+ }
+
+ @Override
+ public Instant instant() {
+ return initialInstant.plusNanos(delta.get());
+ }
+
+ @Override
+ public Clock withZone(ZoneId zone) {
+ return this;
+ }
+
+ @Override
+ public ZoneId getZone() {
+ return ZoneId.systemDefault();
+ }
+}
+
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services