修复kafka-source 作为数据源错误，type='kakfa'

Author: dapeng

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

@@ -76,8 +76,13 @@ 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 = fetcher.getClass().getSuperclass().getDeclaredField("consumerThread");
         consumerThreadField.setAccessible(true);
         KafkaConsumerThread consumerThread = (KafkaConsumerThread) consumerThreadField.get(fetcher);