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
