[Improve][Sort] Refactor MetricOption code structure.

[XiaoYou201]

Description

The MetricOption constructor is not concise. checkArgument method in build() will be better.

inlong/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricOption.java

Lines 59 to 108 in bf42e1a

	private MetricOption(
	String inlongLabels,
	@Nullable String inlongAudit,
	RegisteredMetric registeredMetric,
	long initRecords,
	long initBytes,
	Long initDirtyRecords,
	Long initDirtyBytes,
	Long readPhase,
	String inlongAuditKeys) {
	Preconditions.checkArgument(!StringUtils.isNullOrWhitespaceOnly(inlongLabels),
	"Inlong labels must be set for register metric.");
	this.initRecords = initRecords;
	this.initBytes = initBytes;
	this.initDirtyRecords = initDirtyRecords;
	this.initDirtyBytes = initDirtyBytes;
	this.readPhase = readPhase;
	this.labels = new LinkedHashMap<>();
	String[] inLongLabelArray = inlongLabels.split(DELIMITER);
	Preconditions.checkArgument(Stream.of(inLongLabelArray).allMatch(label -> label.contains("=")),
	"InLong metric label format must be xxx=xxx");
	Stream.of(inLongLabelArray).forEach(label -> {
	String key = label.substring(0, label.indexOf('='));
	String value = label.substring(label.indexOf('=') + 1);
	labels.put(key, value);
	});
	this.ipPorts = inlongAudit;
	if (ipPorts != null) {
	Preconditions.checkArgument(labels.containsKey(GROUP_ID) && labels.containsKey(STREAM_ID),
	"groupId and streamId must be set when enable inlong audit collect.");
	if (inlongAuditKeys == null) {
	LOG.warn("should set inlongAuditKeys when enable inlong audit collect, "
	+ "fallback to use id {} as audit key", AUDIT_SORT_INPUT);
	inlongAuditKeys = AUDIT_SORT_INPUT;
	}
	this.inlongAuditKeys = AuditUtils.extractAuditKeys(inlongAuditKeys);
	this.ipPortList = AuditUtils.extractAuditIpPorts(ipPorts);
	}
	if (registeredMetric != null) {
	this.registeredMetric = registeredMetric;
	}
	}

inlong/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricOption.java

Lines 240 to 246 in 3b70a93

	public MetricOption build() {
	if (inlongLabels == null && inlongAudit == null) {
	return null;
	}
	return new MetricOption(inlongLabels, inlongAudit, registeredMetric, initRecords, initBytes,
	initDirtyRecords, initDirtyBytes, initReadPhase, inlongAuditKeys);
	}

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