Repository: didi/KnowStreaming (language: java)

Issue:
Error reading field 'leader'
- [x] 我已经在 [issues](https://github.com/didi/KnowStreaming/issues) 搜索过相关问题了,并没有重复的。

 你是否希望来认领这个Bug。

 「   Y   」

### 环境信息

* KnowStreaming version :   <font size=4 color =red> latest </font>
* Operating System version :  <font size=4 color =red> docker </font>
* Java version : <font size=4 color =red> 1.8 </font>
* Confluent version : <font size=4 color =red> 7.0.1 </font>
* Kafka version : <font size=4 color =red> 3.0.x </font>


### 重现该问题的步骤

1. 下載docker-compose.yml文件,修改ui映射端口為8080:80
  
2. docker-compose up -d
   

3. 訪問http://localhost:8080,admin/admin登錄,接入集群



### 预期结果

可以查看集群相關信息,log不報錯

### 实际结果

可以查看集群相關信息,log報錯


---

如果有异常，请附上异常Trace:

```
2023-05-09 11:32:24 2023-05-09 11:32:24.741 ERROR 12 --- [skTP-5-thread-3] c.x.k.s.km.common.utils.FutureWaitUtil   : method=stepWaitResult||taskName=class=ClusterMetricCollector||clusterPhyId=1||metricName=BytesIn||errMsg=exception
2023-05-09 11:32:24 
2023-05-09 11:32:24 java.util.concurrent.TimeoutException: null
2023-05-09 11:32:24 at java.util.concurrent.FutureTask.get(FutureTask.java:205)
2023-05-09 11:32:24 at com.xiaojukeji.know.streaming.km.common.utils.FutureWaitUtil.stepWaitResult(FutureWaitUtil.java:155)
2023-05-09 11:32:24 at com.xiaojukeji.know.streaming.km.common.utils.FutureWaitUtil.waitResult(FutureWaitUtil.java:116)
2023-05-09 11:32:24 at com.xiaojukeji.know.streaming.km.common.utils.FutureWaitUtil.waitExecute(FutureWaitUtil.java:76)
2023-05-09 11:32:24 at com.xiaojukeji.know.streaming.km.collector.metric.kafka.ClusterMetricCollector.collectKafkaMetrics(ClusterMetricCollector.java:74)
2023-05-09 11:32:24 at com.xiaojukeji.know.streaming.km.collector.metric.kafka.AbstractKafkaMetricCollector.collectMetrics(AbstractKafkaMetricCollector.java:37)
2023-05-09 11:32:24 at com.xiaojukeji.know.streaming.km.task.kafka.metrics.ClusterMetricCollectorTask.processClusterTask(ClusterMetricCollectorTask.java:25)
2023-05-09 11:32:24 at com.xiaojukeji.know.streaming.km.task.kafka.metrics.AbstractAsyncMetricsDispatchTask.lambda$asyncProcessSubTask$0(AbstractAsyncMetricsDispatchTask.java:33)
2023-05-09 11:32:24 at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
2023-05-09 11:32:24 at java.util.concurrent.FutureTask.run(FutureTask.java:266)
2023-05-09 11:32:24 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
2023-05-09 11:32:24 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
2023-05-09 11:32:24 at java.lang.Thread.run(Thread.java:745)
2023-05-09 11:32:24 
2023-05-09 11:32:24 2023-05-09 11:32:24.741 ERROR 12 --- [skTP-5-thread-3] c.x.k.s.km.common.utils.FutureWaitUtil   : method=waitExecute||taskName=class=ClusterMetricCollector||clusterPhyId=1||metricName=BytesIn||msg=cancel task
2023-05-09 11:32:25 2023-05-09 11:32:25.600 ERROR 12 --- [kTP-6-thread-27] c.x.k.s.k.c.s.g.impl.GroupServiceImpl    : method=getGroupDescription||clusterPhyId=1|groupName=schema-registry||errMsg=exception!
2023-05-09 11:32:25 
2023-05-09 11:32:25 java.util.concurrent.ExecutionException: org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'leader': Error reading string of length 29295, only 194 bytes available
2023-05-09 11:32:25 at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
2023-05-09 11:32:25 at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
2023-05-09 11:32:25 at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
2023-05-09 11:32:25 at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
2023-05-09 11:32:25 at com.xiaojukeji.know.streaming.km.core.service.group.impl.GroupServiceImpl.getGroupDescriptionFromKafka(GroupServiceImpl.java:188)
2023-05-09 11:32:25 at com.xiaojukeji.know.streaming.km.core.service.group.impl.GroupServiceImpl.getGroupFromKafka(GroupServiceImpl.java:112)
2023-05-09 11:32:25 at com.xiaojukeji.know.streaming.km.task.kafka.metadata.SyncKafkaGroupTask.processClusterTask(SyncKafkaGroupTask.java:45)
2023-05-09 11:32:25 at com.xiaojukeji.know.streaming.km.task.kafka.metadata.AbstractAsyncMetadataDispatchTask.lambda$asyncProcessSubTask$0(AbstractAsyncMetadataDispatchTask.java:33)
2023-05-09 11:32:25 at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
2023-05-09 11:32:25 at java.util.concurrent.FutureTask.run(FutureTask.java:266)
2023-05-09 11:32:25 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
2023-05-09 11:32:25 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
2023-05-09 11:32:25 at java.lang.Thread.run(Thread.java:745)
2023-05-09 11:32:25 Caused by: org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'leader': Error reading string of length 29295, only 194 bytes available
2023-05-09 11:32:25 at org.apache.kafka.common.protocol.types.Schema.read(Schema.java:118)
2023-05-09 11:32:25 at org.apache.kafka.connect.runtime.distributed.ConnectProtocol.deserializeAssignment(ConnectProtocol.java:217)
2023-05-09 11:32:25 at com.xiaojukeji.know.streaming.km.common.utils.kafka.KSPartialKafkaAdminClient$2.handleResponse(KSPartialKafkaAdminClient.java:1346)
2023-05-09 11:32:25 at com.xiaojukeji.know.streaming.km.common.utils.kafka.KSPartialKafkaAdminClient$AdminClientRunnable.handleResponses(KSPartialKafkaAdminClient.java:944)
2023-05-09 11:32:25 at com.xiaojukeji.know.streaming.km.common.utils.kafka.KSPartialKafkaAdminClient$AdminClientRunnable.processRequests(KSPartialKafkaAdminClient.java:1096)
2023-05-09 11:32:25 at com.xiaojukeji.know.streaming.km.common.utils.kafka.KSPartialKafkaAdminClient$AdminClientRunnable.run(KSPartialKafkaAdminClient.java:1019)
2023-05-09 11:32:25 ... 1 common frames omitted
```



Discussion / hints:
应该是没有兼容 IncrementalCooperativeConnectProtocol 导致的，后续兼容一下～

Fix patch:
diff --git a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/kafka/KSPartialKafkaAdminClient.java b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/kafka/KSPartialKafkaAdminClient.java
index db8093058..8f53d9988 100644
--- a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/kafka/KSPartialKafkaAdminClient.java
+++ b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/kafka/KSPartialKafkaAdminClient.java
@@ -78,6 +78,8 @@
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.connect.runtime.distributed.ConnectProtocol;
+import org.apache.kafka.connect.runtime.distributed.ExtendedWorkerState;
+import org.apache.kafka.connect.runtime.distributed.IncrementalCooperativeConnectProtocol;
 import org.slf4j.Logger;
 
 import java.net.InetSocketAddress;
@@ -1342,19 +1344,7 @@ void handleResponse(AbstractResponse abstractResponse) {
                             memberBaseAssignment = new KSMemberConsumerAssignment(new HashSet<>());
                         }
                     } else {
-                        ConnectProtocol.Assignment assignment = null;
-                        if (groupMember.memberAssignment().length > 0) {
-                            assignment = ConnectProtocol.
-                                    deserializeAssignment(ByteBuffer.wrap(groupMember.memberAssignment()));
-                        }
-
-                        ConnectProtocol.WorkerState workerState = null;
-                        if (groupMember.memberMetadata().length > 0) {
-                            workerState = ConnectProtocol.
-                                    deserializeMetadata(ByteBuffer.wrap(groupMember.memberMetadata()));
-                        }
-
-                        memberBaseAssignment = new KSMemberConnectAssignment(assignment, workerState);
+                        memberBaseAssignment = deserializeConnectGroupDataCompatibility(groupMember);
                     }
 
                     memberDescriptions.add(new KSMemberDescription(
@@ -1383,6 +1373,36 @@ void handleFailure(Throwable throwable) {
         };
     }
 
+    private KSMemberBaseAssignment deserializeConnectGroupDataCompatibility(DescribedGroupMember groupMember) {
+        try {
+            // 高版本的反序列化方式
+            ExtendedWorkerState workerState = null;
+            if (groupMember.memberMetadata().length > 0) {
+                workerState = IncrementalCooperativeConnectProtocol.
+                        deserializeMetadata(ByteBuffer.wrap(groupMember.memberMetadata()));
+
+                return new KSMemberConnectAssignment(workerState.assignment(), workerState);
+            }
+        } catch (Exception e) {
+            // ignore
+        }
+
+        // 低版本的反序列化方式
+        ConnectProtocol.Assignment assignment = null;
+        if (groupMember.memberAssignment().length > 0) {
+            assignment = ConnectProtocol.
+                    deserializeAssignment(ByteBuffer.wrap(groupMember.memberAssignment()));
+        }
+
+        ConnectProtocol.WorkerState workerState = null;
+        if (groupMember.memberMetadata().length > 0) {
+            workerState = ConnectProtocol.
+                    deserializeMetadata(ByteBuffer.wrap(groupMember.memberMetadata()));
+        }
+
+        return new KSMemberConnectAssignment(assignment, workerState);
+    }
+
 
     private Set<AclOperation> validAclOperations(final int authorizedOperations) {
         if (authorizedOperations == MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED) {


Test patch: