You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "loserwang1024 (via GitHub)" <gi...@apache.org> on 2025/09/24 11:50:47 UTC
[PR] [client] Throw Exception when retry several times rather than release semaphore to avoid deadlock. [fluss]
loserwang1024 opened a new pull request, #1752:
URL: https://github.com/apache/fluss/pull/1752
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #1751
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [client] Throw Exception when retry several times rather than release semaphore to avoid deadlock. [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #1752:
URL: https://github.com/apache/fluss/pull/1752#issuecomment-3344640845
@wuchong , CC
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [client] Throw Exception when retry several times when downloading log. [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #1752:
URL: https://github.com/apache/fluss/pull/1752#issuecomment-5749661896
@swuferhong , I have rebased this PR again
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [client] Throw Exception when retry several times when downloading log. [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #1752:
URL: https://github.com/apache/fluss/pull/1752#issuecomment-4705143585
@swuferhong , I have rebased this PR, CC
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [client] Throw Exception when retry several times when downloading log. [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #1752:
URL: https://github.com/apache/fluss/pull/1752#issuecomment-4167287643
@wuchong , CC, it's important for me.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [client] Throw Exception when retry several times when downloading log. [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on code in PR #1752:
URL: https://github.com/apache/fluss/pull/1752#discussion_r4059173148
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloader.java:
##########
@@ -62,6 +62,7 @@ public class RemoteLogDownloader implements Closeable {
private static final Logger LOG = LoggerFactory.getLogger(RemoteLogDownloader.class);
private static final long POLL_TIMEOUT = 5000L;
+ private static final int MAX_RETRY_COUNT = 5;
Review Comment:
would be better to expose this as a ConfigOption?
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloader.java:
##########
@@ -236,16 +233,49 @@ void fetchOnce() throws Exception {
}
});
} catch (Throwable t) {
- prefetchSemaphore.release();
- // only re-queue the request if the downloader is still active
- if (!closed && !request.future.isCancelled()) {
- segmentsToFetch.add(request);
- } else if (!request.future.isDone()) {
+ if (closed || request.future.isCancelled()) {
+ if (!request.future.isDone()) {
+ request.future.cancel(false);
+ }
+ prefetchSemaphore.release();
+ if (closed) {
+ deleteDirectoryQuietly(localLogDir.toFile());
+ }
+ } else {
+ handleFetchException(request, t, retryCount, startTime);
+ }
+ }
+ }
+
+ private void handleFetchException(
+ RemoteLogDownloadRequest request, Throwable throwable, int retryCount, long startTime) {
+ if (closed || request.future.isCancelled()) {
+ if (!request.future.isDone()) {
request.future.cancel(false);
}
- scannerMetricGroup.remoteFetchErrorCount().inc();
- // log the error and continue instead of shutdown the download thread
- LOG.error("Failed to download remote log segment for table bucket {}.", tableBucket, t);
+ prefetchSemaphore.release();
+ if (closed) {
+ deleteDirectoryQuietly(localLogDir.toFile());
+ }
+ return;
+ }
+
+ LOG.error(
+ "Failed to download remote log segment file {} for table bucket {}.",
+ request.getFsPathAndFileName().getFileName(),
+ request.getTableBucket(),
+ ExceptionUtils.stripExecutionException(throwable));
+ scannerMetricGroup.remoteFetchErrorCount().inc();
+ if (retryCount >= 1) {
Review Comment:
The `5` retries in `handleFetchException` are fired back-to-back with no delay/backoff between attempts. For a permanent error (e.g. the remote file does not exist), this just burns 6 immediate attempts before failing — acceptable,
since it fails fast. But for transient errors (network blips, object-store throttling), 6 retries within a few milliseconds are likely to all fail, so we give up and trigger a failover before the transient condition has a chance to recover, which weakens the self-healing that retrying is supposed to provide. Could we add a (possibly exponential) backoff between retries?
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [client] Throw Exception when retry several times when downloading log. [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #1752:
URL: https://github.com/apache/fluss/pull/1752#discussion_r4062272881
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloader.java:
##########
@@ -236,16 +233,49 @@ void fetchOnce() throws Exception {
}
});
} catch (Throwable t) {
- prefetchSemaphore.release();
- // only re-queue the request if the downloader is still active
- if (!closed && !request.future.isCancelled()) {
- segmentsToFetch.add(request);
- } else if (!request.future.isDone()) {
+ if (closed || request.future.isCancelled()) {
+ if (!request.future.isDone()) {
+ request.future.cancel(false);
+ }
+ prefetchSemaphore.release();
+ if (closed) {
+ deleteDirectoryQuietly(localLogDir.toFile());
+ }
+ } else {
+ handleFetchException(request, t, retryCount, startTime);
+ }
+ }
+ }
+
+ private void handleFetchException(
+ RemoteLogDownloadRequest request, Throwable throwable, int retryCount, long startTime) {
+ if (closed || request.future.isCancelled()) {
+ if (!request.future.isDone()) {
request.future.cancel(false);
}
- scannerMetricGroup.remoteFetchErrorCount().inc();
- // log the error and continue instead of shutdown the download thread
- LOG.error("Failed to download remote log segment for table bucket {}.", tableBucket, t);
+ prefetchSemaphore.release();
+ if (closed) {
+ deleteDirectoryQuietly(localLogDir.toFile());
+ }
+ return;
+ }
+
+ LOG.error(
+ "Failed to download remote log segment file {} for table bucket {}.",
+ request.getFsPathAndFileName().getFileName(),
+ request.getTableBucket(),
+ ExceptionUtils.stripExecutionException(throwable));
+ scannerMetricGroup.remoteFetchErrorCount().inc();
+ if (retryCount >= 1) {
Review Comment:
done it.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [client] Throw Exception when retry several times when downloading log. [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #1752:
URL: https://github.com/apache/fluss/pull/1752#issuecomment-4009851281
@wuchong , fix this test
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [client] Throw Exception when retry several times when downloading log. [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 merged PR #1752:
URL: https://github.com/apache/fluss/pull/1752
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [client] Throw Exception when retry several times when downloading log. [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1752:
URL: https://github.com/apache/fluss/pull/1752#discussion_r2877068773
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloader.java:
##########
@@ -151,51 +152,56 @@ void fetchOnce() throws Exception {
return;
}
- try {
- // 1. cleanup the finished logs first to free up disk space
- cleanupRemoteLogs();
+ // the semaphore will only be released after the remote file are downloaded to local
+ // successfully.
+ downloadRemoteLog(request, MAX_RETRY_COUNT, System.currentTimeMillis());
+ }
- // 2. do the actual download work
- FsPathAndFileName fsPathAndFileName = request.getFsPathAndFileName();
- scannerMetricGroup.remoteFetchRequestCount().inc();
-
- long startTime = System.currentTimeMillis();
- // download the remote file to local
- remoteFileDownloader
- .downloadFileAsync(fsPathAndFileName, localLogDir)
- .whenComplete(
- (bytes, throwable) -> {
- if (throwable != null) {
- LOG.error(
- "Failed to download remote log segment file {}.",
- fsPathAndFileName.getFileName(),
- ExceptionUtils.stripExecutionException(throwable));
- // release the semaphore for the failed request
- prefetchSemaphore.release();
- // add back the request to the queue,
- // so we do not complete the request.future here
- segmentsToFetch.add(request);
- scannerMetricGroup.remoteFetchErrorCount().inc();
- } else {
- LOG.info(
- "Successfully downloaded remote log segment file {} to local cost {} ms.",
- fsPathAndFileName.getFileName(),
- System.currentTimeMillis() - startTime);
- File localFile =
- new File(
- localLogDir.toFile(),
- fsPathAndFileName.getFileName());
- scannerMetricGroup.remoteFetchBytes().inc(bytes);
- request.future.complete(localFile);
- }
- });
- } catch (Throwable t) {
- prefetchSemaphore.release();
- // add back the request to the queue
- segmentsToFetch.add(request);
- scannerMetricGroup.remoteFetchErrorCount().inc();
- // log the error and continue instead of shutdown the download thread
- LOG.error("Failed to download remote log segment.", t);
+ private void downloadRemoteLog(
+ RemoteLogDownloadRequest request, int retryCount, long startTime) {
+ // 1. cleanup the finished logs first to free up disk space
+ cleanupRemoteLogs();
+
+ // 2. do the actual download work
+ FsPathAndFileName fsPathAndFileName = request.getFsPathAndFileName();
+ scannerMetricGroup.remoteFetchRequestCount().inc();
+ remoteFileDownloader
+ .downloadFileAsync(fsPathAndFileName, localLogDir)
+ .whenComplete(
+ (bytes, throwable) -> {
+ if (throwable != null) {
+ handleFetchException(request, throwable, retryCount, startTime);
+ } else {
+ LOG.info(
+ "Successfully downloaded remote log segment file {} to local cost {} ms.",
+ fsPathAndFileName.getFileName(),
+ System.currentTimeMillis() - startTime);
+ File localFile =
+ new File(
+ localLogDir.toFile(),
+ fsPathAndFileName.getFileName());
+ scannerMetricGroup.remoteFetchBytes().inc(bytes);
+ request.future.complete(localFile);
+ }
+ });
Review Comment:
`downloadRemoteLog` no longer guards against synchronous exceptions from `cleanupRemoteLogs()` / `downloadFileAsync(...)` (e.g., bad path / FS init failure). If an exception is thrown before the `whenComplete` is attached, it will bubble out of `fetchOnce()` and can terminate the download thread without completing `request.future`, leaving callers blocked. Wrap the body in a try/catch and route failures through the same error-handling path (completing the future exceptionally after retries).
```suggestion
try {
// 1. cleanup the finished logs first to free up disk space
cleanupRemoteLogs();
// 2. do the actual download work
FsPathAndFileName fsPathAndFileName = request.getFsPathAndFileName();
scannerMetricGroup.remoteFetchRequestCount().inc();
remoteFileDownloader
.downloadFileAsync(fsPathAndFileName, localLogDir)
.whenComplete(
(bytes, throwable) -> {
if (throwable != null) {
handleFetchException(
request, throwable, retryCount, startTime);
} else {
LOG.info(
"Successfully downloaded remote log segment file {} to local cost {} ms.",
fsPathAndFileName.getFileName(),
System.currentTimeMillis() - startTime);
File localFile =
new File(
localLogDir.toFile(),
fsPathAndFileName.getFileName());
scannerMetricGroup.remoteFetchBytes().inc(bytes);
request.future.complete(localFile);
}
});
} catch (Throwable t) {
handleFetchException(request, t, retryCount, startTime);
}
```
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloader.java:
##########
@@ -151,51 +152,56 @@ void fetchOnce() throws Exception {
return;
}
- try {
- // 1. cleanup the finished logs first to free up disk space
- cleanupRemoteLogs();
+ // the semaphore will only be released after the remote file are downloaded to local
+ // successfully.
+ downloadRemoteLog(request, MAX_RETRY_COUNT, System.currentTimeMillis());
+ }
- // 2. do the actual download work
- FsPathAndFileName fsPathAndFileName = request.getFsPathAndFileName();
- scannerMetricGroup.remoteFetchRequestCount().inc();
-
- long startTime = System.currentTimeMillis();
- // download the remote file to local
- remoteFileDownloader
- .downloadFileAsync(fsPathAndFileName, localLogDir)
- .whenComplete(
- (bytes, throwable) -> {
- if (throwable != null) {
- LOG.error(
- "Failed to download remote log segment file {}.",
- fsPathAndFileName.getFileName(),
- ExceptionUtils.stripExecutionException(throwable));
- // release the semaphore for the failed request
- prefetchSemaphore.release();
- // add back the request to the queue,
- // so we do not complete the request.future here
- segmentsToFetch.add(request);
- scannerMetricGroup.remoteFetchErrorCount().inc();
- } else {
- LOG.info(
- "Successfully downloaded remote log segment file {} to local cost {} ms.",
- fsPathAndFileName.getFileName(),
- System.currentTimeMillis() - startTime);
- File localFile =
- new File(
- localLogDir.toFile(),
- fsPathAndFileName.getFileName());
- scannerMetricGroup.remoteFetchBytes().inc(bytes);
- request.future.complete(localFile);
- }
- });
- } catch (Throwable t) {
- prefetchSemaphore.release();
- // add back the request to the queue
- segmentsToFetch.add(request);
- scannerMetricGroup.remoteFetchErrorCount().inc();
- // log the error and continue instead of shutdown the download thread
- LOG.error("Failed to download remote log segment.", t);
+ private void downloadRemoteLog(
+ RemoteLogDownloadRequest request, int retryCount, long startTime) {
+ // 1. cleanup the finished logs first to free up disk space
+ cleanupRemoteLogs();
+
+ // 2. do the actual download work
+ FsPathAndFileName fsPathAndFileName = request.getFsPathAndFileName();
+ scannerMetricGroup.remoteFetchRequestCount().inc();
+ remoteFileDownloader
+ .downloadFileAsync(fsPathAndFileName, localLogDir)
+ .whenComplete(
+ (bytes, throwable) -> {
+ if (throwable != null) {
+ handleFetchException(request, throwable, retryCount, startTime);
+ } else {
+ LOG.info(
+ "Successfully downloaded remote log segment file {} to local cost {} ms.",
+ fsPathAndFileName.getFileName(),
+ System.currentTimeMillis() - startTime);
+ File localFile =
+ new File(
+ localLogDir.toFile(),
+ fsPathAndFileName.getFileName());
+ scannerMetricGroup.remoteFetchBytes().inc(bytes);
+ request.future.complete(localFile);
+ }
+ });
+ }
+
+ private void handleFetchException(
+ RemoteLogDownloadRequest request, Throwable throwable, int retryCount, long startTime) {
+ LOG.error(
+ "Failed to download remote log segment file {}.",
+ request.getFsPathAndFileName().getFileName(),
+ ExceptionUtils.stripExecutionException(throwable));
+ scannerMetricGroup.remoteFetchErrorCount().inc();
+ if (retryCount <= 0) {
+ downloadRemoteLog(request, retryCount - 1, startTime);
+ } else {
+ request.future.completeExceptionally(
+ new IOException(
+ String.format(
+ "Failed to download remote log segment file %s, retry count %d",
+ request.getFsPathAndFileName().getFileName(), MAX_RETRY_COUNT),
+ throwable));
Review Comment:
`handleFetchException` has the retry condition inverted: with `MAX_RETRY_COUNT` (5) the code immediately completes the future exceptionally and never retries. Also, the current `if (retryCount <= 0) downloadRemoteLog(request, retryCount - 1, ...)` path would recurse forever with negative retry counts if it were ever reached. Flip the condition to retry while `retryCount > 0` (decrementing each attempt), and complete exceptionally once retries are exhausted (without scheduling another download with a negative count).
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchBuffer.java:
##########
@@ -109,10 +113,20 @@ void tryComplete(TableBucket tableBucket) {
while (pendings != null && !pendings.isEmpty()) {
PendingFetch pendingFetch = pendings.peek();
if (pendingFetch.isCompleted()) {
- CompletedFetch completedFetch = pendingFetch.toCompletedFetch();
- completedFetches.add(completedFetch);
- pendings.poll();
- hasCompleted = true;
+ try {
+ CompletedFetch completedFetch = pendingFetch.toCompletedFetch();
+ completedFetches.add(completedFetch);
+ pendings.poll();
+ hasCompleted = true;
+ } catch (Throwable t) {
+ LOG.error(
+ "Failed to complete fetch for tableBucket: {}",
+ tableBucket,
+ t);
+ throwable = t;
Review Comment:
In `tryComplete`, if `pendingFetch.toCompletedFetch()` throws, you set `throwable = t` and return without signaling `notEmptyCondition`. Any thread blocked in `awaitNotEmpty()` will continue waiting indefinitely (since `pend()` does not signal and no completed fetch was added). Signal the condition when recording `throwable` so waiters can wake up and observe the failure via `peek()`/`poll()`.
```suggestion
throwable = t;
notEmptyCondition.signalAll();
```
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchBuffer.java:
##########
@@ -73,17 +75,19 @@ public class LogFetchBuffer implements AutoCloseable {
@GuardedBy("lock")
private @Nullable CompletedFetch nextInLineFetch;
+ @GuardedBy("lock")
+ private @Nullable Throwable throwable;
+
public LogFetchBuffer() {
this.completedFetches = new LinkedList<>();
}
/**
- * Returns {@code true} if there are no completed fetches pending to return to the user.
- *
- * @return {@code true} if the buffer is empty, {@code false} otherwise
+ * @return {@code true} if there are no completed fetches pending to return to the user or some
+ * error occurs, {@code false} otherwise
Review Comment:
The `isEmpty()` JavaDoc is now inconsistent with the implementation. The method returns `false` when `throwable != null` (treating an error like “not empty” so it can be surfaced), but the JavaDoc says it returns `true` when “some error occurs”. Update the JavaDoc to match the actual semantics (or adjust the method if the JavaDoc is intended).
```suggestion
* @return {@code true} if there are no completed fetches pending to return to the user and no
* error has been recorded, {@code false} otherwise
```
##########
fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloaderTest.java:
##########
@@ -282,6 +284,52 @@ void testOrderOfRemoteLogDownloadRequest() {
assertThat(results).isEqualTo(expected);
}
+ @Test
+ void testFetchException() {
+ RemoteFileDownloader remoteFileDownloader = new RemoteFileDownloader(1);
+ RemoteLogDownloader remoteLogDownloader =
+ new RemoteLogDownloader(
+ DATA1_TABLE_PATH, conf, remoteFileDownloader, scannerMetricGroup, 10L);
+ try {
+ // trigger auto download.
+ remoteLogDownloader.start();
+
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ FsPath fsPath = new FsPath("test:///empty");
+ RemoteLogSegment nonExistLogSegment =
+ RemoteLogSegment.Builder.builder()
+ .tableBucket(tb)
+ .physicalTablePath(DATA1_PHYSICAL_TABLE_PATH)
+ .remoteLogSegmentId(UUID.randomUUID())
+ .remoteLogStartOffset(1)
+ .remoteLogEndOffset(2)
+ .maxTimestamp(2)
+ .segmentSizeInBytes(Integer.MAX_VALUE)
+ .build();
+
+ RemoteLogDownloadFuture remoteLogDownloadFuture =
+ remoteLogDownloader.requestRemoteLog(fsPath, nonExistLogSegment);
+ retry(
+ Duration.ofMinutes(1),
+ () -> assertThat(remoteLogDownloadFuture.isDone()).isTrue());
+ AbstractThrowableAssert<?, ?> exactlyInstanceOf =
+ assertThatThrownBy(() -> remoteLogDownloadFuture.getFileLogRecords(1))
+ .cause()
+ .isExactlyInstanceOf(IOException.class)
+ .hasMessageContaining(
+ String.format(
+ "Failed to download remote log segment file %s, retry count 5",
+ RemoteLogDownloader.getFsPathAndFileName(
+ fsPath, nonExistLogSegment)
+ .getFileName()))
+ .rootCause();
+
Review Comment:
`exactlyInstanceOf` is assigned but never used, which adds noise to the test and may trigger unused-variable warnings in some builds. Remove the local variable (and `AbstractThrowableAssert` import if it becomes unused).
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java:
##########
@@ -161,7 +162,7 @@ public boolean hasAvailableFetches() {
return !logFetchBuffer.isEmpty();
}
- public Map<TableBucket, List<ScanRecord>> collectFetch() {
+ public Map<TableBucket, List<ScanRecord>> collectFetch() throws FetchException {
return logFetchCollector.collectFetch(logFetchBuffer);
}
Review Comment:
After moving to “fail fast” semantics (completing the download future exceptionally), the existing remote-download completion callback in `pendRemoteFetches` is still registered via `RemoteLogDownloadFuture.onComplete(...)`, which currently only runs on *successful* completion (`thenRun`). This means a failed remote download may never trigger `logFetchBuffer.tryComplete(...)`, so the exception can be silently stuck in the pending queue and the scanner may hang instead of failing. Consider making the callback run for both success and failure (e.g., use a `whenComplete`-style hook) so `tryComplete` can surface the failure via the new exception path.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [client] Throw Exception when retry several times when downloading log. [fluss]
Posted by "Jackeyzhe (via GitHub)" <gi...@apache.org>.
Jackeyzhe commented on PR #1752:
URL: https://github.com/apache/fluss/pull/1752#issuecomment-5496163004
## Review: PR #1752 — remote-log download retry / exception propagation
Thanks for picking this up — surfacing the download error to the consumer instead of retrying
forever is the right call, and the `LogFetchBuffer` error-propagation design (`peek()`/`poll()`
throw `FetchException`) is clean. I left one correctness concern and a couple of nits.
### 🔴 [must-fix] Semaphore permit leak on permanent download failure
The permit accounting changed in a way that can **stall the whole remote-log downloader** after
`CLIENT_SCANNER_REMOTE_LOG_PREFETCH_NUM` permanently-failing segments.
Background: `prefetchSemaphore` is released only in two places — (a) the `closed`/`early-return`
branches, and (b) `RemoteCompletedFetch.drain()` → `recycleCallback.run()` → `recycleRemoteLog()`
(line 140). i.e. the permit is returned when the **consumer recycles the successfully-downloaded
file**, not when the download finishes. That matches the javadoc "blocks until there is capacity
(the fetched file is consumed)".
On the **failure** path this breaks:
- `main` released the permit immediately on every failure (`prefetchSemaphore.release(); segmentsToFetch.add(request);`) — see the removed lines in the diff.
- The new `handleFetchException(...)` does **not** release the permit. On exhaustion it only calls `request.future.completeExceptionally(...)`.
- But the exceptional `RemoteCompletedFetch` is never drained: `LogFetchBuffer.poll()` calls `checkException()` and throws `FetchException` *before* the consumer reaches `drain()`, so `recycleCallback` is never run → the permit is never returned.
Net effect: each segment whose remote log is permanently unreadable (exactly the scenario this PR
wants to surface) leaks one permit. After `PREFETCH_NUM` such segments, `fetchOnce()` blocks
forever at `prefetchSemaphore.acquire()` and the scanner's remote-log prefetch halts.
Suggested fix — release the permit on the **exhaustion** branch (mirror `main`'s immediate
release), but keep holding it while retries are still in flight:
```java
private void handleFetchException(
RemoteLogDownloadRequest request, Throwable throwable, int retryCount, long startTime) {
LOG.error(...);
scannerMetricGroup.remoteFetchErrorCount().inc();
if (retryCount >= 1) {
downloadRemoteLog(request, retryCount - 1, startTime); // permit still held until success-recycle
} else {
prefetchSemaphore.release(); // <-- return the permit; no file to recycle
request.future.completeExceptionally(
new IOException(
String.format("Failed to download remote log segment file %s, retry count %d",
request.getFsPathAndFileName().getFileName(), MAX_RETRY_COUNT),
throwable));
}
}
```
Worth adding a test that asserts the semaphore count returns to full after N permanent failures
(the current `RemoteLogDownloaderTest` manually calls `getRecycleCallback().run()`, which papers
over this path).
### 🟡 [nit] No backoff between retries
`handleFetchException` re-issues the download synchronously (chained through the future callback)
with zero delay. For a transient error that needs the remote FS to recover, 5 tight retries may
all fail in microseconds. Consider a small backoff (or reuse the existing retry config) so retries
actually help with transient blips. Not blocking.
### 🟡 [nit] `MAX_RETRY_COUNT = 5` is a hard-coded magic constant
For a 1.0 stability knob it'd be nicer to expose it as a config option (with a sensible default),
so operators can tune it without recompiling. Optional.
### ✅ Tests
`LogFetcherTest` (+95) and `RemoteLogDownloaderTest` (+47) cover the new behavior well. The only
gap is the permit-leak path above — please extend the test as suggested.
Overall this is a good direction; the permit fix is the one thing I'd want resolved before merge.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org