Hi Di Thank you for your patience and explanation.
If this is a server-side configuration, we currently cannot modify it in the client configuration. If Doris supports client-side configuration in the future, we can reconsider whether to support it. I currently have no other questions regarding this FLIP. LGTM. Best, Feng On Mon, Mar 25, 2024 at 3:42 PM wudi <676366...@qq.com.invalid> wrote: > Hi, Feng > > Yes, if the StreamLoad transaction timeout is very short, you may > encounter this situation. > > The timeout for StreamLoad transactions is controlled by the > streaming_label_keep_max_second parameter [1] in FE (Frontend), and the > default value is 12 hours. Currently, it is a global transaction > configuration and cannot be set separately for a specific transaction. > > However, I understand the default 12-hour timeout should cover most cases > unless you are restarting from a checkpoint that occurred a long time ago. > What do you think? > > > [1] > https://github.com/apache/doris/blob/master/fe/fe-common/src/main/java/org/apache/doris/common/Config.java#L163-L168 > > > Brs > di.wu > > > 2024年3月25日 11:45，Feng Jin <jinfeng1...@gmail.com> 写道： > > > > Hi Di > > > > Thanks for your reply. > > > > The timeout I'm referring to here is not the commit timeout, but rather > the > > timeout for a single streamLoad transaction. > > > > Let's say we have set the transaction timeout for StreamLoad to be 10 > > minutes. Now, imagine there is a Flink job with two subtasks. Due to > > significant data skew and backpressure issues, subtask 0 and subtask 1 > are > > processing at different speeds. Subtask 0 finishes processing this > > checkpoint first, while subtask 1 takes another 10 minutes to complete > its > > processing. At this point, the job's checkpoint is done. However, since > > subtask 0 has been waiting for subtask 1 all along, its corresponding > > streamLoad transaction closes after more than 10 minutes have passed - by > > which time the server has already cleaned up this transaction, leading > to a > > failed commit. > > Therefore, I would like to know if in such situations we can avoid this > > problem by setting a longer lifespan for transactions. > > > > > > Best, > > Feng > > > > > > On Fri, Mar 22, 2024 at 10:24 PM wudi <676366...@qq.com.invalid> wrote: > > > >> Hi, Feng, > >> > >> 1. Are you suggesting that when a commit gets stuck, we can interrupt > the > >> commit request using a timeout parameter? Currently, there is no such > >> parameter. In my understanding, in a two-phase commit, checkpoint must > be > >> enabled, so the commit timeout is essentially the checkpoint timeout. > >> Therefore, it seems unnecessary to add an additional parameter here. > What > >> do you think? > >> > >> 2. In addition to deleting checkpoints to re-consume data again, the > >> Connector also provides an option to ignore commit errors[1]. However, > this > >> option is only used for error recovery scenarios, such as when a > >> transaction is cleared by the server but you want to reuse the upstream > >> offset from the checkpoint. > >> > >> 3. Also, thank you for pointing out the issue with the parameter. It has > >> already been addressed[2], but the FLIP changes were overlooked. It has > >> been updated. > >> > >> [1] > >> > https://github.com/apache/doris-flink-connector/blob/master/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/committer/DorisCommitter.java#L150-L160 > >> [2] > >> > https://github.com/apache/doris-flink-connector/blob/master/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisConfigOptions.java#L89-L98 > >> > >> Brs > >> di.wu > >> > >> > >> > >>> 2024年3月22日 18:28，Feng Jin <jinfeng1...@gmail.com> 写道： > >>> > >>> Hi Di, > >>> > >>> Thank you for the update, as well as quickly implementing corresponding > >>> capabilities including filter push down and project push down. > >>> > >>> Regarding the transaction timeout, I still have some doubts. I would > like > >>> to confirm if we can control this timeout parameter in the connector, > >> such > >>> as setting it to 10 minutes or 1 hour. > >>> Also, when a transaction is cleared by the server, the commit operation > >> of > >>> the connector will fail, leading to job failure. In this case, can > users > >>> only choose to delete the checkpoint and re-consume historical data? > >>> > >>> There is also a small question regarding the parameters*: * > >>> *doris.request.connect.timeout.ms < > >> http://doris.request.connect.timeout.ms>* > >>> and d*oris.request.read.timeout.ms < > http://oris.request.read.timeout.ms > >>> *, > >>> can we change them to Duration type and remove the "ms" suffix.? > >>> This way, all time parameters can be kept uniform in type as duration. > >>> > >>> > >>> Best, > >>> Feng > >>> > >>> On Fri, Mar 22, 2024 at 4:46 PM wudi <676366...@qq.com.invalid> wrote: > >>> > >>>> Hi, Feng, > >>>> Thank you, that's a great suggestion ! > >>>> > >>>> I have already implemented FilterPushDown and removed that parameter > on > >>>> DorisDynamicTableSource[1], and also updated FLIP. > >>>> > >>>> Regarding the mention of [Doris also aborts transactions], it may not > >> have > >>>> been described accurately. It mainly refers to the automatic > expiration > >> of > >>>> long-running transactions in Doris that have not been committed for a > >>>> prolonged period. > >>>> > >>>> As for two-phase commit, when a commit fails, the checkpoint will also > >>>> fail, and the job will be continuously retried. > >>>> > >>>> [1] > >>>> > >> > https://github.com/apache/doris-flink-connector/blob/master/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java#L58 > >>>> > >>>> Brs > >>>> di.wu > >>>> > >>>> > >>>>> 2024年3月15日 14:53，Feng Jin <jinfeng1...@gmail.com> 写道： > >>>>> > >>>>> Hi Di > >>>>> > >>>>> Thank you for initiating this FLIP, +1 for this. > >>>>> > >>>>> Regarding the option `doris.filter.query` of doris source table > >>>>> > >>>>> Can we directly implement the FilterPushDown capability of Flink > Source > >>>>> like Jdbc Source [1] instead of introducing an option? > >>>>> > >>>>> > >>>>> Regarding two-phase commit, > >>>>> > >>>>>> At the same time, Doris will also abort transactions that have not > >> been > >>>>> committed for a long time > >>>>> > >>>>> Can we control the transaction timeout in the connector? > >>>>> And control the behavior when timeout occurs, whether to discard by > >>>> default > >>>>> or trigger job failure? > >>>>> > >>>>> > >>>>> [1]. https://issues.apache.org/jira/browse/FLINK-16024 > >>>>> > >>>>> Best, > >>>>> Feng > >>>>> > >>>>> > >>>>> On Tue, Mar 12, 2024 at 12:12 AM Ferenc Csaky > >> <ferenc.cs...@pm.me.invalid > >>>>> > >>>>> wrote: > >>>>> > >>>>>> Hi, > >>>>>> > >>>>>> Thanks for driving this, +1 for the FLIP. > >>>>>> > >>>>>> Best, > >>>>>> Ferenc > >>>>>> > >>>>>> > >>>>>> > >>>>>> > >>>>>> On Monday, March 11th, 2024 at 15:17, Ahmed Hamdy < > >> hamdy10...@gmail.com > >>>>> > >>>>>> wrote: > >>>>>> > >>>>>>> > >>>>>>> > >>>>>>> Hello, > >>>>>>> Thanks for the proposal, +1 for the FLIP. > >>>>>>> > >>>>>>> Best Regards > >>>>>>> Ahmed Hamdy > >>>>>>> > >>>>>>> > >>>>>>> On Mon, 11 Mar 2024 at 15:12, wudi 676366...@qq.com.invalid wrote: > >>>>>>> > >>>>>>>> Hi, Leonard > >>>>>>>> Thank you for your suggestion. > >>>>>>>> I referred to other Connectors[1], modified the naming and types > of > >>>>>>>> relevant parameters[2], and also updated FLIP. > >>>>>>>> > >>>>>>>> [1] > >>>>>>>> > >>>>>> > >>>> > >> > https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/connectors/table/overview/ > >>>>>>>> [1] > >>>>>>>> > >>>>>> > >>>> > >> > https://github.com/apache/doris-flink-connector/blob/master/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisConfigOptions.java > >>>>>>>> > >>>>>>>> Brs, > >>>>>>>> di.wu > >>>>>>>> > >>>>>>>>> 2024年3月7日 14:33，Leonard Xu xbjt...@gmail.com 写道： > >>>>>>>>> > >>>>>>>>> Thanks wudi for the updating, the FLIP generally looks good to > me, > >> I > >>>>>>>>> only left two minor suggestions: > >>>>>>>>> > >>>>>>>>> (1) The suffix `.s` in configoption doris.request.query.timeout.s > >>>>>> looks > >>>>>>>>> strange to me, could we change all time interval related option > >>>>>> value type > >>>>>>>>> to Duration ? > >>>>>>>>> > >>>>>>>>> (2) Could you check and improve all config options like > >>>>>>>>> `doris.exec.mem.limit` to make them to follow flink config option > >>>>>> naming > >>>>>>>>> and value type? > >>>>>>>>> > >>>>>>>>> Best, > >>>>>>>>> Leonard > >>>>>>>>> > >>>>>>>>>>> 2024年3月6日 06:12，Jing Ge j...@ververica.com.INVALID 写道： > >>>>>>>>>>> > >>>>>>>>>>> Hi Di, > >>>>>>>>>>> > >>>>>>>>>>> Thanks for your proposal. +1 for the contribution. I'd like to > >>>>>> know > >>>>>>>>>>> your > >>>>>>>>>>> thoughts about the following questions: > >>>>>>>>>>> > >>>>>>>>>>> 1. According to your clarification of the exactly-once, thanks > >>>>>> for it > >>>>>>>>>>> BTW, > >>>>>>>>>>> no PreCommitTopology is required. Does it make sense to let > >>>>>>>>>>> DorisSink[1] > >>>>>>>>>>> implement SupportsCommitter, since the TwoPhaseCommittingSink > is > >>>>>>>>>>> deprecated[2] before turning the Doris connector into a Flink > >>>>>>>>>>> connector? > >>>>>>>>>>> 2. OLAP engines are commonly used as the tail/downstream of a > >>>>>> data > >>>>>>>>>>> pipeline > >>>>>>>>>>> to support further e.g. ad-hoc query or cube with feasible > >>>>>>>>>>> pre-aggregation. > >>>>>>>>>>> Just out of curiosity, would you like to share some real use > >>>>>> cases that > >>>>>>>>>>> will use OLAP engines as the source of a streaming data > >>>>>> pipeline? Or it > >>>>>>>>>>> will only be used as the source for the batch? > >>>>>>>>>>> 3. The E2E test only covered sink[3], if I am not mistaken. > >>>>>> Would you > >>>>>>>>>>> like > >>>>>>>>>>> to test the source in E2E too? > >>>>>>>>>>> > >>>>>>>>>>> [1] > >>>>>>>> > >>>>>>>> > >>>>>> > >>>> > >> > https://github.com/apache/doris-flink-connector/blob/43e0e5cf9b832854ea228fb093077872e3a311b6/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/DorisSink.java#L55 > >>>>>>>> > >>>>>>>>>>> [2] > >>>>>>>> > >>>>>>>> > >>>>>> > >>>> > >> > https://cwiki.apache.org/confluence/display/FLINK/FLIP-372%3A+Enhance+and+synchronize+Sink+API+to+match+the+Source+API > >>>>>>>> > >>>>>>>>>>> [3] > >>>>>>>> > >>>>>>>> > >>>>>> > >>>> > >> > https://github.com/apache/doris-flink-connector/blob/43e0e5cf9b832854ea228fb093077872e3a311b6/flink-doris-connector/src/test/java/org/apache/doris/flink/tools/cdc/MySQLDorisE2ECase.java#L96 > >>>>>>>> > >>>>>>>>>>> Best regards, > >>>>>>>>>>> Jing > >>>>>>>>>>> > >>>>>>>>>>> On Tue, Mar 5, 2024 at 11:18 AM wudi 676366...@qq.com.invalid > >>>>>> wrote: > >>>>>>>>>>> > >>>>>>>>>>>> Hi, Jeyhun Karimov. > >>>>>>>>>>>> Thanks for your question. > >>>>>>>>>>>> > >>>>>>>>>>>> - How to ensure Exactly-Once? > >>>>>>>>>>>> 1. When the Checkpoint Barrier arrives, DorisSink will trigger > >>>>>> the > >>>>>>>>>>>> precommit api of StreamLoad to complete the persistence of > >>>>>> data in > >>>>>>>>>>>> Doris > >>>>>>>>>>>> (the data will not be visible at this time), and will also > >>>>>> pass this > >>>>>>>>>>>> TxnID > >>>>>>>>>>>> to the Committer. > >>>>>>>>>>>> 2. When this Checkpoint of the entire Job is completed, the > >>>>>> Committer > >>>>>>>>>>>> will > >>>>>>>>>>>> call the commit api of StreamLoad and commit TxnID to complete > >>>>>> the > >>>>>>>>>>>> visibility of the transaction. > >>>>>>>>>>>> 3. When the task is restarted, the Txn with successful > >>>>>> precommit and > >>>>>>>>>>>> failed commit will be aborted based on the label-prefix, and > >>>>>> Doris' > >>>>>>>>>>>> abort > >>>>>>>>>>>> API will be called. (At the same time, Doris will also abort > >>>>>>>>>>>> transactions > >>>>>>>>>>>> that have not been committed for a long time) > >>>>>>>>>>>> > >>>>>>>>>>>> ps: At the same time, this part of the content has been > >>>>>> updated in > >>>>>>>>>>>> FLIP > >>>>>>>>>>>> > >>>>>>>>>>>> - Because the default table model in Doris is Duplicate ( > >>>>>>>>>>>> https://doris.apache.org/docs/data-table/data-model/), which > >>>>>> does not > >>>>>>>>>>>> have a primary key, batch writing may cause data duplication, > >>>>>> but > >>>>>>>>>>>> UNIQ The > >>>>>>>>>>>> model has a primary key, which ensures the idempotence of > >>>>>> writing, > >>>>>>>>>>>> thus > >>>>>>>>>>>> achieving Exactly-Once > >>>>>>>>>>>> > >>>>>>>>>>>> Brs, > >>>>>>>>>>>> di.wu > >>>>>>>>>>>> > >>>>>>>>>>>>> 2024年3月2日 17:50，Jeyhun Karimov je.kari...@gmail.com 写道： > >>>>>>>>>>>>> > >>>>>>>>>>>>> Hi, > >>>>>>>>>>>>> > >>>>>>>>>>>>> Thanks for the proposal. +1 for the FLIP. > >>>>>>>>>>>>> I have a few questions: > >>>>>>>>>>>>> > >>>>>>>>>>>>> - How exactly the two (Stream Load's two-phase commit and > >>>>>> Flink's > >>>>>>>>>>>>> two-phase > >>>>>>>>>>>>> commit) combination will ensure the e2e exactly-once > >>>>>> semantics? > >>>>>>>>>>>>> > >>>>>>>>>>>>> - The FLIP proposes to combine Doris's batch writing with the > >>>>>>>>>>>>> primary key > >>>>>>>>>>>>> table to achieve Exactly-Once semantics. Could you elaborate > >>>>>> more on > >>>>>>>>>>>>> that? > >>>>>>>>>>>>> Why it is not the default behavior but a workaround? > >>>>>>>>>>>>> > >>>>>>>>>>>>> Regards, > >>>>>>>>>>>>> Jeyhun > >>>>>>>>>>>>> > >>>>>>>>>>>>> On Sat, Mar 2, 2024 at 10:14 AM Yanquan Lv > >>>>>> decq12y...@gmail.com > >>>>>>>>>>>>> wrote: > >>>>>>>>>>>>> > >>>>>>>>>>>>>> Thanks for driving this. > >>>>>>>>>>>>>> The content is very detailed, it is recommended to add a > >>>>>> section on > >>>>>>>>>>>>>> Test > >>>>>>>>>>>>>> Plan for more completeness. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Di Wu d...@apache.org 于2024年1月25日周四 15:40写道： > >>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Hi all, > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Previously, we had some discussions about contributing > >>>>>> Flink Doris > >>>>>>>>>>>>>>> Connector to the Flink community [1]. I want to further > >>>>>> promote > >>>>>>>>>>>>>>> this > >>>>>>>>>>>>>>> work. > >>>>>>>>>>>>>>> I hope everyone will help participate in this FLIP > >>>>>> discussion and > >>>>>>>>>>>>>>> provide > >>>>>>>>>>>>>>> more valuable opinions and suggestions. > >>>>>>>>>>>>>>> Thanks. > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> [1] > >>>>>>>>>>>>>>> > >>>>>> https://lists.apache.org/thread/lvh8g9o6qj8bt3oh60q81z0o1cv3nn8p > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Brs, > >>>>>>>>>>>>>>> di.wu > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> On 2023/12/07 05:02:46 wudi wrote: > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> Hi all, > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> As discussed in the previous email [1], about > >>>>>> contributing the > >>>>>>>>>>>>>>>> Flink > >>>>>>>>>>>>>>>> Doris Connector to the Flink community. > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> Apache Doris[2] is a high-performance, real-time > >>>>>> analytical > >>>>>>>>>>>>>>>> database > >>>>>>>>>>>>>>>> based on MPP architecture, for scenarios where Flink > >>>>>> is used for > >>>>>>>>>>>>>>>> data > >>>>>>>>>>>>>>>> analysis, processing, or real-time writing on Doris, > >>>>>> Flink Doris > >>>>>>>>>>>>>>>> Connector > >>>>>>>>>>>>>>>> is an effective tool. > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> At the same time, Contributing Flink Doris Connector > >>>>>> to the Flink > >>>>>>>>>>>>>>>> community will further expand the Flink Connectors > >>>>>> ecosystem. > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> So I would like to start an official discussion > >>>>>> FLIP-399: Flink > >>>>>>>>>>>>>>>> Connector Doris[3]. > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> Looking forward to comments, feedbacks and suggestions > >>>>>> from the > >>>>>>>>>>>>>>>> community on the proposal. > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> [1] > >>>>>>>>>>>>>>>> > >>>>>> https://lists.apache.org/thread/lvh8g9o6qj8bt3oh60q81z0o1cv3nn8p > >>>>>>>>>>>>>>>> [2] > >>>>>>>> > >>>>>>>> > >> https://doris.apache.org/docs/dev/get-starting/what-is-apache-doris/ > >>>>>>>> > >>>>>>>>>>>>>>>> [3] > >>>>>>>> > >>>>>>>> > >>>>>> > >>>> > >> > https://cwiki.apache.org/confluence/display/FLINK/FLIP-399%3A+Flink+Connector+Doris > >>>>>>>> > >>>>>>>>>>>>>>>> Brs, > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> di.wu > >>>>>> > >>>> > >>>> > >>> > >> > >> > >
