[Feature][Sort] The Pulsar connector on flink1.18 not set audit time, it should be set as input time through 'consume_time' metadata field

[XiaoYou201]

Description

The SourceMetricData not set timestamp as input time, So the report audit time will be wrong.It should be set through 'consume_time' metadata field as follow.

inlong/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/ExtractNode.java

Lines 73 to 79 in 5d4128d

	public abstract class ExtractNode implements Node {
	public static final String INLONG_MSG = "inlong-msg";
	public static final String INLONG_MSG_AUDIT_TIME = "value.data-time";
	public static final String CONSUME_AUDIT_TIME = "consume_time";

reference kafka #10393

Use case

Pulsar source connector not attach input time when send audit information.

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