[
https://issues.apache.org/jira/browse/FLINK-40660?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40660:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Affects Versions: 2.0.2, 2.3.0, 2.2.1, 2.1.3
> Reporter: Martijn Visser
> Priority: Major
> Labels: pull-request-available
>
> {{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)
