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 2021/07/08 02:02:28 UTC
[GitHub] [pulsar] aloyszhang commented on a change in pull request #11242: fix retention size policy delete too much ledgers
aloyszhang commented on a change in pull request #11242:
URL: https://github.com/apache/pulsar/pull/11242#discussion_r665817228
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -2369,33 +2375,52 @@ void internalTrimLedgers(boolean isTruncate, CompletableFuture<?> promise) {
if (log.isDebugEnabled()) {
log.debug("[{}] Slowest consumer ledger id: {}", name, slowestReaderLedgerId);
}
+
+
+ long totalSizeToDelete = 0;
+ boolean retentionSizeQuotaMet = false;
// skip ledger if retention constraint met
for (LedgerInfo ls : ledgers.headMap(slowestReaderLedgerId, false).values()) {
- boolean expired = hasLedgerRetentionExpired(ls.getTimestamp());
- boolean overRetentionQuota = isLedgerRetentionOverSizeQuota();
-
- if (log.isDebugEnabled()) {
- log.debug(
- "[{}] Checking ledger {} -- time-old: {} sec -- "
- + "expired: {} -- over-quota: {} -- current-ledger: {}",
- name, ls.getLedgerId(), (clock.millis() - ls.getTimestamp()) / 1000.0, expired,
- overRetentionQuota, currentLedger.getId());
- }
+ // currentLedger can not be deleted
if (ls.getLedgerId() == currentLedger.getId()) {
log.debug("[{}] Ledger {} skipped for deletion as it is currently being written to", name,
ls.getLedgerId());
break;
- } else if (expired || isTruncate) {
- log.debug("[{}] Ledger {} has expired or be truncated, expired is {}, isTruncate is {}, ts {}", name, ls.getLedgerId(), expired, isTruncate, ls.getTimestamp());
+ }
+ // if truncate, all ledgers besides currentLedger are going to be deleted
+ if (isTruncate){
+ log.debug("[{}] Ledger {} has been truncated, , ts {}", name, ls.getLedgerId(), ls.getTimestamp());
ledgersToDelete.add(ls);
- } else if (overRetentionQuota || isTruncate) {
- log.debug("[{}] Ledger {} is over quota or be truncated, overRetentionQuota is {}, isTruncate is {}", name, ls.getLedgerId(), overRetentionQuota, isTruncate);
+ continue;
+ }
+
+ // if neither currentLedger nor truncate
+ if (!retentionSizeQuotaMet) {
+ totalSizeToDelete += ls.getSize();
+ boolean overRetentionQuota = isLedgerRetentionOverSizeQuota();
+ boolean overRetentionQuotaAfterDelete = isLedgerRetentionOverSizeQuotaAfterDelete(totalSizeToDelete);
+ if (overRetentionQuota) {
+ if (overRetentionQuotaAfterDelete) {
+ log.debug("[{}] Ledger {} is over quota", name, ls.getLedgerId());
+ ledgersToDelete.add(ls);
+ continue;
+ } else {
+ retentionSizeQuotaMet = true;
+ ledgersToDelete.add(ls);
+ continue;
+ }
+ } else {
+ retentionSizeQuotaMet = true;
+ }
+ }
+
+ if (hasLedgerRetentionExpired(ls.getTimestamp())) {
Review comment:
> the correct logic is if the ledger over retention quota and expired, we should add it into the ledgersToDelete right?
I think the logic of retention policy means either over retention quota or expired, the ledger should be delete.
--
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-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org