8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2025-12-24 11:52:08 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

[Bugfix]修复Connect-GroupDescription解析失败的问题(#1010)

Browse Source 
1、先尝试使用IncrementalCooperativeConnectProtocol协议进行解析；
2、IncrementalCooperativeConnectProtocol协议解析失败后，再维持原先的情况，使用ConnectProtocol协议进行解析；
This commit is contained in:
ZQKC
2023-05-12 14:25:20 +08:00
committed by EricZeng
parent e975932d41
commit 2256e8bbdb
1 changed files with 33 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/kafka/KSPartialKafkaAdminClient.java
View File

@@ -78,6 +78,8 @@ import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils; import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.connect.runtime.distributed.ConnectProtocol; 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 org.slf4j.Logger;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
@@ -1342,19 +1344,7 @@ public class KSPartialKafkaAdminClient {
memberBaseAssignment = new KSMemberConsumerAssignment(new HashSet<>()); memberBaseAssignment = new KSMemberConsumerAssignment(new HashSet<>());
} }
} else { } else {
ConnectProtocol.Assignment assignment = null; memberBaseAssignment = deserializeConnectGroupDataCompatibility(groupMember);
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);
} }
memberDescriptions.add(new KSMemberDescription( memberDescriptions.add(new KSMemberDescription(
@@ -1383,6 +1373,36 @@ public class KSPartialKafkaAdminClient {
}; };
} }
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) { private Set<AclOperation> validAclOperations(final int authorizedOperations) {
if (authorizedOperations == MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED) { if (authorizedOperations == MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED) {