[INLONG-7567][Sort] Extract metrics as common parameters

[e-mhui]

Prepare a Pull Request

Extract metrics as common parameters

• Fixes [Improve][Sort] Extract metrics as common parameters #7567

Motivation

In order to better adapt to metrics from other CDCs, make metric parameters as common parameter. Other CDC's SourceConfig only need to implements MetricConfig.

Modifications

1. Add the metricConfig interface

public interface MetricConfig extends Serializable {

    String getInlongMetric();

    String getInlongAudit();

}

2. Put metric parameters inside BaseSourceConfig and implements MetricConfig.

public abstract class BaseSourceConfig implements SourceConfig, MetricConfig {

        /****/

    // --------------------------------------------------------------------------------------------
    // Metric Configurations
    // --------------------------------------------------------------------------------------------
    protected final String inlongMetric;
    protected final String inlongAudit;

        /****/

    @Override
    public String getInlongMetric() {
        return inlongMetric;
    }

    @Override
    public String getInlongAudit() {
        return inlongAudit;
    }

}

3. In the method of registering metrics, use MetricConfig metricConfig = (MetricConfig) sourceConfig; to obtain metric parameters.

   @Override
    public IncrementalSourceReader<T, C> createReader(SourceReaderContext readerContext)
            throws Exception {
        // create source config for the given subtask (e.g. unique server id)
        C sourceConfig = configFactory.create(readerContext.getIndexOfSubtask());
        MetricConfig metricConfig = (MetricConfig) sourceConfig;
        /****/

        // create source config for the given subtask (e.g. unique server id)
        MetricOption metricOption = MetricOption.builder()
                .withInlongLabels(metricConfig.getInlongMetric())
                .withAuditAddress(metricConfig.getInlongAudit())
                .withRegisterMetric(RegisteredMetric.ALL)
                .build();

        sourceReaderMetrics.registerMetrics(metricOption);
        /****/
    }