[jira] [Updated] (HUDI-9381) Async instant time generation of Flink writers

["Danny Chen (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/HUDI-9381?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Danny Chen updated HUDI-9381:
-----------------------------
    Summary: Async instant time generation of Flink writers  (was: Remove Ckp metadata of Flink writers)

> Async instant time generation of Flink writers
> ----------------------------------------------
>
>                 Key: HUDI-9381
>                 URL: https://issues.apache.org/jira/browse/HUDI-9381
>             Project: Apache Hudi
>          Issue Type: Improvement
>          Components: writer-core
>            Reporter: Danny Chen
>            Assignee: voon
>            Priority: Major
>             Fix For: 1.1.0
>
>
> Since Flink 1.15, a new API was introduced to support responsive comminication between write tasks and the coordinator: https://issues.apache.org/jira/browse/FLINK-26077.
> We can use this API to fetch the in-memory current instant from the coordinator and remove/deprecate the ckp metadata.
> The current impl of task gateway -> coordinator communitation:
>  * in WriteOperatorFactory#createStreamOperator, the OperatorEventGateway is set up for each write function;
>  * in each write function's #snapshotState, the gateway sends the write metadata event to the coordinator;
> While in FLINK-26077, the OperatorEventGateway API was not changed, instead it changed a more underneath API named TaskOperatorEventGateway, which can be fetched through:
> {code:java}
> // parameters is an instance of StreamOperatorParameters<Object> in the WriteOperatorFactory
> parameters.getContainingTask().getEnvironment().getOperatorCoordinatorEventGateway();
> {code}
> And we can send responsive msg with the new API:
> {code:java}
> TaskOperatorEventGateway#sendRequestToCoordinator
> {code}
> So somehow we need to figure out a way to setup/register the TaskOperatorEventGateway into each write function, we need to wrap up the OperatorID together with it becasue the OperatorID is required by the method TaskOperatorEventGateway#sendRequestToCoordinator.



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