Merge branch 'hotfix_1.8_3.10.x_23728' into 1.8_release_3.10.x

Author: gituser

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

@@ -76,8 +76,7 @@ protected void beforeDeserialize() throws IOException {
     }
 
     protected void registerPtMetric(AbstractFetcher<Row, ?> fetcher) throws Exception {
-
-        Field consumerThreadField = fetcher.getClass().getSuperclass().getDeclaredField("consumerThread");
+        Field consumerThreadField = getConsumerThreadField(fetcher);
         consumerThreadField.setAccessible(true);
         KafkaConsumerThread consumerThread = (KafkaConsumerThread) consumerThreadField.get(fetcher);
 
@@ -118,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");
+        }
+    }
 }