protected void registerPtMetric(AbstractFetcher<Row, ?> fetcher) throws Exception {
        Field consumerThreadField = ReflectionUtils.getDeclaredField(fetcher, "consumerThread");
        consumerThreadField.setAccessible(true);
        KafkaConsumerThread consumerThread = (KafkaConsumerThread) consumerThreadField.get(fetcher);

        Field hasAssignedPartitionsField = consumerThread.getClass().getDeclaredField("hasAssignedPartitions");
        hasAssignedPartitionsField.setAccessible(true);

        // get subtask unassigned kafka topic partition
        Field subscribedPartitionStatesField = ReflectionUtils.getDeclaredField(fetcher, "subscribedPartitionStates");
        subscribedPartitionStatesField.setAccessible(true);
        List<KafkaTopicPartitionState<KafkaTopicPartition>> subscribedPartitionStates = (List<KafkaTopicPartitionState<KafkaTopicPartition>>) subscribedPartitionStatesField.get(fetcher);

        // init partition lag metric
        for (KafkaTopicPartitionState<KafkaTopicPartition> kafkaTopicPartitionState : subscribedPartitionStates) {
            KafkaTopicPartition kafkaTopicPartition = kafkaTopicPartitionState.getKafkaTopicPartition();
            MetricGroup topicMetricGroup = getRuntimeContext().getMetricGroup().addGroup(DT_TOPIC_GROUP, kafkaTopicPartition.getTopic());

            MetricGroup metricGroup = topicMetricGroup.addGroup(DT_PARTITION_GROUP, kafkaTopicPartition.getPartition() + "");
            metricGroup.gauge(DT_TOPIC_PARTITION_LAG_GAUGE, new Gauge<Long>() {
                // tmp variable
                boolean initLag = true;
                int partitionIndex;
                SubscriptionState subscriptionState;
                TopicPartition topicPartition;

                @Override
                public Long getValue() {
                    // first time register metrics
                    if (initLag) {
                        partitionIndex = kafkaTopicPartition.getPartition();
                        initLag = false;
                        return -1L;
                    }
                    // when kafka topic partition assigned calc metrics
                    if (subscriptionState == null) {
                        try {
                            Field consumerField = consumerThread.getClass().getDeclaredField("consumer");
                            consumerField.setAccessible(true);

                            KafkaConsumer kafkaConsumer = (KafkaConsumer) consumerField.get(consumerThread);
                            Field subscriptionStateField = kafkaConsumer.getClass().getDeclaredField("subscriptions");
                            subscriptionStateField.setAccessible(true);

                            boolean hasAssignedPartitions = (boolean) hasAssignedPartitionsField.get(consumerThread);

                            if (!hasAssignedPartitions) {
                                LOG.error("wait 50 secs, but not assignedPartitions");
                            }

                            subscriptionState = (SubscriptionState) subscriptionStateField.get(kafkaConsumer);

                            topicPartition = subscriptionState
                                    .assignedPartitions()
                                    .stream()
                                    .filter(x -> x.partition() == partitionIndex)
                                    .findFirst()
                                    .get();

                        } catch (Exception e) {
                            LOG.error(e.getMessage());
                        }
                        return -1L;
                    } else {
                        return calculate.calc(subscriptionState, topicPartition);
                    }
                }
            });
        }
    }
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