[jira] [Created] (FLINK-40660) SplitFetcher hangs on shutdown when its element queue wakeUp flag is still set

["Martijn Visser (Jira)" <ji...@apache.org>]

Martijn Visser created FLINK-40660:
--------------------------------------

             Summary: SplitFetcher hangs on shutdown when its element queue wakeUp flag is still set
                 Key: FLINK-40660
                 URL: https://issues.apache.org/jira/browse/FLINK-40660
             Project: Flink
          Issue Type: Bug
          Components: Connectors / Common
            Reporter: Martijn Visser


{{SplitFetcher#run}} puts an empty synchronisation batch into the element queue after its loop ends
and discards the return value:

{code:java}
elementsQueue.put(
        fetcherId(),
        new RecordsBySplits<E>(Collections.emptyMap(), Collections.emptySet()) {
{code}

{{put}} returns false without enqueueing when the queue is full and that fetcher's wakeUp flag is
set. The batch is then dropped, {{recycle()}} never runs, {{recordsProcessedLatch.await()}} never
returns, the {{SplitReader}} is never closed and the shutdown hook never runs, so
{{fetchersToShutDown}} is never decremented and {{SplitFetcherManager#close}} blocks until its
timeout. {{maybeShutdownFinishedFetchers}} has already removed the fetcher from the map, so nothing
else can release it.

Reproduced on master by filling a capacity-1 queue, calling {{wakeUpPuttingThread}} for the fetcher's
own index and then {{shutdown(true)}}: the shutdown hook does not run, the reader is not closed and
the fetcher thread sits in WAITING. The flag is sticky, and the {{finally}} in {{FetchTask#run}}
clears only its own field, so a {{wakeUp}} that arrives just after the records were enqueued leaves
it set with nothing to consume it.

The put should be retried rather than dropped.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)