Hi, Ok, I see. You can try to rewrite your logic (or maybe records schema by adding some ID fields) to manually deduplicating the records after processing them with at least once semantic. Such setup is usually simpler, with slightly better throughput and significantly better latency (end-to-end exactly once latency is limited by checkpointing time).
Piotrek > On 12 Aug 2019, at 11:12, Tony Wei <tony19920...@gmail.com> wrote: > > Hi Piotr, > > Thanks a lot. I need exactly once in my use case, but instead of having the > risk of losing data, at least once is more acceptable when error occurred. > > Best, > Tony Wei > > Piotr Nowojski <pi...@data-artisans.com <mailto:pi...@data-artisans.com>> 於 > 2019年8月12日 週一 下午3:27寫道： > Hi, > > Yes, if it’s due to transaction timeout you will lose the data. > > Whether can you fallback to at least once, that depends on Kafka, not on > Flink, since it’s the Kafka that timeouts those transactions and I don’t see > in the documentation anything that could override this [1]. You might try > disabling the mechanism via setting > `transaction.abort.timed.out.transaction.cleanup.interval.ms > <http://transaction.abort.timed.out.transaction.cleanup.interval.ms/>` or > `transaction.remove.expired.transaction.cleanup.interval.ms > <http://transaction.remove.expired.transaction.cleanup.interval.ms/>`, but > that’s question more to Kafka guys. Maybe Becket could help with this. > > Also it MIGHT be that Kafka doesn’t remove records from the topics when > aborting the transaction and MAYBE you can still access them via > “READ_UNCOMMITTED” mode. But that’s again, question to Kafka. > > Sorry that I can not help more. > > If you do not care about exactly once, why don’t you just set the connector > to at least once mode? > > Piotrek > >> On 12 Aug 2019, at 06:29, Tony Wei <tony19920...@gmail.com >> <mailto:tony19920...@gmail.com>> wrote: >> >> Hi, >> >> I had the same exception recently. I want to confirm that if it is due to >> transaction timeout, >> then I will lose those data. Am I right? Can I make it fall back to at least >> once semantic in >> this situation? >> >> Best, >> Tony Wei >> >> Piotr Nowojski <pi...@data-artisans.com <mailto:pi...@data-artisans.com>> 於 >> 2018年3月21日 週三 下午10:28寫道： >> Hi, >> >> But that’s exactly the case: producer’s transaction timeout starts when the >> external transaction starts - but FlinkKafkaProducer011 keeps an active >> Kafka transaction for the whole period between checkpoints. >> >> As I wrote in the previous message: >> >> > in case of failure, your timeout must also be able to cover the additional >> > downtime required for the successful job restart. Thus you should increase >> > your timeout accordingly. >> >> I think that 15 minutes timeout is a way too small value. If your job fails >> because of some intermittent failure (for example worker crash/restart), you >> will only have a couple of minutes for a successful Flink job restart. >> Otherwise you will lose some data (because of the transaction timeouts). >> >> Piotrek >> >>> On 21 Mar 2018, at 10:30, Dongwon Kim <eastcirc...@gmail.com >>> <mailto:eastcirc...@gmail.com>> wrote: >>> >>> Hi Piotr, >>> >>> Now my streaming pipeline is working without retries. >>> I decreased Flink's checkpoint interval from 15min to 10min as you >>> suggested [see screenshot_10min_ckpt.png]. >>> >>> I though that producer's transaction timeout starts when the external >>> transaction starts. >>> The truth is that Producer's transaction timeout starts after the last >>> external checkpoint is committed. >>> Now that I have 15min for Producer's transaction timeout and 10min for >>> Flink's checkpoint interval, and every checkpoint takes less than 5 >>> minutes, everything is working fine. >>> Am I right? >>> >>> Anyway thank you very much for the detailed explanation! >>> >>> Best, >>> >>> Dongwon >>> >>> >>> >>> On Tue, Mar 20, 2018 at 8:10 PM, Piotr Nowojski <pi...@data-artisans.com >>> <mailto:pi...@data-artisans.com>> wrote: >>> Hi, >>> >>> Please increase transaction.timeout.ms <http://transaction.timeout.ms/> to >>> a greater value or decrease Flink’s checkpoint interval, I’m pretty sure >>> the issue here is that those two values are overlapping. I think that’s >>> even visible on the screenshots. First checkpoint completed started at >>> 14:28:48 and ended at 14:30:43, while the second one started at 14:45:53 >>> and ended at 14:49:16. That gives you minimal transaction duration of 15 >>> minutes and 10 seconds, with maximal transaction duration of 21 minutes. >>> >>> In HAPPY SCENARIO (without any failure and restarting), you should assume >>> that your timeout interval should cover with some safety margin the period >>> between start of a checkpoint and end of the NEXT checkpoint, since this is >>> the upper bound how long the transaction might be used. In your case at >>> least ~25 minutes. >>> >>> On top of that, as described in the docs, >>> https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/connectors/kafka.html#kafka-producers-and-fault-tolerance >>> >>> <https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/connectors/kafka.html#kafka-producers-and-fault-tolerance> >>> , in case of failure, your timeout must also be able to cover the >>> additional downtime required for the successful job restart. Thus you >>> should increase your timeout accordingly. >>> >>> Piotrek >>> >>> >>>> On 20 Mar 2018, at 11:58, Dongwon Kim <eastcirc...@gmail.com >>>> <mailto:eastcirc...@gmail.com>> wrote: >>>> >>>> Hi Piotr, >>>> >>>> We have set producer's [transaction.timeout.ms >>>> <http://transaction.timeout.ms/>] to 15 minutes and have used the default >>>> setting for broker (15 mins). >>>> As Flink's checkpoint interval is 15 minutes, it is not a situation where >>>> Kafka's timeout is smaller than Flink's checkpoint interval. >>>> As our first checkpoint just takes 2 minutes, it seems like transaction is >>>> not committed properly. >>>> >>>> Best, >>>> >>>> - Dongwon >>>> >>>> >>>> >>>> >>>> >>>> On Tue, Mar 20, 2018 at 6:32 PM, Piotr Nowojski <pi...@data-artisans.com >>>> <mailto:pi...@data-artisans.com>> wrote: >>>> Hi, >>>> >>>> What’s your Kafka’s transaction timeout setting? Please both check Kafka >>>> producer configuration (transaction.timeout.ms >>>> <http://transaction.timeout.ms/> property) and Kafka broker configuration. >>>> The most likely cause of such error message is when Kafka's timeout is >>>> smaller then Flink’s checkpoint interval and transactions are not >>>> committed quickly enough before timeout occurring. >>>> >>>> Piotrek >>>> >>>>> On 17 Mar 2018, at 07:24, Dongwon Kim <eastcirc...@gmail.com >>>>> <mailto:eastcirc...@gmail.com>> wrote: >>>>> >>>>> >>>>> Hi, >>>>> >>>>> I'm faced with the following ProducerFencedException after 1st, 3rd, 5th, >>>>> 7th, ... checkpoints: >>>>> -- >>>>> java.lang.RuntimeException: Error while confirming checkpoint >>>>> at org.apache.flink.runtime.taskmanager.Task$3.run(Task.java:1260) >>>>> at >>>>> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) >>>>> at java.util.concurrent.FutureTask.run(FutureTask.java:266) >>>>> at >>>>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) >>>>> at >>>>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) >>>>> at java.lang.Thread.run(Thread.java:748) >>>>> Caused by: org.apache.kafka.common.errors.ProducerFencedException: >>>>> Producer attempted an operation with an old epoch. Either there is a >>>>> newer producer with the same transactionalId, or the producer's >>>>> transaction has been expired by the broker. >>>>> -- >>>>> >>>>> FYI, I'm using Flink 1.4.0 and testing end-to-end exactly once processing >>>>> using Kafka sink. >>>>> We use FsStateBackend to store snapshot data on HDFS. >>>>> >>>>> As shown in configuration.png, my checkpoint configuration is: >>>>> - Checkpointing Mode : Exactly Once >>>>> - Interval : 15m 0s >>>>> - Timeout : 10m 0s >>>>> - Minimum Pause Between Checkpoints : 5m 0s >>>>> - Maximum Concurrent Checkpoints : 1 >>>>> - Persist Checkpoints Externally : Disabled >>>>> >>>>> After the first checkpoint completed [see history after 1st ckpt.png], >>>>> the job is restarted due to the ProducerFencedException [see exception >>>>> after 1st ckpt.png]. >>>>> The first checkpoint takes less than 2 minutes while my checkpoint >>>>> interval is 15m and minimum pause between checkpoints is 5m. >>>>> After the job is restarted, the second checkpoint is triggered after a >>>>> while [see history after 2nd ckpt.png] and this time I've got no >>>>> exception. >>>>> The third checkpoint results in the same exception as after the first >>>>> checkpoint. >>>>> >>>>> Can anybody let me know what's going wrong behind the scene? >>>>> >>>>> Best, >>>>> >>>>> Dongwon >>>>> <history after 3rd ckpt.png><exception after 3rd ckpt.png><history after >>>>> 2nd ckpt.png><configuration.png><summary.png><exception after 1st >>>>> ckpt.png><history after 1st ckpt.png> >>>> >>>> >>> >>> >>> <screenshot_10min_ckpt.png> >> >
