[Bug][Sort] MetricStateUtils threw NPE when write to es and no dirtyRecordsOut counter

[wangpeix]

What happened

MetricStateUtils threw NPE when write to es and no dirtyRecordsOut counter.

What you expected to happen

java.lang.Exception: Could not perform checkpoint 40 for operator Source: TableSourceScan(table=[[default_catalog, default_database, table_stream_01]], fields=[uid, event_time]) -> Sink: Sink(table=[default_catalog.default_database.node_sink_01_test_index6], fields=[uid, event_time]) (1/1)#39.
	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointAsyncInMailbox(StreamTask.java:1006)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$triggerCheckpointAsync$7(StreamTask.java:958)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93)
	at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:344)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:330)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:202)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:684)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:639)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:623)
	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)
Caused by: org.apache.flink.runtime.checkpoint.CheckpointException: Could not complete snapshot 40 for operator Source: TableSourceScan(table=[[default_catalog, default_database, table_stream_01]], fields=[uid, event_time]) -> Sink: Sink(table=[default_catalog.default_database.node_sink_01_test_index6], fields=[uid, event_time]) (1/1)#39. 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:1092)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:1076)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointAsyncInMailbox(StreamTask.java:994)
	... 13 more
Caused by: java.lang.NullPointerException
	at org.apache.inlong.sort.base.util.MetricStateUtils.snapshotMetricStateForSinkMetricData(MetricStateUtils.java:225)
	at org.apache.inlong.sort.elasticsearch.table.RowElasticsearchSinkFunction.snapshotState(RowElasticsearchSinkFunction.java:142)
	at org.apache.inlong.sort.elasticsearch.ElasticsearchSinkBase.snapshotState(ElasticsearchSinkBase.java:312)
	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)
	... 23 more

How to reproduce

Use the following sql can reproduce the problem.

CREATE TABLE `node_sink_01_test_index6`(
    `uid` STRING,
    `event_time` STRING)
    WITH (
    'inlong.metric.labels' = 'groupId=group_06&streamId=stream_01&nodeId=sink_01',
    'metrics.audit.proxy.hosts' = 'x.x.x.x:10081',
    'connector' = 'elasticsearch-6-inlong',
    'document-type' = 'null',
    'hosts' = 'http://x.x.x.x:9200',
    'index' = 'test_index6',
    'password' = 'xxx',
    'username' = 'xxx'
)

Environment

No response

InLong version

master

InLong Component

InLong Sort

Are you willing to submit PR?

• Yes, I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct