public void setFetcher(AbstractFetcher<Row, ?> fetcher) {
        try {
            registerPtMetric(fetcher);
        } catch (Exception e) {
            LOG.error("register topic partition metric error.", e);
        }
    }
add metric for source
<p>
company: www.dtstack.com
@author: toutian
create: 2019/12/24
kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/deserialization/KafkaDeserializationMetricWrapper.java