dajac commented on a change in pull request #9318: URL: https://github.com/apache/kafka/pull/9318#discussion_r500544582
########## File path: core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala ########## @@ -1321,25 +1106,15 @@ object GroupMetadataManager { */ def readMessageKey(buffer: ByteBuffer): BaseKey = { val version = buffer.getShort - val keySchema = schemaForKey(version) - val key = keySchema.read(buffer) - - if (version <= CURRENT_OFFSET_KEY_SCHEMA_VERSION) { + if (version >= OffsetCommitKey.LOWEST_SUPPORTED_VERSION && version <= OffsetCommitKey.HIGHEST_SUPPORTED_VERSION) { // version 0 and 1 refer to offset - val group = key.get(OFFSET_KEY_GROUP_FIELD).asInstanceOf[String] - val topic = key.get(OFFSET_KEY_TOPIC_FIELD).asInstanceOf[String] - val partition = key.get(OFFSET_KEY_PARTITION_FIELD).asInstanceOf[Int] - - OffsetKey(version, GroupTopicPartition(group, new TopicPartition(topic, partition))) - - } else if (version == CURRENT_GROUP_KEY_SCHEMA_VERSION) { + val key = new OffsetCommitKey(new ByteBufferAccessor(buffer), version) + OffsetKey(version, GroupTopicPartition(key.group, new TopicPartition(key.topic, key.partition))) + } else if (version >= GroupMetadataKeyData.LOWEST_SUPPORTED_VERSION && version <= GroupMetadataKeyData.HIGHEST_SUPPORTED_VERSION) { // version 2 refers to offset Review comment: nit: Not related to your changes but could fix this comment? `refers to offset` => `refers to group metadata`? ########## File path: core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala ########## @@ -1257,60 +1070,32 @@ object GroupMetadataManager { assignment: Map[String, Array[Byte]], apiVersion: ApiVersion): Array[Byte] = { - val (version, value) = { - if (apiVersion < KAFKA_0_10_1_IV0) - (0.toShort, new Struct(GROUP_METADATA_VALUE_SCHEMA_V0)) - else if (apiVersion < KAFKA_2_1_IV0) - (1.toShort, new Struct(GROUP_METADATA_VALUE_SCHEMA_V1)) - else if (apiVersion < KAFKA_2_3_IV0) - (2.toShort, new Struct(GROUP_METADATA_VALUE_SCHEMA_V2)) - else - (3.toShort, new Struct(GROUP_METADATA_VALUE_SCHEMA_V3)) - } - - value.set(PROTOCOL_TYPE_KEY, groupMetadata.protocolType.getOrElse("")) - value.set(GENERATION_KEY, groupMetadata.generationId) - value.set(PROTOCOL_KEY, groupMetadata.protocolName.orNull) - value.set(LEADER_KEY, groupMetadata.leaderOrNull) - - if (version >= 2) - value.set(CURRENT_STATE_TIMESTAMP_KEY, groupMetadata.currentStateTimestampOrDefault) - - val memberArray = groupMetadata.allMemberMetadata.map { memberMetadata => - val memberStruct = value.instance(MEMBERS_KEY) - memberStruct.set(MEMBER_ID_KEY, memberMetadata.memberId) - memberStruct.set(CLIENT_ID_KEY, memberMetadata.clientId) - memberStruct.set(CLIENT_HOST_KEY, memberMetadata.clientHost) - memberStruct.set(SESSION_TIMEOUT_KEY, memberMetadata.sessionTimeoutMs) - - if (version > 0) - memberStruct.set(REBALANCE_TIMEOUT_KEY, memberMetadata.rebalanceTimeoutMs) - - if (version >= 3) - memberStruct.set(GROUP_INSTANCE_ID_KEY, memberMetadata.groupInstanceId.orNull) - - // The group is non-empty, so the current protocol must be defined - val protocol = groupMetadata.protocolName.orNull - if (protocol == null) - throw new IllegalStateException("Attempted to write non-empty group metadata with no defined protocol") - - val metadata = memberMetadata.metadata(protocol) - memberStruct.set(SUBSCRIPTION_KEY, ByteBuffer.wrap(metadata)) - - val memberAssignment = assignment(memberMetadata.memberId) - assert(memberAssignment != null) - - memberStruct.set(ASSIGNMENT_KEY, ByteBuffer.wrap(memberAssignment)) - - memberStruct - } - - value.set(MEMBERS_KEY, memberArray.toArray) - - val byteBuffer = ByteBuffer.allocate(2 /* version */ + value.sizeOf) - byteBuffer.putShort(version) - value.writeTo(byteBuffer) - byteBuffer.array() + val version = + if (apiVersion < KAFKA_0_10_1_IV0) 0.toShort + else if (apiVersion < KAFKA_2_1_IV0) 1.toShort + else if (apiVersion < KAFKA_2_3_IV0) 2.toShort + else 3.toShort + + serializeMessage(version, new GroupMetadataValue() + .setProtocolType(groupMetadata.protocolType.getOrElse("")) + .setGeneration(groupMetadata.generationId) + .setProtocol(groupMetadata.protocolName.orNull) + .setLeader(groupMetadata.leaderOrNull) + .setCurrentStateTimestamp(groupMetadata.currentStateTimestampOrDefault) + .setMembers(groupMetadata.allMemberMetadata.map { memberMetadata => + new GroupMetadataValue.MemberMetadata() + .setMemberId(memberMetadata.memberId) + .setClientId(memberMetadata.clientId) + .setClientHost(memberMetadata.clientHost) + .setSessionTimeout(memberMetadata.sessionTimeoutMs) + .setRebalanceTimeout(memberMetadata.rebalanceTimeoutMs) + .setGroupInstanceId(memberMetadata.groupInstanceId.orNull) + // The group is non-empty, so the current protocol must be defined + .setSubscription(groupMetadata.protocolName.map(memberMetadata.metadata) + .getOrElse(throw new IllegalStateException(s"Attempted to write member ${memberMetadata.memberId} of group ${groupMetadata.groupId} with no defined protocol"))) Review comment: The message sounds a bit weird to me cause it is the group which does not have a defined protocol, not the member. I would just keep the previous message here and also add a `.` as you put one in the other message. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org