Remote log consumption is stuck because the semaphore permit cannot be acquired

[zuston]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

0.9.0 (latest release)

Please describe the bug 🐞

We have a Flink job that has been consuming a Fluss log table for about a week, but recently its throughput dropped to zero. After analyzing the subtask thread dump, it appears that remote log fetching is stuck because a semaphore permit cannot be acquired.

One subtask's thread dump is as follows

and then to use arthas to print the internal variable.

vmtool -c 312787f0 --action getInstances --className org.apache.fluss.client.table.scanner.log.LogFetchBuffer --limit 5 --express '
instances.length == 0 ? "no LogFetchBuffer" : (
  #buf = instances[0],
  #f = #buf.getClass().getDeclaredField("pendingFetches"),
  #f.setAccessible(true),
  #pmap = #f.get(#buf),
  #pmap.isEmpty() ? "pendingFetches empty" : (
    #firstList = #pmap.values().iterator().next(),
    #firstList.isEmpty() ? "first list empty" : #firstList.get(0).toString()
  )
)

and then to print some vars in the RemoteLogDownloader

vmtool -c 312787f0 --action getInstances --className org.apache.fluss.client.table.scanner.log.RemoteLogDownloader --limit 10 --express 'instances.length==0 ? "no RemoteLogDownloader" : (#d=instances[0],#c=#d.getClass(),#f1=#c.getDeclaredField("segmentsToFetch"),#f1.setAccessible(true),#f2=#c.getDeclaredField("segmentsToRecycle"),#f2.setAccessible(true),#f3=#c.getDeclaredField("prefetchSemaphore"),#f3.setAccessible(true),#m=new java.util.LinkedHashMap(),#m.put("segmentsToFetch_size",#f1.get(#d).size()),#m.put("segmentsToRecycle_size",#f2.get(#d).size()),#m.put("availablePermits",#f3.get(#d).availablePermits()),#m)'

And there's no any download failure logs in the taskmanager logs.

Solution

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[wguo-talend]

Solution : Add Defensive Semaphore Management per RemoteLogDownloader

void fetchOnce() throws Exception {
    prefetchSemaphore.acquire();
    boolean semaphoreReleased = false;
    
    try {
        RemoteLogDownloadRequest request = segmentsToFetch.poll(pollTimeout, TimeUnit.MILLISECONDS);
        if (request == null) {
            prefetchSemaphore.release();
            semaphoreReleased = true;
            return;
        }

        TableBucket tableBucket = request.getTableBucket();
        try {
            cleanupRemoteLogs();
            FsPathAndFileName fsPathAndFileName = request.getFsPathAndFileName();
            scannerMetricGroup.remoteFetchRequestCount().inc();

            long startTime = System.currentTimeMillis();
            remoteFileDownloader
                    .downloadFileAsync(fsPathAndFileName, localLogDir)
                    .exceptionally(throwable -> {
                        LOG.error("Failed to download remote log segment file {} for table bucket {}.",
                                fsPathAndFileName.getFileName(), tableBucket, throwable);
                        prefetchSemaphore.release();
                        segmentsToFetch.add(request);
                        scannerMetricGroup.remoteFetchErrorCount().inc();
                        return null;
                    })
                    .thenAccept(bytes -> {
                        LOG.info("Successfully downloaded remote log segment file {} to local for table bucket {} cost {} ms.",
                                fsPathAndFileName.getFileName(), tableBucket, 
                                System.currentTimeMillis() - startTime);
                        File localFile = new File(localLogDir.toFile(), fsPathAndFileName.getFileName());
                        scannerMetricGroup.remoteFetchBytes().inc(bytes);
                        request.future.complete(localFile);
                    });
            semaphoreReleased = true;
        } catch (Throwable t) {
            prefetchSemaphore.release();
            semaphoreReleased = true;
            segmentsToFetch.add(request);
            scannerMetricGroup.remoteFetchErrorCount().inc();
            LOG.error("Failed to download remote log segment for table bucket {}.", tableBucket, t);
        }
    } finally {
        if (!semaphoreReleased) {
            prefetchSemaphore.release();
        }
    }
}

[wuchong]

@zuston could you also open a pull request for release-0.9 branch?