You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Fangliang Liu (Jira)" <ji...@apache.org> on 2022/03/02 13:02:00 UTC
[jira] [Comment Edited] (FLINK-26380) java.lang.IllegalStateException:Pending record count must be zero at this point: 5592
[ https://issues.apache.org/jira/browse/FLINK-26380?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17500129#comment-17500129 ]
Fangliang Liu edited comment on FLINK-26380 at 3/2/22, 1:01 PM:
----------------------------------------------------------------
Hi [~fpaul], thanks for your replay.
Checkpoint settings
{code:java}
execution.checkpointing.interval 60s
execution.checkpointing.timeout 60s
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);{code}
FlinkKafkaProducer settings
{code:java}
Properties properties = new Properties();
properties.setProperty("transaction.timeout.ms", 1000 * 60 * 5 + "");
// 重试次数3次
properties.setProperty("retries", 3 + "");
// 设置FlinkKafkaProducer的Semantic为精准一致性
FlinkKafkaProducer<Record> kafkaProducer = new FlinkKafkaProducer<>(topic,
new FilterRecordSerializationSchema(topic, appContext),
properties,
FlinkKafkaProducer.Semantic.EXACTLY_ONCE, 5);
return kafkaProducer;
} {code}
Please let me know if you need any other information.
was (Author: liufangliang):
Hi [~fpaul], thanks for your replay.
Checkpoint settings
{code:java}
execution.checkpointing.interval 60s
execution.checkpointing.timeout 60s
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);{code}
FlinkKafkaProducer settings
{code:java}
Properties properties = new Properties();
properties.setProperty("transaction.timeout.ms", 1000 * 60 * 5 + "");
// 重试次数3次
properties.setProperty("retries", 3 + "");
// 设置FlinkKafkaProducer的Semantic为精准一致性
FlinkKafkaProducer<Record> kafkaProducer = new FlinkKafkaProducer<>(topic,
new FilterRecordSerializationSchema(topic, appContext),
properties,
FlinkKafkaProducer.Semantic.EXACTLY_ONCE, 5);
return kafkaProducer;
} {code}
Please let me know if you need any other information.
> java.lang.IllegalStateException:Pending record count must be zero at this point: 5592
> -------------------------------------------------------------------------------------
>
> Key: FLINK-26380
> URL: https://issues.apache.org/jira/browse/FLINK-26380
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: 1.13.2
> Reporter: Fangliang Liu
> Priority: Major
>
> Hi,[~lzljs3620320], [~becket_qin] ,[~jark] .
> Help introduce developers who are familiar with checkpoint and kafka connector.
> The _exactly-once_ of kafka connector and checkpoint is turned on, and the reason for the following exception occurs ？
> Main exception information:
> {code:java}
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Failed to send data to Kafka: Pending record count must be zero at this point: 5592 {code}
> Complete exception information:
> {code:java}
> java.io.IOException: Could not perform checkpoint 2273 for operator filterRuleProcess -> Sink: data_filter_sink (1/1)#1477. at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:1045) at org.apache.flink.streaming.runtime.io.checkpointing.CheckpointBarrierHandler.notifyCheckpoint(CheckpointBarrierHandler.java:135) at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.triggerCheckpoint(SingleCheckpointBarrierHandler.java:250) at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.access$100(SingleCheckpointBarrierHandler.java:61) at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler$ControllerImpl.triggerGlobalCheckpoint(SingleCheckpointBarrierHandler.java:431) at org.apache.flink.streaming.runtime.io.checkpointing.AbstractAlignedBarrierHandlerState.barrierReceived(AbstractAlignedBarrierHandlerState.java:61) at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.processBarrier(SingleCheckpointBarrierHandler.java:227) at org.apache.flink.streaming.runtime.io.checkpointing.CheckpointedInputGate.handleEvent(CheckpointedInputGate.java:180) at org.apache.flink.streaming.runtime.io.checkpointing.CheckpointedInputGate.pollNext(CheckpointedInputGate.java:158) at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:110) at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66) at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:423) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:204) at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:681) at org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:636) at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:647) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:620) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:779) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566) at java.lang.Thread.run(Thread.java:748) Suppressed: org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Failed to send data to Kafka: Pending record count must be zero at this point: 5592 at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.checkErroneous(FlinkKafkaProducer.java:1392) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.close(FlinkKafkaProducer.java:965) at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:41) at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.dispose(AbstractUdfStreamOperator.java:117) at org.apache.flink.streaming.runtime.tasks.StreamTask.disposeAllOperators(StreamTask.java:861) at org.apache.flink.streaming.runtime.tasks.StreamTask.runAndSuppressThrowable(StreamTask.java:840) at org.apache.flink.streaming.runtime.tasks.StreamTask.cleanUpInvoke(StreamTask.java:753) at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:659) ... 4 more Caused by: java.lang.IllegalStateException: Pending record count must be zero at this point: 5592 at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.flush(FlinkKafkaProducer.java:1090) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.close(FlinkKafkaProducer.java:925) ... 10 moreCaused by: org.apache.flink.runtime.checkpoint.CheckpointException: Could not complete snapshot 2273 for operator filterRuleProcess -> Sink: data_filter_sink (1/1)#1477. Failure reason: Checkpoint was declined. at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:264) at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:169) at org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:371) at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointStreamOperator(SubtaskCheckpointCoordinatorImpl.java:706) at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.buildOperatorSnapshotFutures(SubtaskCheckpointCoordinatorImpl.java:627) at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.takeSnapshotSync(SubtaskCheckpointCoordinatorImpl.java:590) at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:312) at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$8(StreamTask.java:1089) at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50) at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:1073) at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:1029) ... 19 moreCaused by: java.lang.IllegalStateException: Pending record count must be zero at this point: 5592 at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.flush(FlinkKafkaProducer.java:1090) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.preCommit(FlinkKafkaProducer.java:1002) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.preCommit(FlinkKafkaProducer.java:99) at org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.snapshotState(TwoPhaseCommitSinkFunction.java:320) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.snapshotState(FlinkKafkaProducer.java:1100) 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:89) at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:218) ... 29 more {code}
>
>
--
This message was sent by Atlassian Jira
(v8.20.1#820001)