[
https://issues.apache.org/jira/browse/FLINK-36546?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Sergei Morozov updated FLINK-36546:
-----------------------------------
Description:
Currently, there is hard-coded {{boolean isBatchMode = false}} in
{{DataSinkTranslator}}.
Assume the following pipeline:
1. The source is a bounded source.
2. The sink is a two-phase committing sink.
In this setup, when {{CommitterOperator#endInput()}} is invoked, it will not
call {{notifyCheckpointComplete(Long.MAX_VALUE)}}. As a result, when
{{CommitterOperator#processElement()}} is invoked with the committable with
checkpoint id set to {{Long.MAX_VALUE}}, it will be ignored because it will be
greater than {{lastCompletedCheckpointId}}.
was:
Currently, there is hard-coded {{boolean isBatchMode = false}} in
{{DataSinkTranslator}}.
Assume that there is a sink that has some logic executed at the end of the
batch. This is is determined by the {{endOfInput}} parameter passed to
{{SinkWriter#flush}}.
Testing such a sink with the "values" connector as the source is challenging
since the sink never receives the end-of-input signal.
> Handle batch sources in DataSinkTranslator
> ------------------------------------------
>
> Key: FLINK-36546
> URL: https://issues.apache.org/jira/browse/FLINK-36546
> Project: Flink
> Issue Type: Improvement
> Components: Flink CDC
> Affects Versions: cdc-3.2.0
> Reporter: Sergei Morozov
> Priority: Minor
> Labels: pull-request-available
>
> Currently, there is hard-coded {{boolean isBatchMode = false}} in
> {{DataSinkTranslator}}.
> Assume the following pipeline:
> 1. The source is a bounded source.
> 2. The sink is a two-phase committing sink.
> In this setup, when {{CommitterOperator#endInput()}} is invoked, it will not
> call {{notifyCheckpointComplete(Long.MAX_VALUE)}}. As a result, when
> {{CommitterOperator#processElement()}} is invoked with the committable with
> checkpoint id set to {{Long.MAX_VALUE}}, it will be ignored because it will
> be greater than {{lastCompletedCheckpointId}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
