public static AbstractResponse parseResponse(ApiKeys apiKey, ByteBuffer responseBuffer, short version) {
        switch (apiKey) {
            case PRODUCE:
                return ProduceResponse.parse(responseBuffer, version);
            case FETCH:
                return FetchResponse.parse(responseBuffer, version);
            case LIST_OFFSETS:
                return ListOffsetsResponse.parse(responseBuffer, version);
            case METADATA:
                return MetadataResponse.parse(responseBuffer, version);
            case OFFSET_COMMIT:
                return OffsetCommitResponse.parse(responseBuffer, version);
            case OFFSET_FETCH:
                return OffsetFetchResponse.parse(responseBuffer, version);
            case FIND_COORDINATOR:
                return FindCoordinatorResponse.parse(responseBuffer, version);
            case JOIN_GROUP:
                return JoinGroupResponse.parse(responseBuffer, version);
            case HEARTBEAT:
                return HeartbeatResponse.parse(responseBuffer, version);
            case LEAVE_GROUP:
                return LeaveGroupResponse.parse(responseBuffer, version);
            case SYNC_GROUP:
                return SyncGroupResponse.parse(responseBuffer, version);
            case STOP_REPLICA:
                return StopReplicaResponse.parse(responseBuffer, version);
            case CONTROLLED_SHUTDOWN:
                return ControlledShutdownResponse.parse(responseBuffer, version);
            case UPDATE_METADATA:
                return UpdateMetadataResponse.parse(responseBuffer, version);
            case LEADER_AND_ISR:
                return LeaderAndIsrResponse.parse(responseBuffer, version);
            case DESCRIBE_GROUPS:
                return DescribeGroupsResponse.parse(responseBuffer, version);
            case LIST_GROUPS:
                return ListGroupsResponse.parse(responseBuffer, version);
            case SASL_HANDSHAKE:
                return SaslHandshakeResponse.parse(responseBuffer, version);
            case API_VERSIONS:
                return ApiVersionsResponse.parse(responseBuffer, version);
            case CREATE_TOPICS:
                return CreateTopicsResponse.parse(responseBuffer, version);
            case DELETE_TOPICS:
                return DeleteTopicsResponse.parse(responseBuffer, version);
            case DELETE_RECORDS:
                return DeleteRecordsResponse.parse(responseBuffer, version);
            case INIT_PRODUCER_ID:
                return InitProducerIdResponse.parse(responseBuffer, version);
            case OFFSET_FOR_LEADER_EPOCH:
                return OffsetsForLeaderEpochResponse.parse(responseBuffer, version);
            case ADD_PARTITIONS_TO_TXN:
                return AddPartitionsToTxnResponse.parse(responseBuffer, version);
            case ADD_OFFSETS_TO_TXN:
                return AddOffsetsToTxnResponse.parse(responseBuffer, version);
            case END_TXN:
                return EndTxnResponse.parse(responseBuffer, version);
            case WRITE_TXN_MARKERS:
                return WriteTxnMarkersResponse.parse(responseBuffer, version);
            case TXN_OFFSET_COMMIT:
                return TxnOffsetCommitResponse.parse(responseBuffer, version);
            case DESCRIBE_ACLS:
                return DescribeAclsResponse.parse(responseBuffer, version);
            case CREATE_ACLS:
                return CreateAclsResponse.parse(responseBuffer, version);
            case DELETE_ACLS:
                return DeleteAclsResponse.parse(responseBuffer, version);
            case DESCRIBE_CONFIGS:
                return DescribeConfigsResponse.parse(responseBuffer, version);
            case ALTER_CONFIGS:
                return AlterConfigsResponse.parse(responseBuffer, version);
            case ALTER_REPLICA_LOG_DIRS:
                return AlterReplicaLogDirsResponse.parse(responseBuffer, version);
            case DESCRIBE_LOG_DIRS:
                return DescribeLogDirsResponse.parse(responseBuffer, version);
            case SASL_AUTHENTICATE:
                return SaslAuthenticateResponse.parse(responseBuffer, version);
            case CREATE_PARTITIONS:
                return CreatePartitionsResponse.parse(responseBuffer, version);
            case CREATE_DELEGATION_TOKEN:
                return CreateDelegationTokenResponse.parse(responseBuffer, version);
            case RENEW_DELEGATION_TOKEN:
                return RenewDelegationTokenResponse.parse(responseBuffer, version);
            case EXPIRE_DELEGATION_TOKEN:
                return ExpireDelegationTokenResponse.parse(responseBuffer, version);
            case DESCRIBE_DELEGATION_TOKEN:
                return DescribeDelegationTokenResponse.parse(responseBuffer, version);
            case DELETE_GROUPS:
                return DeleteGroupsResponse.parse(responseBuffer, version);
            case ELECT_LEADERS:
                return ElectLeadersResponse.parse(responseBuffer, version);
            case INCREMENTAL_ALTER_CONFIGS:
                return IncrementalAlterConfigsResponse.parse(responseBuffer, version);
            case ALTER_PARTITION_REASSIGNMENTS:
                return AlterPartitionReassignmentsResponse.parse(responseBuffer, version);
            case LIST_PARTITION_REASSIGNMENTS:
                return ListPartitionReassignmentsResponse.parse(responseBuffer, version);
            case OFFSET_DELETE:
                return OffsetDeleteResponse.parse(responseBuffer, version);
            case DESCRIBE_CLIENT_QUOTAS:
                return DescribeClientQuotasResponse.parse(responseBuffer, version);
            case ALTER_CLIENT_QUOTAS:
                return AlterClientQuotasResponse.parse(responseBuffer, version);
            case DESCRIBE_USER_SCRAM_CREDENTIALS:
                return DescribeUserScramCredentialsResponse.parse(responseBuffer, version);
            case ALTER_USER_SCRAM_CREDENTIALS:
                return AlterUserScramCredentialsResponse.parse(responseBuffer, version);
            case VOTE:
                return VoteResponse.parse(responseBuffer, version);
            case BEGIN_QUORUM_EPOCH:
                return BeginQuorumEpochResponse.parse(responseBuffer, version);
            case END_QUORUM_EPOCH:
                return EndQuorumEpochResponse.parse(responseBuffer, version);
            case DESCRIBE_QUORUM:
                return DescribeQuorumResponse.parse(responseBuffer, version);
            case ALTER_PARTITION:
                return AlterPartitionResponse.parse(responseBuffer, version);
            case UPDATE_FEATURES:
                return UpdateFeaturesResponse.parse(responseBuffer, version);
            case ENVELOPE:
                return EnvelopeResponse.parse(responseBuffer, version);
            case FETCH_SNAPSHOT:
                return FetchSnapshotResponse.parse(responseBuffer, version);
            case DESCRIBE_CLUSTER:
                return DescribeClusterResponse.parse(responseBuffer, version);
            case DESCRIBE_PRODUCERS:
                return DescribeProducersResponse.parse(responseBuffer, version);
            case BROKER_REGISTRATION:
                return BrokerRegistrationResponse.parse(responseBuffer, version);
            case BROKER_HEARTBEAT:
                return BrokerHeartbeatResponse.parse(responseBuffer, version);
            case UNREGISTER_BROKER:
                return UnregisterBrokerResponse.parse(responseBuffer, version);
            case DESCRIBE_TRANSACTIONS:
                return DescribeTransactionsResponse.parse(responseBuffer, version);
            case LIST_TRANSACTIONS:
                return ListTransactionsResponse.parse(responseBuffer, version);
            case ALLOCATE_PRODUCER_IDS:
                return AllocateProducerIdsResponse.parse(responseBuffer, version);
            case CONSUMER_GROUP_HEARTBEAT:
                return ConsumerGroupHeartbeatResponse.parse(responseBuffer, version);
            case CONSUMER_GROUP_DESCRIBE:
                return ConsumerGroupDescribeResponse.parse(responseBuffer, version);
            case CONTROLLER_REGISTRATION:
                return ControllerRegistrationResponse.parse(responseBuffer, version);
            case GET_TELEMETRY_SUBSCRIPTIONS:
                return GetTelemetrySubscriptionsResponse.parse(responseBuffer, version);
            case PUSH_TELEMETRY:
                return PushTelemetryResponse.parse(responseBuffer, version);
            case ASSIGN_REPLICAS_TO_DIRS:
                return AssignReplicasToDirsResponse.parse(responseBuffer, version);
            case LIST_CLIENT_METRICS_RESOURCES:
                return ListClientMetricsResourcesResponse.parse(responseBuffer, version);
            case DESCRIBE_TOPIC_PARTITIONS:
                return DescribeTopicPartitionsResponse.parse(responseBuffer, version);

            // AutoMQ for Kafka inject start
            case CREATE_STREAMS:
                return CreateStreamsResponse.parse(responseBuffer, version);
            case OPEN_STREAMS:
                return OpenStreamsResponse.parse(responseBuffer, version);
            case DELETE_STREAMS:
                return DeleteStreamsResponse.parse(responseBuffer, version);
            case CLOSE_STREAMS:
                return CloseStreamsResponse.parse(responseBuffer, version);
            case TRIM_STREAMS:
                return TrimStreamsResponse.parse(responseBuffer, version);
            case PREPARE_S3_OBJECT:
                return PrepareS3ObjectResponse.parse(responseBuffer, version);
            case COMMIT_STREAM_OBJECT:
                return CommitStreamObjectResponse.parse(responseBuffer, version);
            case COMMIT_STREAM_SET_OBJECT:
                return CommitStreamSetObjectResponse.parse(responseBuffer, version);
            case GET_OPENING_STREAMS:
                return GetOpeningStreamsResponse.parse(responseBuffer, version);
            case GET_KVS:
                return GetKVsResponse.parse(responseBuffer, version);
            case PUT_KVS:
                return PutKVsResponse.parse(responseBuffer, version);
            case DELETE_KVS:
                return DeleteKVsResponse.parse(responseBuffer, version);
            case AUTOMQ_REGISTER_NODE:
                return AutomqRegisterNodeResponse.parse(responseBuffer, version);
            case AUTOMQ_GET_NODES:
                return AutomqGetNodesResponse.parse(responseBuffer, version);
            case AUTOMQ_ZONE_ROUTER:
                return AutomqZoneRouterResponse.parse(responseBuffer, version);
            case AUTOMQ_GET_PARTITION_SNAPSHOT:
                return AutomqGetPartitionSnapshotResponse.parse(responseBuffer, version);
            case GET_NEXT_NODE_ID:
                return GetNextNodeIdResponse.parse(responseBuffer, version);
            case DESCRIBE_STREAMS:
                return DescribeStreamsResponse.parse(responseBuffer, version);
            case AUTOMQ_UPDATE_GROUP:
                return AutomqUpdateGroupResponse.parse(responseBuffer, version);

            // AutoMQ for Kafka inject end
            case SHARE_GROUP_HEARTBEAT:
                return ShareGroupHeartbeatResponse.parse(responseBuffer, version);
            case SHARE_GROUP_DESCRIBE:
                return ShareGroupDescribeResponse.parse(responseBuffer, version);
            case SHARE_FETCH:
                return ShareFetchResponse.parse(responseBuffer, version);
            case SHARE_ACKNOWLEDGE:
                return ShareAcknowledgeResponse.parse(responseBuffer, version);
            case ADD_RAFT_VOTER:
                return AddRaftVoterResponse.parse(responseBuffer, version);
            case REMOVE_RAFT_VOTER:
                return RemoveRaftVoterResponse.parse(responseBuffer, version);
            case UPDATE_RAFT_VOTER:
                return UpdateRaftVoterResponse.parse(responseBuffer, version);
            case INITIALIZE_SHARE_GROUP_STATE:
                return InitializeShareGroupStateResponse.parse(responseBuffer, version);
            case READ_SHARE_GROUP_STATE:
                return ReadShareGroupStateResponse.parse(responseBuffer, version);
            case WRITE_SHARE_GROUP_STATE:
                return WriteShareGroupStateResponse.parse(responseBuffer, version);
            case DELETE_SHARE_GROUP_STATE:
                return DeleteShareGroupStateResponse.parse(responseBuffer, version);
            case READ_SHARE_GROUP_STATE_SUMMARY:
                return ReadShareGroupStateSummaryResponse.parse(responseBuffer, version);
            default:
                throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseResponse`, the " +
                        "code should be updated to do so.", apiKey));
        }
    }
Parse a response from the provided buffer. The buffer is expected to hold both
the {@link ResponseHeader} as well as the response payload.
clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java