[
https://issues.apache.org/jira/browse/HUDI-4741?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
voon updated HUDI-4741:
-----------------------
Reviewers: Teng Huo (was: Teng Huo, 冯健)
> Deadlock when restarting failed TM in AbstractStreamWriteFunction
> -----------------------------------------------------------------
>
> Key: HUDI-4741
> URL: https://issues.apache.org/jira/browse/HUDI-4741
> Project: Apache Hudi
> Issue Type: Bug
> Components: flink
> Reporter: voon
> Assignee: voon
> Priority: Major
> Fix For: 0.12.1
>
>
> h1. Summary of Events
> # TM heartbeat not sent to JM (Can be triggered by killing a container), JM
> kills the TM/container
> # JM restarts the container, but the restarting code is not handled
> properly, causing there to be a deadlock
> # Deadlock causes instantToWrite() to loop for 10 minutes (default Flink
> checkpoint timeout), causing a instant initialization timeout error
> # JM is restarted
> # JM restore state from previously successful checkpoint
> # Ckp metadata path is tainted with multiple {_}INFLIGHT{_}s
> # Synchronisation issue occurs if TM executes *`lastPendingInstant()`*
> before JM executes *`startInstant()`*
> # Single commit multi instant issue occurs
> # When tainted TM reads obtained the wrong _INFLIGHT_ instant, it will start
> a new write cycle with the correct _INFLIGHT_ instant while a checkpoint is
> being performed.
> # *`reconcileAgainstMarkers()`* will delete files that tainted TMs are
> writing to with the correct _INFLIGHT_ instant in the next cycle, causing
> FileNotFoundException, _COLUMN_ state errors and parquet corruption errors.
>
> h1. Code for reproducing
> h2. Flink SQL Code
> {code:java}
> CREATE TABLE input_table (
> `val` STRING
> ,`event_time` TIMESTAMP(3)
> ,`partition` BIGINT
> ,`offset` BIGINT
> ) WITH (
> 'connector' = 'datagen',
> 'fields.val.length' = '99999',
> 'rows-per-second' = '15000'
> );CREATE TABLE test_hudi
> (
> `val` STRING
> ,`event_time` TIMESTAMP(3)
> ,`partition` BIGINT
> ,`offset` BIGINT
> ,`dt` STRING
> ,`hh` STRING
> ) PARTITIONED BY (dt, hh)
> WITH (
> 'connector' = 'hudi',
> 'path' = 'hdfs://jm_tm_sync_error/',
> 'table.type' = 'COPY_ON_WRITE',
> 'write.operation' = 'insert',
> 'hoodie.parquet.small.file.limit' = '104857600',
> 'hoodie.parquet.max.file.size' = '268435456',
> 'hoodie.datasource.write.recordkey.field' = 'partition,offset',
> 'hoodie.datasource.write.hive_style_partitioning' = 'true',
> 'hoodie.datasource.write.partitionpath.field' = 'dt,hh',
> 'write.bulk_insert.sort_input' = 'false',
> 'index.bootstrap.enabled' = 'false',
> 'index.state.ttl' = '60',
> 'index.type' = 'FLINK_STATE',
> 'hoodie.datasource.write.keygenerator.class' =
> 'org.apache.hudi.keygen.ComplexAvroKeyGenerator',
> 'write.tasks' = '8',
> 'hive_sync.enable' = 'false'
> );insert into test_hudi
> select `val`
> ,`event_time`
> ,`partition`
> ,`offset`
> ,DATE_FORMAT(event_time, 'yyyy-MM-dd')
> ,DATE_FORMAT(event_time, 'HH')
> from input_table; {code}
>
> h2. Advanced Properties
> {code:java}
> execution.checkpointing.interval=60000ms {code}
>
> h2. Job Profile Properties
> {code:java}
> flink.version=1.13.14
> default.parallelism=8
> restart.from.savepoint=true
> sql.job.mode=normal
> running.mode=streaming
> slots.per.tm=2
> cpu.per.tm=2vcore
> memory.per.tm=6G
> jvm.heap.ratio=70% {code}
>
>
> h1. Issues
> There are 2 main issues here:
> # *TM failing + starting a TM in a new container causing deadlock*
> # *Single commit multi-instant causing various base file parquet errors*
> ** java.io.FileNotFoundException: File does not exist:
> 20220712165157207.parquet (inode 1234567890) Holder
> DFSClient_NONMAPREDUCE_1111111111_22 does not have any open files.
> ** java.io.IOException: The file being written is in an invalid state.
> Probably caused by an error thrown previously. Current state: COLUMN
> ** java.lang.RuntimeException: 20220805210423582.parquet is not a Parquet
> file. expected magic number at tail [80, 65, 82, 49] but found [1, 0, -38, 2]
>
> h1. TM failing + starting a TM in a new container causing deadlock
> # When a TM fails + starting and restoring a TM in a new container creates a
> deadlock situation
> ** TM is waiting for JM to create a new _INFLIGHT_ instant, and the
> ** JM is waiting for TM to send a success WriteMetadataEvent
> # The deadlock above will cause either of the errors below:
> ** org.apache.hudi.exception.HoodieException: Timeout(601000ms) while
> waiting for instant initialize
> ** org.apache.flink.runtime.checkpoint.CheckpointException: Checkpoint
> expired before completing.
> # This will trigger org.apache.flink.runtime.jobmaster.JobMaster [] - Trying
> to recover from a global failure.
> # JM will try to restore itself from the last successful checkpoint
> # This will cause the next issue (illustrated below)
> h2. Root cause
> When restoring the TM, *`AbstractStreamWriteFunction#initializeState()`* will
> attempt to restore the state of the TM. At this stage,
> *`this.currentInstant`* will be initialized by invoking
> {*}`lastPendingInstant()`{*}, in which the ckp metadata path will be loaded
> and a _INFLIGHT_ instant is returned.
>
> When invoking {*}`instantToWrite()`{*},
> *`instant.equals(this.currentInstant)`* will always be true as the local
> *`instant`* is equal to {*}`this.currentInstant`{*}. Hence, the current
> implementation will be stuck in an infinite loop as
> {*}`lastPendingInstant()`{*}, which governs both *`instant`* and
> *`this.currentInstant`* will always return the same value as the state of the
> ckp metadata path is never changed.
>
> This is so as JM is waiting for the TM to finish writing for the batch for
> the _INFLIGHT_ instant. At the same time TM is waiting for JM to create a new
> _INFLIGHT_ instant, hence the deadlock.
>
> The fix is to add additional logic to handle such a case to ensure that a TM
> can obtain a correct _INFLIGHT_ instant when being recovering.
>
> h1. Single commit multi-instant
> # JM restore state from previously successful checkpoint
> # Ckp metadata path is tainted with multiple {_}INFLIGHT{_}s
> # Synchronisation issue occurs if TM executes *`lastPendingInstant()`*
> before JM executes *`startInstant()`*
> # Single commit multi instant issue occurs
> # When tainted TM reads obtained the wrong _INFLIGHT_ instant, it will start
> a new write cycle with the correct _INFLIGHT_ instant while a checkpoint is
> being performed.
> # *`reconcileAgainstMarkers()`* will delete files that tainted TMs are
> writing to with the correct _INFLIGHT_ instant in the next cycle, causing
> FileNotFoundException, _COLUMN_ state errors and parquet corruption errors.
>
> This is caused by the synchronisation issue due to Task Manager (TM) running
> *`ckpMetaData#lastPendingInstant()`* before the Job Manager (JM) executes
> {*}`StreamWriteOperatorCoordinator#startInstant()`{*}, causing the TM to
> fetch an old instantTime.
>
> Since _ABORTED_ statuses are not written to the metadata path, TM will fetch
> a previously aborted instant, thinking that it is still {_}INFLIGHT{_}.
>
> Hence, the new commit will have files of 2 instants should there be a restart
> AND the timeline contains a _INCOMPLETE_ instant
> ({_}ABORTED{_}/{_}INFLIGHT{_}).
>
> Please refer to the example below if a RESTART + job restore is triggered,
> causing a tainted ckp metadata path to be produced. A tainted ckp metadata
> path is a path in which there are more than 1 INFLIGHT files.
>
> h2. CORRECT Instant fetched after JM restores from checkpoint
> After the job restarts, commit *`20220828165937711`* is created, and has
> successfully completed.
>
> If TM invokes *`ckpMetaData#lastPendingInstant()`* AFTER JM runs
> {*}`StreamWriteOperatorCoordinator#startInstant()`{*}, it will read the NEW
> instant on the .aux hdfs path.
>
> *`StreamWriteOperatorCoordinator#startInstant()`* will create the instant
> {*}`20220828170053510`{*}, causing there to be 2 _INFLIGHT_ commits. Since
> the most recent _INCOMPLETE_ instant is fetched, the correct instant, which
> is *`20220828170053510`* will be returned.
>
> The .aux hdfs path will look like this when
> *`ckpMetaData#lastPendingInstant()`* is invoked.
>
> {code:java}
> [
> Ckp{instant='20220828164426755', state='INFLIGHT'},
> Ckp{instant='20220828165937711', state='COMPLETED'},
> Ckp{instant='20220828170053510', state='INFLIGHT'}
> ] {code}
>
>
> h2. INCORRECT Instant fetched after JM restores from checkpoint
> If the TM invokes *`ckpMetaData#lastPendingInstant()`* BEFORE JM runs
> {*}`StreamWriteOperatorCoordinator#startInstant()`{*}, it will read the OLD
> instant on the .aux hdfs path.
> The .aux hdfs path will look like this when
> *`ckpMetaData#lastPendingInstant()`* is invoked.
>
> {code:java}
> [
> Ckp{instant='20220828164123688', state='COMPLETED'},
> Ckp{instant='20220828164426755', state='INFLIGHT'},
> Ckp{instant='20220828165937711', state='COMPLETED'}
> ] {code}
>
> Since the new instant file has not been created yet, the most _INCOMPLETE_ is
> fetched, which is {*}`20220828164426755`{*}.
> In such a case, when there is a possibility that different TMs can obtain
> different instants time from {*}`ckpMetaData#lastPendingInstant()`{*}, a
> single commit might contain multiple instants.
>
> h2. FileNotFoundException and various parquet corruption errors
> Building upon the example in {_}Single commit multiple instant error{_}, when
> *`AppendWriteFunction#flushData()`* is invoked, the current *`writerHelper`*
> will be cleaned up by closing all existing file handles. The *`writerHelper`*
> will then be set to {_}NULL{_}.
> At this point, Hudi is performing a checkpoint and is about to write to
> Hudi's timeline by creating {*}`{*}.commit`* file with the commit's metadata.
> Suppose that a TM (Let's call this the {*}tainted TM{*}) is using the an old
> instant in the tainted ckp metadata path as such: ({*}`20220828164426755`{*}
> will be used)
> {code:java}
> [
> Ckp{instant='20220828164123688', state='COMPLETED'},
> Ckp{instant='20220828164426755', state='INFLIGHT'},
> Ckp{instant='20220828165937711', state='COMPLETED'},
> Ckp{instant='20220828170053510', state='INFLIGHT'}
> ] {code}
>
> Once *`AppendWriteFunction#flushData()`* has completed, on the next
> invocation of {*}`AppendWriteFunction#processElement()`{*},
> *`AppendWriteFunction#initWriterHelper()`* will be invoked, causing a new
> writerHelper with the instant *`20220828170053510`* to be created, and
> writing will begin to files of this instant.
>
> While all this is happening, Flink is performing a checkpoint and is writing
> to Hudi's timeline by creating a {*}`{*}.commit`* file with the commit's
> metadata.
> Before it can create and write to the {*}`{*}.commit`{*},
> *`HoodieTable#finalizeWrite()`{*} will be invoked.
>
> Subsequently, *`HoodieTable#reconcileAgainstMarkers()`* is invoked. What this
> method does is to cleanup partially written data-files due to failed (but
> succeeded on retry) tasks. (SUPPOSEDLY)
> Since the *tainted TM* is using the instant that is being committed to the
> Hudi timeline, the files that it are currently being written to will be
> cleaned up as the marker files can be found, but data files are not found in
> the {*}`HoodieWriteStat`{*}.
>
> In essence, the *tainted TM* is starting a write cycle when it should not be
> doing so with an instant that is being committed Hudi timeline, hence,
> causing files to be deleted/corrupted/being in the wrong state.
>
> Hence, when the JM is performing a checkpoint + commit, the *tainted TM*
> might try to close the file handle if it has reached the
> {*}`parquet.max.filesize`{*}. When trying to close the file handle, this file
> might have already been deleted by
> {*}`HoodieTable#reconcileAgainstMarkers()`{*}, causing the
> {*}`java.io.FileNotFoundException`{*}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
