@Override
    public void init(InstanceProfile profile) {
        this.profile = profile;
        jobInstanceId = profile.getInstanceId();
        inlongGroupId = profile.getInlongGroupId();
        inlongStreamId = profile.getInlongStreamId();
        cache = new ProxyMessageCache(this.profile, inlongGroupId, inlongStreamId);
        batchFlushInterval = agentConf.getInt(PROXY_BATCH_FLUSH_INTERVAL, DEFAULT_PROXY_BATCH_FLUSH_INTERVAL);

        this.dimensions = new HashMap<>();
        dimensions.put(KEY_PLUGIN_ID, this.getClass().getSimpleName());
        dimensions.put(KEY_INLONG_GROUP_ID, inlongGroupId);
        dimensions.put(KEY_INLONG_STREAM_ID, inlongStreamId);
        String metricName = String.join("-", this.getClass().getSimpleName(),
                String.valueOf(METRIC_INDEX.incrementAndGet()));
        this.metricItemSet = new AgentMetricItemSet(metricName);
        MetricRegister.register(metricItemSet);
        sinkMetric = metricItemSet.findMetricItem(dimensions);
    }
abstract sink: sink data to remote data center
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/AbstractSink.java