@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