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 2022/01/26 07:14:25 UTC
[GitHub] [pulsar] wuzhanpeng commented on a change in pull request #13575: [PIP-129] Introduce intermediate state for ledger deletion
wuzhanpeng commented on a change in pull request #13575:
URL: https://github.com/apache/pulsar/pull/13575#discussion_r792360487
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -4039,6 +4122,190 @@ public void clearBacklogFailed(ManagedLedgerException exception, Object ctx) {
return statFuture;
}
+ /**
+ * During the execution of this method, lock {@code metadataMutex} needs to be held
+ * because the {@code propertiesMap} would be updated (not thread-safe).
+ * @param deletableLedgerIds
+ */
+ @Override
+ public void markDeletableLedgers(Collection<Long> deletableLedgerIds,
+ Collection<Long> deletableOffloadedLedgerIds) {
+ for (Long ledgerId : deletableLedgerIds) {
+ final String deletableLedgerMarker = DELETABLE_LEDGER_MARKER_PREFIX + ledgerId;
+ propertiesMap.put(deletableLedgerMarker, DELETABLE_LEDGER_PLACEHOLDER);
+ }
+ for (Long ledgerId : deletableOffloadedLedgerIds) {
+ final String deletableOffloadedLedgerMarker = DELETABLE_OFFLOADED_LEDGER_MARKER_PREFIX + ledgerId;
+ // Offload context info is required in ledger cleanup, therefore the serialized info object
+ // is kept in the propertiesMap until the ledger deletion is done
+ final String offloadedLedgerInfo = BaseEncoding.base64().encode(ledgers.get(ledgerId).toByteArray());
+ propertiesMap.put(deletableOffloadedLedgerMarker, offloadedLedgerInfo);
+ }
+ }
+
+ private Set<Long> getAllDeletableLedgers(String prefix) {
+ Set<Long> deletableLedgers = propertiesMap.keySet().stream()
+ .filter(k -> k.startsWith(prefix))
+ .map(k -> {
+ Long ledgerId = Long.parseLong(k.substring(prefix.length()));
+ if (deletableLedgerRetryCounter.containsKey(ledgerId)
+ && deletableLedgerRetryCounter.get(ledgerId).get() >= DEFAULT_LEDGER_DELETE_RETRIES) {
+ log.error("[{}] Cannot delete ledger:{} after {} reties and now stop retrying on this broker",
+ name, ledgerId, DEFAULT_LEDGER_DELETE_RETRIES);
+ return null;
+ }
+ return ledgerId;
+ })
+ .filter(Objects::nonNull)
+ .collect(Collectors.toSet());
+ if (!deletableLedgers.isEmpty()) {
+ return deletableLedgers;
+ }
+ return Sets.newHashSet();
+ }
+
+ @Override
+ public Set<Long> getAllDeletableLedgers() {
+ return getAllDeletableLedgers(DELETABLE_LEDGER_MARKER_PREFIX);
+ }
+
+ @Override
+ public Set<Long> getAllDeletableOffloadedLedgers() {
+ return getAllDeletableLedgers(DELETABLE_OFFLOADED_LEDGER_MARKER_PREFIX);
+ }
+
+ /**
+ * During the execution of this method, lock {@code metadataMutex} needs to be held
+ * because the {@code propertiesMap} would be updated (not thread-safe).
+ */
+ @Override
+ public void removeAllDeletableLedgers() {
+ Set<Long> deletableLedgers = getAllDeletableLedgers();
+ Set<Long> deletableOffloadedLedgers = getAllDeletableOffloadedLedgers();
+ final CountDownLatch counter = new CountDownLatch(deletableLedgers.size() + deletableOffloadedLedgers.size());
+
+ Set<Long> finishedDeletedLedgers = ConcurrentHashMap.newKeySet();
+ Set<Long> finishedDeletedOffloadedLedgers = ConcurrentHashMap.newKeySet();
+ Set<Long> timeoutDeletedLedgers = ConcurrentHashMap.newKeySet();
+
+ Set<Long> succeedDeletedLedgers = ConcurrentHashMap.newKeySet();
+ Set<Long> failDeletedLedgers = ConcurrentHashMap.newKeySet();
+
+ Set<Long> succeedDeletedOffloadedLedgers = ConcurrentHashMap.newKeySet();
+ Set<Long> failDeletedOffloadedLedgers = ConcurrentHashMap.newKeySet();
+
+ for (Long deletableLedger : deletableLedgers) {
+ asyncDeleteLedger(deletableLedger, DEFAULT_LEDGER_DELETE_RETRIES,
+ new DeleteLedgerCallback() {
+ @Override
+ public void deleteLedgerComplete(Object ctx) {
+ counter.countDown();
+ finishedDeletedLedgers.add(deletableLedger);
+ succeedDeletedLedgers.add(deletableLedger);
+ }
+
+ @Override
+ public void deleteLedgerFailed(ManagedLedgerException exception, Object ctx) {
+ log.warn("[{}] Failed to delete bookkeeper ledger:{} due to",
+ name, deletableLedger, exception);
+ counter.countDown();
+ finishedDeletedLedgers.add(deletableLedger);
+ failDeletedLedgers.add(deletableLedger);
+ }
+ }, 0, null);
+ }
+
+ for (Long deletableOffloadedLedger : deletableOffloadedLedgers) {
+ final String deletableOffloadedLedgerMarker =
+ DELETABLE_OFFLOADED_LEDGER_MARKER_PREFIX + deletableOffloadedLedger;
+
+ try {
+ final LedgerInfo deletableOffloadedLedgerInfo = LedgerInfo.parseFrom(
+ BaseEncoding.base64().decode(propertiesMap.get(deletableOffloadedLedgerMarker)));
+ asyncDeleteOffloadedLedger(deletableOffloadedLedger, deletableOffloadedLedgerInfo,
+ DEFAULT_LEDGER_DELETE_RETRIES,
+ new DeleteLedgerCallback() {
+ @Override
+ public void deleteLedgerComplete(Object ctx) {
+ counter.countDown();
+ finishedDeletedOffloadedLedgers.add(deletableOffloadedLedger);
+ succeedDeletedOffloadedLedgers.add(deletableOffloadedLedger);
+ }
+
+ @Override
+ public void deleteLedgerFailed(ManagedLedgerException exception, Object ctx) {
+ log.warn("[{}] Failed to delete offloaded ledger:{} due to",
+ name, deletableOffloadedLedger, exception);
+ counter.countDown();
+ finishedDeletedOffloadedLedgers.add(deletableOffloadedLedger);
+ failDeletedOffloadedLedgers.add(deletableOffloadedLedger);
+ }
+ });
+ } catch (Exception e) {
+ log.warn("[{}] Failed to retrieve offloaded ledger info of {} due to",
+ name, deletableOffloadedLedger, e);
+ counter.countDown();
+ finishedDeletedOffloadedLedgers.add(deletableOffloadedLedger);
+ failDeletedOffloadedLedgers.add(deletableOffloadedLedger);
+ }
+ }
+
+ try {
+ if (!counter.await(AsyncOperationTimeoutSeconds, TimeUnit.SECONDS)) {
+ for (Long deletableLedger : deletableLedgers) {
+ if (!finishedDeletedLedgers.contains(deletableLedger)) {
+ log.warn("[{}] Failed to delete ledger:{} due to operation timeout", name, deletableLedger);
+ timeoutDeletedLedgers.add(deletableLedger);
+ }
+ }
+ for (Long deletableOffloadedLedger : deletableOffloadedLedgers) {
+ if (!finishedDeletedOffloadedLedgers.contains(deletableOffloadedLedger)) {
+ log.warn("[{}] Failed to delete offloaded ledger:{} due to operation timeout",
+ name, deletableOffloadedLedger);
+ timeoutDeletedLedgers.add(deletableOffloadedLedger);
+ }
+ }
+ }
+
+ // remove markers after deleting ledgers
+ for (Long ledgerId : succeedDeletedLedgers) {
+ final String deletableLedgerMarker = DELETABLE_LEDGER_MARKER_PREFIX + ledgerId;
+ propertiesMap.remove(deletableLedgerMarker);
+ }
+ for (Long ledgerId : succeedDeletedOffloadedLedgers) {
+ final String deletableLedgerMarker = DELETABLE_OFFLOADED_LEDGER_MARKER_PREFIX + ledgerId;
+ propertiesMap.remove(deletableLedgerMarker);
+ }
+
+ // update retry count to track whether the max limit is reached
+ Set<Long> allFailedLedgers = new HashSet<>();
+ allFailedLedgers.addAll(failDeletedLedgers);
+ allFailedLedgers.addAll(failDeletedOffloadedLedgers);
+ allFailedLedgers.addAll(timeoutDeletedLedgers);
+
+ if (allFailedLedgers.isEmpty()) {
+ log.info("[{}] ledgers: {} and offloaded ledgers: {} are deleted successfully.",
+ name, deletableLedgers, deletableOffloadedLedgers);
Review comment:
Those collection objects that inherit `java.util.AbstractCollection` have the ability to return a string representation of this collection (`java.util.AbstractCollection#toString`).
--
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