fix 类型判断bug

Author: dapeng

kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaDeserializationMetricWrapper.java

@@ -76,13 +76,7 @@ protected void beforeDeserialize() throws IOException {
     }
 
     protected void registerPtMetric(AbstractFetcher<Row, ?> fetcher) throws Exception {
-        Field consumerThreadField = null;
-        if(fetcher.getClass().getDeclaredField("consumerThread") != null){
-            consumerThreadField = fetcher.getClass().getDeclaredField("consumerThread");
-        }  else {
-            consumerThreadField = fetcher.getClass().getSuperclass().getDeclaredField("consumerThread");
-        }
-
+        Field consumerThreadField = getConsumerThreadField(fetcher);
         consumerThreadField.setAccessible(true);
         KafkaConsumerThread consumerThread = (KafkaConsumerThread) consumerThreadField.get(fetcher);
 
@@ -123,4 +117,12 @@ public Long getValue() {
     public void setFetcher(AbstractFetcher<Row, ?> fetcher) {
         this.fetcher = fetcher;
     }
+
+    private Field getConsumerThreadField(AbstractFetcher fetcher) throws NoSuchFieldException {
+        try {
+            return fetcher.getClass().getDeclaredField("consumerThread");
+        } catch (Exception e) {
+            return fetcher.getClass().getSuperclass().getDeclaredField("consumerThread");
+        }
+    }
 }