You should increase the kafka transaction timeout --
transaction.max.timeout.ms -- to something much larger than the default,
which I believe is 15 minutes. Suitable values are more on the order of a
few hours to a few days -- long enough to allow for any conceivable outage.
This way, if a request does timeout and causes the Flink job to fail, so
long as Kafka and Flink recover within the transaction timeout you won't
lose any data.
Regards,
David
On Sat, Mar 20, 2021 at 12:02 AM Rainie Li <raini...@pinterest.com> wrote:
> Hi Arvid,
>
> After increasing producer.kafka.request.timeout.ms from 90000 to 120000.
> The job has been running fine for almost 5 days, but one of the tasks
> failed again recently for the same timeout error. (attached stack trace
> below)
> Should I keep increasing producer.kafka.request.timeout.ms value?
>
> Thanks again for the help.
> Best regards
> Rainie
>
> *Stacktrace:*
> {job_name}/{job_id}/chk-43556/_metadata, reference=(default),
> fileStateSizeThreshold=2048, writeBufferSize=4096}, synchronous part) in
> thread Thread[Process-Event -> Filter-data08 (237/240),5,Flink Task
> Threads] took 0 ms.
> Canceler.run(Task.java:1434)
> ... 1 more2021-03-17 17:46:42,284 INFO
> org.apache.flink.streaming.api.operators.AbstractStreamOperator - Could
> not complete snapshot 43556 for operator Sink: Sink-data08 (237/240).
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Failed to
> send data to Kafka: Expiring 53 record(s) for frontend_event_core-46:
> 122269 ms has passed since last append
> at
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.checkErroneous(FlinkKafkaProducer.java:1196)
> at
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.flush(FlinkKafkaProducer.java:968)
> at
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.preCommit(FlinkKafkaProducer.java:892)
> at
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.preCommit(FlinkKafkaProducer.java:98)
> at
> org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.snapshotState(TwoPhaseCommitSinkFunction.java:310)
> at
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.snapshotState(FlinkKafkaProducer.java:973)
> at
> org.apache.flink.streaming.util.functions.StreamingFunctionUtils.trySnapshotFunctionState(StreamingFunctionUtils.java:118)
> at
> org.apache.flink.streaming.util.functions.StreamingFunctionUtils.snapshotFunctionState(StreamingFunctionUtils.java:99)
> at
> org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.snapshotState(AbstractUdfStreamOperator.java:90)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:399)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.checkpointStreamOperator(StreamTask.java:1302)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.executeCheckpointing(StreamTask.java:1236)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.checkpointState(StreamTask.java:892)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:797)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:728)
> at
> org.apache.flink.streaming.runtime.io.CheckpointBarrierHandler.notifyCheckpoint(CheckpointBarrierHandler.java:88)
> at
> org.apache.flink.streaming.runtime.io.CheckpointBarrierAligner.processBarrier(CheckpointBarrierAligner.java:177)
> at
> org.apache.flink.streaming.runtime.io.CheckpointedInputGate.pollNext(CheckpointedInputGate.java:155)
> at
> org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.pollNextNullable(StreamTaskNetworkInput.java:102)
> at
> org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.pollNextNullable(StreamTaskNetworkInput.java:47)
> at
> org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:135)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:279)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.run(StreamTask.java:321)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.runAndHandleCancel(StreamTask.java:286)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:426)
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530)
> at java.lang.Thread.run(Thread.java:748)
> Caused by: org.apache.kafka.common.errors.TimeoutException: Expiring 53
> record(s) for frontend_event_core-46: 122269 ms has passed since last append
> 2021-03-17 17:46:42,354 INFO org.apache.flink.runtime.taskmanager.Task
> - Attempting to cancel task Source: {operator name}-Event
> (214/240) (69ec48cfb074de8812eb622ffa097233).
> 2021-03-17 17:46:42,354 INFO org.apache.flink.runtime.taskmanager.Task
> - Source: {operator name}-Event (214/240)
> (69ec48cfb074de8812eb622ffa097233) switched from RUNNING to CANCELING.
> 2021-03-17 17:46:42,355 INFO org.apache.flink.runtime.taskmanager.Task
> - Triggering cancellation of task code Source: {operator
> name}-Event (214/240) (69ec48cfb074de8812eb622ffa097233).
> 2021-03-17 17:46:42,358 WARN
> org.apache.flink.streaming.runtime.tasks.StreamTask - Error
> while canceling task.
> java.lang.Exception:
> org.apache.flink.streaming.connectors.kafka.internal.Handover$ClosedException
> at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.checkThrowSourceExecutionException(SourceStreamTask.java:232)
> at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask.processInput(SourceStreamTask.java:133)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.run(StreamTask.java:321)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.runAndHandleCancel(StreamTask.java:286)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:426)
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530)
> at java.lang.Thread.run(Thread.java:748)
> Caused by:
> org.apache.flink.streaming.connectors.kafka.internal.Handover$ClosedException
> at
> org.apache.flink.streaming.connectors.kafka.internal.Handover.close(Handover.java:182)
> at
> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.cancel(KafkaFetcher.java:175)
> at
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.cancel(FlinkKafkaConsumerBase.java:818)
> at
> com.pinterest.xenon.unified.XenonUnifiedSource.cancel(XenonUnifiedSource.java:436)
> at
> org.apache.flink.streaming.api.operators.StreamSource.cancel(StreamSource.java:134)
> at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask.cancelTask(SourceStreamTask.java:158)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.cancel(StreamTask.java:528)
> at org.apache.flink.runtime.taskmanager.Task$Task
> 2021-03-17 17:46:42,391 INFO org.apache.flink.runtime.taskmanager.Task
> - Attempting to cancel task Process-Event ->
> Filter-data08 (237/240) (34cf65b29f7153d6f0a82819eeaf218d).
>
>
> On Thu, Mar 11, 2021 at 7:12 AM Rainie Li <raini...@pinterest.com> wrote:
>
>> Thanks for the suggestion, Arvid.
>> Currently my job is using producer.kafka.request.timeout.ms=90000
>> I will try to increase to 120000.
>>
>> Best regards
>> Rainie
>>
>> On Thu, Mar 11, 2021 at 3:58 AM Arvid Heise <ar...@apache.org> wrote:
>>
>>> Hi Rainie,
>>>
>>> This looks like the record batching in Kafka producer timed out. At this
>>> point, the respective records are lost forever. You probably want to tweak
>>> your Kafka settings [1].
>>>
>>> Usually, Flink should fail and restart at this point and recover without
>>> data loss. However, if the transactions are also timing out, that may
>>> explain the data loss. So you probably also want to increase the
>>> transaction timeout.
>>>
>>> [1]
>>> https://stackoverflow.com/questions/53223129/kafka-producer-timeoutexception
>>>
>>> On Mon, Mar 8, 2021 at 8:34 PM Rainie Li <raini...@pinterest.com> wrote:
>>>
>>>> Thanks for the info, David.
>>>> The job has checkpointing.
>>>> I saw some tasks failed due to
>>>> "org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Failed to
>>>> send data to Kafka"
>>>> Here is stacktrack from JM log:
>>>>
>>>> container_e17_1611597945897_8007_01_000240 @ worker-node-host
>>>> (dataPort=42321).
>>>> 2021-02-10 01:19:27,206 INFO
>>>> org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Discarding
>>>> checkpoint 21355 of job 7dab4c1a1c6984e70732b8e3f218020f.
>>>> org.apache.flink.runtime.checkpoint.CheckpointException: Could not
>>>> complete snapshot 21355 for operator Sink: Sink-data08 (208/240). Failure
>>>> reason: Checkpoint was declined.
>>>> at
>>>> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:431)
>>>> at
>>>> org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.checkpointStreamOperator(StreamTask.java:1302)
>>>> at
>>>> org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.executeCheckpointing(StreamTask.java:1236)
>>>> at
>>>> org.apache.flink.streaming.runtime.tasks.StreamTask.checkpointState(StreamTask.java:892)
>>>> at
>>>> org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:797)
>>>> at
>>>> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:728)
>>>> at
>>>> org.apache.flink.streaming.runtime.io.CheckpointBarrierHandler.notifyCheckpoint(CheckpointBarrierHandler.java:88)
>>>> at
>>>> org.apache.flink.streaming.runtime.io.CheckpointBarrierAligner.processBarrier(CheckpointBarrierAligner.java:177)
>>>> at
>>>> org.apache.flink.streaming.runtime.io.CheckpointedInputGate.pollNext(CheckpointedInputGate.java:155)
>>>> at
>>>> org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.pollNextNullable(StreamTaskNetworkInput.java:102)
>>>> at
>>>> org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.pollNextNullable(StreamTaskNetworkInput.java:47)
>>>> at
>>>> org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:135)
>>>> at
>>>> org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:279)
>>>> at
>>>> org.apache.flink.streaming.runtime.tasks.StreamTask.run(StreamTask.java:321)
>>>> at
>>>> org.apache.flink.streaming.runtime.tasks.StreamTask.runAndHandleCancel(StreamTask.java:286)
>>>> at
>>>> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:426)
>>>> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705)
>>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530)
>>>> at java.lang.Thread.run(Thread.java:748)
>>>> Caused by:
>>>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Failed to
>>>> send data to Kafka: Expiring 42 record(s) for topic-name-38: 116447 ms has
>>>> passed since last append
>>>> at
>>>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.checkErroneous(FlinkKafkaProducer.java:1196)
>>>> at
>>>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.flush(FlinkKafkaProducer.java:968)
>>>> at
>>>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.preCommit(FlinkKafkaProducer.java:892)
>>>> at
>>>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.preCommit(FlinkKafkaProducer.java:98)
>>>> at
>>>> org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.snapshotState(TwoPhaseCommitSinkFunction.java:310)
>>>> at
>>>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.snapshotState(FlinkKafkaProducer.java:973)
>>>> at
>>>> org.apache.flink.streaming.util.functions.StreamingFunctionUtils.trySnapshotFunctionState(StreamingFunctionUtils.java:118)
>>>> at
>>>> org.apache.flink.streaming.util.functions.StreamingFunctionUtils.snapshotFunctionState(StreamingFunctionUtils.java:99)
>>>> at
>>>> org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.snapshotState(AbstractUdfStreamOperator.java:90)
>>>> at
>>>> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:399)
>>>> ... 18 more
>>>> Caused by: org.apache.kafka.common.errors.TimeoutException: Expiring 42
>>>> record(s) for frontend_event_core-38: 116447 ms has passed since last
>>>> append
>>>> 2021-02-10 01:19:27,216 INFO
>>>> org.apache.flink.runtime.executiongraph.ExecutionGraph - Job
>>>> (7dab4c1a1c6984e70732b8e3f218020f) switched from state RUNNING to FAILING.
>>>> org.apache.flink.util.FlinkRuntimeException: Exceeded checkpoint
>>>> tolerable failure threshold.
>>>> at
>>>> org.apache.flink.runtime.checkpoint.CheckpointFailureManager.handleTaskLevelCheckpointException(CheckpointFailureManager.java:87)
>>>> at
>>>> org.apache.flink.runtime.checkpoint.CheckpointCoordinator.failPendingCheckpointDueToTaskFailure(CheckpointCoordinator.java:1410)
>>>> at
>>>> org.apache.flink.runtime.checkpoint.CheckpointCoordinator.discardCheckpoint(CheckpointCoordinator.java:1320)
>>>> at
>>>> org.apache.flink.runtime.checkpoint.CheckpointCoordinator.receiveDeclineMessage(CheckpointCoordinator.java:689)
>>>> at
>>>> org.apache.flink.runtime.scheduler.LegacyScheduler.lambda$declineCheckpoint$2(LegacyScheduler.java:573)
>>>> at
>>>> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
>>>> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
>>>> at
>>>> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
>>>> at
>>>> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
>>>> 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)
>>>>
>>>> Best regards
>>>> Rainie
>>>>
>>>> On Mon, Mar 8, 2021 at 11:09 AM David Anderson <dander...@apache.org>
>>>> wrote:
>>>>
>>>>> Rainie,
>>>>>
>>>>> A restart after a failure can cause data loss if you aren't using
>>>>> checkpointing, or if you experience a transaction timeout.
>>>>>
>>>>> A manual restart can also lead to data loss, depending on how you
>>>>> manage the offsets, transactions, and other state during the restart. What
>>>>> happened in this case?
>>>>>
>>>>> David
>>>>>
>>>>> On Mon, Mar 8, 2021 at 7:53 PM Rainie Li <raini...@pinterest.com>
>>>>> wrote:
>>>>>
>>>>>> Thanks Yun and David.
>>>>>> There were some tasks that got restarted. We configured the restart
>>>>>> policy and the job didn't fail.
>>>>>> Will task restart cause data loss?
>>>>>>
>>>>>> Thanks
>>>>>> Rainie
>>>>>>
>>>>>>
>>>>>> On Mon, Mar 8, 2021 at 10:42 AM David Anderson <dander...@apache.org>
>>>>>> wrote:
>>>>>>
>>>>>>> Rainie,
>>>>>>>
>>>>>>> Were there any failures/restarts, or is this discrepancy observed
>>>>>>> without any disruption to the processing?
>>>>>>>
>>>>>>> Regards,
>>>>>>> David
>>>>>>>
>>>>>>> On Mon, Mar 8, 2021 at 10:14 AM Rainie Li <raini...@pinterest.com>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> Thanks for the quick response, Smile.
>>>>>>>> I don't use window operators or flatmap.
>>>>>>>> Here is the core logic of my filter, it only iterates on filters
>>>>>>>> list. Will *rebalance() *cause it?
>>>>>>>>
>>>>>>>> Thanks again.
>>>>>>>> Best regards
>>>>>>>> Rainie
>>>>>>>>
>>>>>>>> SingleOutputStreamOperator<SplitterIntermediateRecord<T>>
>>>>>>>> matchedRecordsStream =
>>>>>>>> eventStream
>>>>>>>> .rebalance()
>>>>>>>> .process(new ProcessFunction<T,
>>>>>>>> SplitterIntermediateRecord<T>>() {
>>>>>>>> public void processElement(
>>>>>>>> T element,
>>>>>>>> ProcessFunction<T,
>>>>>>>> SplitterIntermediateRecord<T>>.Context context,
>>>>>>>> Collector<SplitterIntermediateRecord<T>> collector) {
>>>>>>>> for (StreamFilter filter : filters) {
>>>>>>>> if (filter.match(element)) {
>>>>>>>> SubstreamConfig substreamConfig =
>>>>>>>> filter.getSubstreamConfig();
>>>>>>>> SplitterIntermediateRecord<T> result = new
>>>>>>>> SplitterIntermediateRecord<>(
>>>>>>>> substreamConfig.getKafkaCluster(),
>>>>>>>> substreamConfig.getKafkaTopic(),
>>>>>>>> substreamConfig.getCutoverKafkaTopic(),
>>>>>>>> substreamConfig.getCutoverTimestampInMs(),
>>>>>>>> element);
>>>>>>>> collector.collect(result);
>>>>>>>> }
>>>>>>>> }
>>>>>>>> }
>>>>>>>> })
>>>>>>>> .name("Process-" + eventClass.getSimpleName());
>>>>>>>>
>>>>>>>>
>>>>>>>> On Mon, Mar 8, 2021 at 1:03 AM Smile <letters_sm...@163.com> wrote:
>>>>>>>>
>>>>>>>>> Hi Rainie,
>>>>>>>>>
>>>>>>>>> Could you please provide more information about your processing
>>>>>>>>> logic?
>>>>>>>>> Do you use window operators？
>>>>>>>>> If there's no time-based operator in your logic, late arrival data
>>>>>>>>> won't be
>>>>>>>>> dropped by default and there might be something wrong with your
>>>>>>>>> flat map or
>>>>>>>>> filter operator. Otherwise, you can use sideOutputLateData() to
>>>>>>>>> get the late
>>>>>>>>> data of the window and have a look at them. See [1] for more
>>>>>>>>> information
>>>>>>>>> about sideOutputLateData().
>>>>>>>>>
>>>>>>>>> [1].
>>>>>>>>>
>>>>>>>>> https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/stream/operators/windows.html#getting-late-data-as-a-side-output
>>>>>>>>>
>>>>>>>>> Regards,
>>>>>>>>> Smile
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> --
>>>>>>>>> Sent from:
>>>>>>>>> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
>>>>>>>>>
>>>>>>>>
