[
https://issues.apache.org/jira/browse/HIVE-30245?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Denys Kuzmenko updated HIVE-30245:
----------------------------------
Description:
HIVE-27899 makes a task attempt wait for the AM's permission to commit before
it closes its outputs:
{code:java}
while (!getContext().canCommit()) {
Thread.sleep(500);
}
{code}
The wait ends only when canCommit() returns true or the sleep throws
InterruptedException. A killed attempt is no longer running, so canCommit()
stays false, and the kill's interrupt is the only way out.
A kill (TezTaskRunner2: a kill request, a preemption, a shutdown) first aborts
the processor and then interrupts the task thread once. If run() consumes that
interrupt and returns because of the abort, the attempt enters the wait with no
interrupt pending and never leaves it. In LLAP it keeps its executor until the
daemon restarts. The query itself succeeds, so nothing fails; the cluster just
has fewer executors.
h3. How it happens
Observed on TPC-DS 10 TB (Parquet on S3, LLAP, Hadoop 3.4.3, the native Parquet
cache of HIVE-30059):
# LLAP preempts a running Map attempt. TezTaskRunner2 aborts it, then
interrupts it.
# The thread is reading a Parquet footer through the native cache
(LlapParquetIoImpl.getFooterBuffersFromCache <-
ParquetEncodedDataReader.readFooter) on the S3A analytics stream, the default
since Hadoop 3.4.3.
# analytics-accelerator-s3 1.3.1 (Block.awaitData) catches the
InterruptedException and throws IOException("Error while reading data. Read
interrupted while waiting for data"), without restoring the interrupt flag.
# ParquetColumnVectorProducer catches the IOException and falls back to the
normal reader ("Parquet native cache: falling back to normal reader").
# run() returns on the abort, and the attempt sits in the canCommit loop
forever, while TezTaskRunner2 logs "returning canCommit=false since task is not
in a running state" twice a second.
After one run of the 99 queries, 17 of 144 executors were held this way, all of
them q16 Map attempts sleeping in TezProcessor.initializeAndRunProcessor. After
another run, 22 were held.
Any code that swallows an interrupt during run() leads to the same state, so
the wait must not depend on the interrupt alone.
h3. Fix
The processor is marked aborted before the kill interrupts it. The wait now
checks that too: an aborted attempt throws InterruptedException("Aborted while
waiting to commit"). That is the same path a delivered interrupt takes: the
record processor is aborted and closed, the outputs are aborted, and nothing is
committed.
h3. Test
TestTezProcessor#testKilledTaskStopsWaitingToCommit: canCommit() returns false,
and run() aborts the processor and returns without an interrupt. Without the
fix the test times out; with it, initializeAndRunProcessor throws
InterruptedException.
> TezProcessor: a killed task attempt whose interrupt was consumed waits
> forever to commit
> ----------------------------------------------------------------------------------------
>
> Key: HIVE-30245
> URL: https://issues.apache.org/jira/browse/HIVE-30245
> Project: Hive
> Issue Type: Bug
> Reporter: Denys Kuzmenko
> Priority: Major
>
> HIVE-27899 makes a task attempt wait for the AM's permission to commit before
> it closes its outputs:
> {code:java}
> while (!getContext().canCommit()) {
> Thread.sleep(500);
> }
> {code}
> The wait ends only when canCommit() returns true or the sleep throws
> InterruptedException. A killed attempt is no longer running, so canCommit()
> stays false, and the kill's interrupt is the only way out.
> A kill (TezTaskRunner2: a kill request, a preemption, a shutdown) first
> aborts the processor and then interrupts the task thread once. If run()
> consumes that interrupt and returns because of the abort, the attempt enters
> the wait with no interrupt pending and never leaves it. In LLAP it keeps its
> executor until the daemon restarts. The query itself succeeds, so nothing
> fails; the cluster just has fewer executors.
> h3. How it happens
> Observed on TPC-DS 10 TB (Parquet on S3, LLAP, Hadoop 3.4.3, the native
> Parquet cache of HIVE-30059):
> # LLAP preempts a running Map attempt. TezTaskRunner2 aborts it, then
> interrupts it.
> # The thread is reading a Parquet footer through the native cache
> (LlapParquetIoImpl.getFooterBuffersFromCache <-
> ParquetEncodedDataReader.readFooter) on the S3A analytics stream, the default
> since Hadoop 3.4.3.
> # analytics-accelerator-s3 1.3.1 (Block.awaitData) catches the
> InterruptedException and throws IOException("Error while reading data. Read
> interrupted while waiting for data"), without restoring the interrupt flag.
> # ParquetColumnVectorProducer catches the IOException and falls back to the
> normal reader ("Parquet native cache: falling back to normal reader").
> # run() returns on the abort, and the attempt sits in the canCommit loop
> forever, while TezTaskRunner2 logs "returning canCommit=false since task is
> not in a running state" twice a second.
> After one run of the 99 queries, 17 of 144 executors were held this way, all
> of them q16 Map attempts sleeping in TezProcessor.initializeAndRunProcessor.
> After another run, 22 were held.
> Any code that swallows an interrupt during run() leads to the same state, so
> the wait must not depend on the interrupt alone.
> h3. Fix
> The processor is marked aborted before the kill interrupts it. The wait now
> checks that too: an aborted attempt throws InterruptedException("Aborted
> while waiting to commit"). That is the same path a delivered interrupt takes:
> the record processor is aborted and closed, the outputs are aborted, and
> nothing is committed.
> h3. Test
> TestTezProcessor#testKilledTaskStopsWaitingToCommit: canCommit() returns
> false, and run() aborts the processor and returns without an interrupt.
> Without the fix the test times out; with it, initializeAndRunProcessor throws
> InterruptedException.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
