vaijosh commented on code in PR #3081:
URL: https://github.com/apache/hugegraph/pull/3081#discussion_r4270505542
##########
hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/AppConfig.java:
##########
@@ -312,6 +631,100 @@ public class RocksdbConfig {
private Map<String, String> rocksdb = new HashMap<>();
}
+ /**
+ * Spring {@link ConfigurationProperties} wrapper for the common cloud
storage properties.
+ *
+ * <p>Provider-specific keys (e.g. {@code cloud.storage.s3.bucket}) are
read
+ * directly from the Spring {@link Environment} at conversion time, so
this class
+ * has zero knowledge of any specific cloud provider.
+ *
+ * <pre>
+ * cloud:
+ * storage:
+ * enabled: true
+ * provider: s3 # selects which sub-namespace to forward to
the provider
+ * path-prefix: hugegraph
+ * s3: # all keys here are forwarded as-is to the S3
provider
+ * bucket: my-bucket
+ * region: us-east-1
+ * access-key: ${AWS_ACCESS_KEY_ID}
+ * secret-key: ${AWS_SECRET_ACCESS_KEY}
+ * </pre>
+ */
+ @Data
+ @Configuration
+ @ConfigurationProperties(prefix = "cloud.storage")
+ public class CloudStorageSpringConfig {
Review Comment:
Addressed
##########
hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/cloud/CloudUploadRetryQueue.java:
##########
@@ -0,0 +1,1159 @@
+/*
+ * 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.hugegraph.store.node.cloud;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.hugegraph.store.cloud.CloudStorageProvider;
+import org.apache.hugegraph.store.cloud.CloudStorageNonRetryableException;
+import org.apache.hugegraph.store.cloud.CloudStorageProviderFactory;
+
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Asynchronous retry queue with a file-backed dead-letter queue (DLQ) for
failed
+ * cloud SST uploads.
+ *
+ * <h3>Retry strategy</h3>
+ * <p>The number of whole-file retries is configured by the caller via
+ * {@code cloud.storage.upload-retry-max-attempts}, which defaults to {@code
3} (whole-file retries
+ * enabled). {@code maxAttempts == 0} is an opt-in mode — not the shipped
default — for deployments
+ * whose provider already performs sufficient internal retries (e.g. the S3
provider's underlying
+ * AWS SDK client, which retries transient per-request/per-part failures on
its own).
+ * <ul>
+ * <li>When {@code maxAttempts == 0}, failures go directly to the DLQ with
no whole-file retry.</li>
+ * <li>When {@code maxAttempts > 0} (the default), the first retry is
scheduled after
+ * {@code initialDelayMs} milliseconds.</li>
+ * <li>Subsequent retries use exponential backoff: {@code delay =
initialDelayMs * 2^(attempt-1)},
+ * capped at {@code maxDelayMs}.</li>
+ * <li>After {@code maxAttempts} consecutive failures the task is moved to
the DLQ.</li>
+ * <li>If the local SST file no longer exists at retry time (compacted away
by RocksDB)
+ * the retry is silently dropped — there is nothing to upload.</li>
+ * </ul>
+ *
+ * <h3>Dead-letter queue (DLQ)</h3>
+ * <ul>
+ * <li>In-memory: {@link #getDlqEntries()} returns a snapshot.</li>
+ * <li>On-disk: entries are appended to {@code
<dataRoot>/.cloud-upload-dlq.tsv} in a
+ * tab-separated format so they survive JVM restarts. Existing entries
are loaded
+ * back into memory at construction time.</li>
+ * <li>{@link #replayDlq()} retries all DLQ entries immediately; successful
uploads are
+ * removed and the file is rewritten. Failed entries are re-submitted to
the retry
+ * queue with a fresh attempt cycle.</li>
+ * </ul>
+ *
+ * <h3>Thread safety</h3>
+ * All public methods are thread-safe. The in-memory DLQ is a {@link
ConcurrentLinkedDeque};
+ * disk I/O is serialised onto the internal {@link ScheduledExecutorService}
(for appends)
+ * or synchronised explicitly (for rewrites).
+ *
+ * <h3>Lifecycle</h3>
+ * Call {@link #close()} (e.g. from a Spring {@code @PreDestroy} method) to
shut down the
+ * background executor gracefully.
+ */
+@Slf4j
+public class CloudUploadRetryQueue implements Closeable {
+
+ // -----------------------------------------------------------------------
+ // Constants
+ // -----------------------------------------------------------------------
+
+ /** Name of the on-disk DLQ file, relative to {@code dataRoot}. */
+ static final String DLQ_FILE_NAME = ".cloud-upload-dlq.tsv";
+
+ private static final String DLQ_COMMENT =
+ "# Cloud SST upload dead-letter queue – tab-separated: "
+ +
"failedAt\\tattemptCount\\tdbName\\tcfName\\tfilePath\\tremoteKey\\tlastError"
+ + "\\tsourceSstPath\\tuploadEpoch";
+
+ /** Max provider-unavailable reschedules before surfacing a terminal
local-only DLQ signal. */
+ private static final int DEFAULT_MAX_PROVIDER_UNAVAILABLE_RETRIES = 256;
+ /** Backoff cap for provider-unavailable reschedules (ms), independent of
upload-attempt cap. */
+ private static final long DEFAULT_PROVIDER_UNAVAILABLE_MAX_DELAY_MS =
30_000L;
+
+ // -----------------------------------------------------------------------
+ // Configuration
+ // -----------------------------------------------------------------------
+
+ private final int maxAttempts;
+ private final long initialDelayMs;
+ private final long maxDelayMs;
+ private volatile int maxProviderUnavailableRetries =
+ DEFAULT_MAX_PROVIDER_UNAVAILABLE_RETRIES;
+ private volatile long providerUnavailableMaxDelayMs =
+ DEFAULT_PROVIDER_UNAVAILABLE_MAX_DELAY_MS;
+
+ /**
+ * Invoked with {@code (dbName, sourceSstPath, uploadEpoch)} whenever an
upload succeeds via
+ * retry or DLQ replay, so the caller can mark the file confirmed-present
in cloud.
+ * The {@code uploadEpoch} is the {@link CloudSyncTracker} epoch captured
at submission time;
+ * the callback should use {@link CloudSyncTracker#markConfirmedIfEpoch}
to drop stale callbacks.
+ */
+ @FunctionalInterface
+ public interface UploadConfirmedCallback {
+ void onConfirmed(String dbName, String sourceSstPath, long
uploadEpoch);
+ }
+
+ private final UploadConfirmedCallback onUploadConfirmed;
+
+ /**
+ * Invoked with {@code dbName} after an upload becomes durable via retry
or DLQ replay, so the
+ * caller can (debounced) publish CURRENT/MANIFEST for that DB. Without
this, a retry that
+ * succeeds during a quiet period would leave the SST in cloud but the
mirrored CURRENT stale
+ * until some unrelated sync fires (or never, on an idle DB) — widening
the recovery point.
+ */
+ @FunctionalInterface
+ public interface MetadataSyncTrigger {
+ void onUploadDurable(String dbName);
+ }
+
+ /** Optional; set via {@link #setMetadataSyncTrigger} after the listener
is constructed. */
+ @Setter
+ private volatile MetadataSyncTrigger metadataSyncTrigger;
+
+ // -----------------------------------------------------------------------
+ // State
+ // -----------------------------------------------------------------------
+
+ /** In-memory dead-letter queue – holds tasks that exhausted all retries.
*/
+ private final Deque<FailedUploadTask> dlq = new ConcurrentLinkedDeque<>();
+
+ /**
+ * Retries that are scheduled but have not started running yet. A
scheduled task is added here
+ * before {@code scheduler.schedule(...)} and removed the moment it starts
executing. On
+ * {@link #close()}, a forced {@code shutdownNow()} silently discards the
executor's queued
+ * (delayed) tasks — but those wrappers are opaque, so we cannot recover
their payload from the
+ * returned runnables. Tracking the payload here lets us move any retry
that never ran into the
+ * DLQ, so retry intent (the only remaining upload path for a
compaction-raced file) is not lost.
+ */
+ private final Set<RetryContext> pendingRetries =
ConcurrentHashMap.newKeySet();
+
+ /**
+ * Retries that have STARTED executing but not yet finished. A task moves
from
+ * {@link #pendingRetries} to this set when it begins running, and is
removed when it completes
+ * (success, DLQ, or reschedule). On {@link #close()}, a retry still here
after the shutdown
+ * timeout is one hung in {@code provider.uploadFile(...)}; it is
persisted to the DLQ so its
+ * upload intent survives even a forced shutdown. DLQ replay is idempotent
(idempotent PUT +
+ * epoch-guarded confirmation), so a duplicate entry for a retry that
actually completed is safe.
+ */
+ private final Set<RetryContext> inFlightRetries =
ConcurrentHashMap.newKeySet();
+
+ /**
+ * Upper bound on DLQ entries (in memory and, amortized, on disk). Without
a cap, a prolonged
+ * provider outage with {@code maxAttempts == 0} routes every failed SST
upload straight to the
+ * DLQ, growing process memory and the on-disk file without bound (OOM /
disk exhaustion).
+ * When the cap is exceeded the oldest entries are evicted; those files
remain recoverable via
+ * the delete guard and startup SST backfill, which re-upload any live SST
missing from cloud.
+ */
+ static final int DEFAULT_MAX_DLQ_SIZE = 100_000;
+ private volatile int maxDlqSize = DEFAULT_MAX_DLQ_SIZE;
+
+ /** Count of DLQ entries evicted due to the size cap (monitoring / tests).
*/
+ private final AtomicInteger droppedDlqEntries = new AtomicInteger(0);
+
+ /** Appends since the last on-disk compaction, used to amortize {@link
#rewriteDlqFile}. */
+ private final AtomicInteger appendsSinceRewrite = new AtomicInteger(0);
+
+ /**
+ * DLQ on-disk persistence health. The in-memory DLQ is authoritative at
runtime, but the
+ * on-disk file is what survives a process restart. If a persist
(append/rewrite) fails, retry
+ * intent for the failed entry would be lost on a crash before the next
successful persist.
+ * We flip this to {@code false} and increment {@link
#dlqPersistenceFailures} so operators can
+ * alert on a "degraded durability" state via {@link
#isDlqPersistenceHealthy()} rather than
+ * treating a silently-swallowed IO error as healthy. It recovers to
{@code true} on the next
+ * successful persist. Live SSTs remain recoverable via the delete guard
and startup backfill,
+ * so this is a degraded signal, not hard data loss.
+ */
+ @Getter
+ private volatile boolean dlqPersistenceHealthy = true;
+
+ /** Cumulative count of DLQ persistence failures (monitoring / tests). */
+ private final AtomicInteger dlqPersistenceFailures = new AtomicInteger(0);
+
+ /** Absolute path of the on-disk DLQ file. */
+ private final Path dlqFile;
+
+ /** Background thread pool used to schedule retries. */
+ private final ScheduledExecutorService scheduler;
+
+ /** Counter of in-flight retry tasks (for testing / monitoring). */
+ private final AtomicInteger inFlightCount = new AtomicInteger(0);
+
+ /** Serialises all DLQ file mutations (append and rewrite) across
scheduler threads. */
+ private final ReentrantLock dlqFileLock = new ReentrantLock();
+
+ // -----------------------------------------------------------------------
+ // Constructor
+ // -----------------------------------------------------------------------
+
+ /**
+ * Fully-configured constructor. Prefer this over the {@link
#setMaxDlqSize(int)} setter when the
+ * cap is known at construction time: setting it here applies the cap
<em>before</em> the on-disk
+ * DLQ is loaded, so a large persisted file is bounded to the configured
cap during the load
+ * itself — avoiding the post-load trim-and-rewrite the setter must
perform.
+ *
+ * @param maxAttempts maximum whole-file upload retries after a
first failure.
+ * {@code 0} means no retries – failures go
directly to DLQ
+ * (provider is expected to handle its own
retries internally).
+ * Positive values enable exponential-backoff
whole-file retries.
+ * @param initialDelayMs delay before the first retry, in milliseconds;
clamped to ≥ 100.
+ * Ignored when {@code maxAttempts == 0}.
+ * @param maxDelayMs upper bound for exponential backoff delay;
clamped to
+ * ≥ {@code initialDelayMs}. Ignored when
{@code maxAttempts == 0}.
+ * @param dataRoot absolute path of the store's data directory –
the DLQ file is
+ * written here
+ * @param onUploadConfirmed epoch-aware callback invoked on every
successful retry/DLQ-replay
+ * upload; pass {@code null} for none.
+ * @param maxDlqSize maximum number of DLQ entries retained before
the oldest are evicted;
+ * a non-positive value falls back to {@link
#DEFAULT_MAX_DLQ_SIZE} so a
+ * misconfiguration cannot turn the DLQ into a
zero-capacity queue.
+ */
+ public CloudUploadRetryQueue(int maxAttempts, long initialDelayMs,
+ long maxDelayMs, String dataRoot,
+ UploadConfirmedCallback onUploadConfirmed,
int maxDlqSize) {
+ this.maxAttempts = Math.max(0, maxAttempts);
+ this.initialDelayMs = Math.max(100L, initialDelayMs);
+ this.maxDelayMs = Math.max(this.initialDelayMs, maxDelayMs);
+ this.onUploadConfirmed = onUploadConfirmed != null
+ ? onUploadConfirmed
+ : (db, path, epoch) -> { };
+ this.dlqFile = Paths.get(dataRoot, DLQ_FILE_NAME);
+ // Apply the cap before loadDlqFromDisk() so the load bounds the
in-memory DLQ (and the
+ // rewritten on-disk file) to the configured value directly. A
non-positive value keeps the
+ // default cap set at field initialisation, matching setMaxDlqSize()'s
guard.
+ if (maxDlqSize <= 0) {
+ log.warn("Ignoring non-positive DLQ max size {}; keeping default
{}",
+ maxDlqSize, this.maxDlqSize);
+ } else {
+ this.maxDlqSize = maxDlqSize;
+ }
+
+ java.util.concurrent.ScheduledThreadPoolExecutor sched =
+ new java.util.concurrent.ScheduledThreadPoolExecutor(2, r -> {
+ Thread t = new Thread(r, "cloud-upload-retry");
+ t.setDaemon(true);
+ return t;
+ });
+ // On shutdown(), cancel not-yet-due delayed retries instead of
holding the executor open
+ // until their (possibly minutes-long) delay elapses. close() then
rescues these pending
+ // retries into the DLQ via drainPendingRetriesToDlq(), so their
upload intent is preserved
+ // without a slow, pointless wait for retries that will never get to
run.
+ sched.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
+ this.scheduler = sched;
+
+ loadDlqFromDisk();
+ }
+
+ // -----------------------------------------------------------------------
+ // Public API
+ // -----------------------------------------------------------------------
+
+ /**
+ * Submits a failed upload to the retry queue or directly to the DLQ.
+ *
+ * <p>If {@code maxAttempts == 0} (opt-in mode, not the shipped default),
the task goes directly
+ * to the DLQ — the provider is expected to handle its own retries
internally.
+ *
+ * <p>If {@code maxAttempts > 0} (the default is {@code 3}), the first
retry is scheduled after
+ * {@code initialDelayMs} ms with exponential backoff. After all attempts
are exhausted the task
+ * is moved to the DLQ.
+ *
+ * <p>If the file no longer exists at retry time (compacted away by
RocksDB) the task is
+ * silently dropped.
+ *
+ * @param dbName RocksDB instance name (partition id)
+ * @param cfName column-family name
+ * @param filePath absolute local path of the SST file (also used for
confirmation)
+ * @param remoteKey remote object key (relative to bucket root)
+ * @param cause the exception from the first upload attempt
+ */
+ public void submit(String dbName, String cfName, String filePath,
+ String remoteKey, Exception cause) {
+ submitPinned(dbName, cfName, filePath, filePath, remoteKey, 0L, cause);
+ }
+
+ /**
+ * Epoch-aware variant of {@link #submit}. Callers that use an
epoch-guarded confirmation
+ * callback ({@link CloudSyncTracker#markConfirmedIfEpoch}) MUST use this
overload and pass the
+ * epoch captured before the upload was initiated. The plain {@link
#submit} hardcodes epoch
+ * {@code 0}, which never matches a live {@link CloudSyncTracker} epoch
(they start at 1), so a
+ * retried upload confirmed through it would be silently dropped.
+ *
+ * @param uploadEpoch the {@link CloudSyncTracker} epoch captured at
upload-initiation time
+ */
+ public void submit(String dbName, String cfName, String filePath,
+ String remoteKey, long uploadEpoch, Exception cause) {
+ submitPinned(dbName, cfName, filePath, filePath, remoteKey,
uploadEpoch, cause);
+ }
+
+ /**
+ * Submits a failed upload where the upload file is a hard-linked staging
copy that
+ * differs from the original source SST.
+ *
+ * <p>The retry will upload from {@code pinnedPath} (the staging
hard-link). On success,
+ * {@code onUploadConfirmed} is called with {@code sourceSstPath} (the
original {@code *.sst}
+ * path), and the staging file at {@code pinnedPath} is deleted.
+ *
+ * @param dbName RocksDB instance name (partition id)
+ * @param cfName column-family name
+ * @param pinnedPath path of the staging hard-link to upload from
+ * @param sourceSstPath original {@code *.sst} path used for confirmation
and cleanup
+ * @param remoteKey remote object key (relative to bucket root)
+ * @param cause the exception from the first upload attempt
+ */
+ public void submitPinned(String dbName, String cfName, String pinnedPath,
+ String sourceSstPath, String remoteKey, Exception
cause) {
+ submitPinned(dbName, cfName, pinnedPath, sourceSstPath, remoteKey, 0L,
cause);
+ }
+
+ /**
+ * Epoch-aware variant. The {@code uploadEpoch} is passed through to the
+ * {@link UploadConfirmedCallback} so that stale callbacks after DB
recreation are discarded.
+ */
+ public void submitPinned(String dbName, String cfName, String pinnedPath,
+ String sourceSstPath, String remoteKey,
+ long uploadEpoch, Exception cause) {
+ if (cause instanceof CloudStorageNonRetryableException || maxAttempts
== 0) {
+ moveToDlq(dbName, cfName, pinnedPath, sourceSstPath, remoteKey,
uploadEpoch, 0,
+ cause.getMessage() != null ? cause.getMessage() :
"non-retryable");
+ return;
+ }
+ // This method is invoked from RocksDB flush/compaction callbacks (via
+ // CloudStorageEventListener). It must never throw back across that
JNI/event boundary.
+ // When the scheduler is shutting down, scheduler.schedule(...) throws
+ // RejectedExecutionException — in that race we persist the task
directly to the DLQ
+ // (best-effort disk append) so the retry intent is not lost, and
return normally.
+ try {
+ scheduleRetry(dbName, cfName, pinnedPath, sourceSstPath,
remoteKey, uploadEpoch,
+ 1, 0);
+ } catch (RejectedExecutionException e) {
+ log.warn("Cloud upload retry scheduling rejected (scheduler
shutting down); "
+ + "persisting directly to DLQ: db={}, cf={}, path={}",
+ dbName, cfName, pinnedPath);
+ moveToDlq(dbName, cfName, pinnedPath, sourceSstPath, remoteKey,
uploadEpoch, 1,
+ "retry scheduling rejected during shutdown: "
+ + (cause.getMessage() != null ? cause.getMessage() :
cause.toString()));
+ }
+ }
+
+ /**
+ * Returns an unmodifiable snapshot of all current DLQ entries.
+ */
+ public List<FailedUploadTask> getDlqEntries() {
+ return List.copyOf(dlq);
+ }
+
+ /**
+ * Returns the number of entries currently in the DLQ.
+ */
+ public int getDlqSize() {
+ return dlq.size();
+ }
+
+ /**
+ * Returns the number of DLQ entries evicted because the DLQ size cap was
exceeded (e.g. during
+ * a prolonged provider outage). A non-zero value indicates local-only
files whose DLQ record
+ * was dropped; they remain recoverable via the delete guard and startup
SST backfill.
+ */
+ public int getDroppedDlqCount() {
+ return droppedDlqEntries.get();
+ }
+
+ /** Cumulative number of DLQ on-disk persistence failures since startup
(monitoring / tests). */
+ public int getDlqPersistenceFailureCount() {
+ return dlqPersistenceFailures.get();
+ }
+
+ /** Fires the metadata-sync trigger (if wired), swallowing any error — the
upload is durable. */
+ private void triggerMetadataSync(String dbName) {
+ MetadataSyncTrigger trigger = this.metadataSyncTrigger;
+ if (trigger == null) {
+ return;
+ }
+ try {
+ trigger.onUploadDurable(dbName);
+ } catch (Exception e) {
+ log.warn("Post-upload metadata-sync trigger failed for db={}
(upload is durable, "
+ + "CURRENT/MANIFEST will catch up on the next sync): {}",
dbName,
+ e.getMessage());
+ }
+ }
+
+ /**
+ * Sets the maximum number of DLQ entries retained before the oldest are
evicted. A
+ * non-positive value is ignored (keeps the current cap) so a
misconfiguration cannot turn the
+ * DLQ into a zero-capacity queue that drops every entry.
+ */
+ public void setMaxDlqSize(int maxSize) {
+ if (maxSize <= 0) {
+ log.warn("Ignoring non-positive DLQ max size {}; keeping {}",
maxSize, this.maxDlqSize);
+ return;
+ }
+ this.maxDlqSize = maxSize;
+ // The startup load bounds the in-memory DLQ to the DEFAULT cap (the
configured value is not
+ // known during construction). If the operator configured a SMALLER
cap, apply it now so a
+ // large persisted file loaded at startup is trimmed to the configured
bound and the on-disk
+ // file is rewritten to match — otherwise up to DEFAULT_MAX_DLQ_SIZE
stale entries could
+ // linger in memory and on disk despite a tighter configuration.
+ int evicted = 0;
+ while (dlq.size() > maxSize) {
+ if (dlq.pollFirst() == null) {
+ break;
+ }
+ evicted++;
+ }
+ if (evicted > 0) {
+ int total = droppedDlqEntries.addAndGet(evicted);
+ rewriteDlqFile();
+ log.warn("DLQ: max size lowered to {} — evicted {} oldest
entry(ies) to honor the "
+ + "configured cap (total dropped={}).", maxSize, evicted,
total);
+ }
+ }
+
+ /** Test seam: cap provider-unavailable reschedules so terminal behavior
is deterministic. */
+ @SuppressWarnings("SameParameterValue")
+ void setMaxProviderUnavailableRetriesForTest(int retries) {
+ this.maxProviderUnavailableRetries = Math.max(1, retries);
+ }
+
+ /** Test seam: speed up provider-unavailable retry cadence for
deterministic timing tests. */
+ @SuppressWarnings("SameParameterValue")
+ void setProviderUnavailableMaxDelayMsForTest(long maxDelayMs) {
+ this.providerUnavailableMaxDelayMs = Math.max(10L, maxDelayMs);
+ }
+
+ /**
+ * Returns the number of retry tasks currently scheduled or executing.
+ */
+ public int getInFlightCount() {
+ return inFlightCount.get();
+ }
+
+ /**
+ * Replays all DLQ entries immediately by attempting to re-upload each
file.
+ *
+ * <ul>
+ * <li>Entries that succeed are removed from the in-memory DLQ and the
DLQ file is
+ * rewritten without them.</li>
+ * <li>Entries whose local file no longer exists are silently
dropped.</li>
+ * <li>Entries that fail are re-submitted to the retry queue with a
fresh attempt cycle
+ * (they are removed from the DLQ for now and will re-enter it only
if all retries
+ * fail again).</li>
+ * </ul>
+ */
+ public void replayDlq() {
Review Comment:
Addressed
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
