Hi wudi I'm afraid it cannot reduce back pressure. If Writer and Commiter are not chained together, and the Commiter task runs slowly, it can block the upstream Writer tasks by back pressure too.
On the other hand, you can try to increase the parallelism of sink node to speedup the Commiter operation Best, Shammon On Thu, Feb 16, 2023 at 11:38 PM wudi <676366...@qq.com.invalid> wrote: > thanks for your replies. > I think that if Writer and Commiter are not chained together, data > consistency can be guaranteed, right? > Because when the Commiter does not block the Writer, at the next > Checkpoint, if the Commit is not completed, the SinkWriter.precommit will > not be triggered > > In addition, in this scenario (writer blocking caused by slow commit), may > the performance of disabling Sink's chain be better? Because it can reduce > a lot of back pressure. > > Thanks && Regards, > di.wu > > > > 2023年2月16日 下午10:05，Chesnay Schepler <ches...@apache.org> 写道： > > > > As far as I know that chain between committer and writer is also > required for correctness. > > > > On 16/02/2023 14:53, weijie guo wrote: > >> Hi wu, > >> > >> I don't think it is a good choice to directly change the strategy of > chain. Operator chain usually has better performance and resource > utilization. If we directly change the chain policy between them, users can > no longer chain them together, which is not a good starting point. > >> > >> Best regards, > >> > >> Weijie > >> > >> > >> > >> wudi <676366...@qq.com> 于2023年2月16日周四 19:29写道： > >> > >> Thank you for your reply. > >> > >> Currently in the custom Sink Connector, the Flink task will > >> combine Writer and Committer into one thread, and the thread name > >> is similar: *[Sink: Writer -> Sink: Committer (1/1)#0]*. > >> In this way, when the *Committer.commit()* method is very slow, it > >> will block the*SinkWriter.write()* method to receive upstream data. > >> > >> The client can use the *env.disableOperatorChaining() *method to > >> split the thread into two threads:*[Sink: Writer (1/1)#0] *and > >> *[Sink: Committer (1/1)#0]*. This Committer. The commit method > >> will not block the SinkWriter.write method. > >> > >> If the chain policy can be disabled in the custom Sink Connector, > >> the client can be prevented from setting and disabling the chain. > >> Or is there a better way to make*Committer.commit()* not block > >> *SinkWriter.write()*? > >> > >> Looking forward for your reply. > >> Thanks && Regards, > >> di.wu > >> > >>> 2023年2月16日 下午6:54，Shammon FY <zjur...@gmail.com> 写道： > >>> > >>> Hi > >>> > >>> Do you mean how to disable `chain` in your custom sink > >>> connector? Can you > >>> give an example of what you want? > >>> > >>> On Wed, Feb 15, 2023 at 10:42 PM di wu <676366...@qq.com> wrote: > >>> > >>>> Hello > >>>> > >>>> The current Sink operator will be split into two operations, > >>>> Writer and > >>>> Commiter. By default, they will be chained together and executed > >>>> on the > >>>> same thread. > >>>> So sometimes when the commiter is very slow, it will block the data > >>>> writer, causing back pressure. > >>>> > >>>> At present, FlinkSQL can be solved by disabling the chain > >>>> globally, and > >>>> DataStream can partially disable the chain through the > >>>> disableChaining > >>>> method, but both of them need to be set by the user. > >>>> > >>>> Can the strategy of the Chain be changed in the Custom Sink > >>>> Connector to > >>>> separate Writer and Commiter? > >>>> > >>>> Thanks && Regards, > >>>> di.wu > >>>> > >>> > >> > > > >
