val key = GroupMetadataManager.groupMetadataKey(group.groupId)
val value = GroupMetadataManager.groupMetadataValue(group, groupAssignment, interBrokerProtocolVersion)

case class GroupMetadataKey(version: Short, key: String) extends BaseKey {
  // 保存的是消费者组名称
  override def toString: String = key
}

private[group] def groupMetadataKey(group: String): Array[Byte] = {
    
    val key = new Struct(CURRENT_GROUP_KEY_SCHEMA)
    key.set(GROUP_KEY_GROUP_FIELD, group)
		// 构造ByteBuffer对象，容纳version和key
    val byteBuffer = ByteBuffer.allocate(2 /* version */ + key.sizeOf)
    byteBuffer.putShort(CURRENT_GROUP_KEY_SCHEMA_VERSION)
  	// 写入byteBuffer
    key.writeTo(byteBuffer)
    byteBuffer.array()
  }

/**
 * GroupMetadata 消费者组元数据对象
 * assignment 分区消费分配方案
 * apiVersion api版本号
 */
private[group] def groupMetadataValue(groupMetadata: GroupMetadata,
                                        assignment: Map[String, Array[Byte]],
                                        apiVersion: ApiVersion): Array[Byte] = {
  	...
  	// 依次写入消费者组主要的元数据信息
    value.set(PROTOCOL_TYPE_KEY, groupMetadata.protocolType.getOrElse(""))
    value.set(GENERATION_KEY, groupMetadata.generationId)
    value.set(PROTOCOL_KEY, groupMetadata.protocolOrNull)
    value.set(LEADER_KEY, groupMetadata.leaderOrNull)
  
    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)
			...
      // 写入消费订阅信息和费分配方案信息
      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)
  	// 写入byteBuffer 并返回
    val byteBuffer = ByteBuffer.allocate(2 /* version */ + value.sizeOf)
    byteBuffer.putShort(version)
    value.writeTo(byteBuffer)
    byteBuffer.array()
  }

// build records
val records = {
          val buffer = ByteBuffer.allocate(AbstractRecords.estimateSizeInBytes(magicValue, compressionType,
            Seq(new SimpleRecord(timestamp, key, value)).asJava))
          val builder = MemoryRecords.builder(buffer, magicValue, compressionType, timestampType, 0L)
          builder.append(timestamp, key, value)
          builder.build()

// 参数是version和GroupTopicPartition
case class OffsetKey(version: Short, key: GroupTopicPartition) extends BaseKey {

  override def toString: String = key.toString
}

// GroupTopicPartition 类型是 <消费者组名，主题，分区> 的三元组。
OffsetKey(version, GroupTopicPartition(group, new TopicPartition(topic, partition)))

val offsetAndMetadata = GroupMetadataManager.readOffsetMessageValue(record.value)
...
private[group] def offsetCommitValue(offsetAndMetadata: OffsetAndMetadata,
                                       apiVersion: ApiVersion): Array[Byte] = {
  	// 根据不同的消息格式版本，创建对应的对象
    val (version, value) = {
      
      if (apiVersion < KAFKA_2_1_IV0 || offsetAndMetadata.expireTimestamp.nonEmpty) {
				...
      } else if (apiVersion < KAFKA_2_1_IV1) {
        ...
      } else {
        // 依次写入位移值、Leader Epoch值、自定义元数据以及时间戳
        val value = new Struct(OFFSET_COMMIT_VALUE_SCHEMA_V3)
        value.set(OFFSET_VALUE_OFFSET_FIELD_V3, offsetAndMetadata.offset)
        value.set(OFFSET_VALUE_LEADER_EPOCH_FIELD_V3,
          offsetAndMetadata.leaderEpoch.orElse(RecordBatch.NO_PARTITION_LEADER_EPOCH))
        value.set(OFFSET_VALUE_METADATA_FIELD_V3, offsetAndMetadata.metadata)
        value.set(OFFSET_VALUE_COMMIT_TIMESTAMP_FIELD_V3, offsetAndMetadata.commitTimestamp)
        (3, value)
      }
    }
		// 构建ByteBuffer，写入消息格式版本和结构体并返回
    val byteBuffer = ByteBuffer.allocate(2 /* version */ + value.sizeOf)
    byteBuffer.putShort(version.toShort)
    value.writeTo(byteBuffer)
    byteBuffer.array()
  }

val records = filteredOffsetMetadata.map { case (topicPartition, offsetAndMetadata) =>
            val key = GroupMetadataManager.offsetCommitKey(group.groupId, topicPartition)
            val value = GroupMetadataManager.offsetCommitValue(offsetAndMetadata, interBrokerProtocolVersion)
            new SimpleRecord(timestamp, key, value)
          }

// build MemoryRecords
val builder = MemoryRecords.builder(buffer, magicValue, compressionType, timestampType, 0L, time.milliseconds(),
            producerId, producerEpoch, 0, isTxnOffsetCommit, RecordBatch.NO_PARTITION_LEADER_EPOCH)
// 可能会有多个record，遍历append build
records.foreach(builder.append)

// entries终将会传入appendForGroup方法
val entries = Map(offsetTopicPartition -> builder.build())

/**
 * log Kafka消息对象
 * lazyOffsetIndex 位移索引文件
 * lazyTimeIndex 时间戳索引文件
 * txnIndex 已中止事务索引文件
 * baseOffset 每个日志段对象的起始位移
 * indexIntervalBytes 日志对象新增索引项的频率
 */
class LogSegment private[log] (val log: FileRecords,
                               val lazyOffsetIndex: LazyOffsetIndex,
                               val lazyTimeIndex: LazyTimeIndex,
                               val txnIndex: TransactionIndex,
                               val baseOffset: Long,
                               val indexIntervalBytes: Int,
                               val rollJitterMs: Long,
                               val time: Time) extends Logging

...
/**
 * largestOffset  待写入消息批次中消息的最大位移值
 * largestTimestamp 最大时间戳
 * shallowOffsetOfMaxTimestamp 最大时间戳对应消息的位移
 * records 待写入的消息体
 */
def append(largestOffset: Long,
             largestTimestamp: Long,
             shallowOffsetOfMaxTimestamp: Long,
             records: MemoryRecords): Unit = {...}

/**
 * startOffset 要读取的第一条消息的位移
 * maxSize 能读取的最大字节数
 * maxPosition 能读到的最大文件位置
 * minOneMessage 是否允许在消息体过大时至少返回第一条消息
 */
def read(startOffset: 
         Long, maxOffset: Option[Long], 
         maxSize: Int, 
         maxPosition: Long = size,
         minOneMessage: Boolean = false): FetchDataInfo = {...}

/**
 * leaderEpochCache 在恢复过程中缓存的leaderEpoch值
 */
def recover(producerStateManager: ProducerStateManager, 
            leaderEpochCache: Option[LeaderEpochFileCache] = None): Int = {...}

v0
Message => Crc MagicByte Attributes Key Value
  Crc => int32
  MagicByte => int8
  Attributes => int8
  Key => bytes
  Value => bytes
  
v1 (supported since 0.10.0)
Message => Crc MagicByte Attributes Key Value
	// CRC用于检查代理和消息的完整
  Crc => int32
	// 版本id，用于允许消息二进制格式的向后兼容演变。当前值为1。
  MagicByte => int8
	// 保存有关消息的元数据属性
	// 最低3位包含用于消息的压缩解码器
	// 最低的第4位用于表示时间戳类型，0代表createtime,1代表logappendtime。
  Attributes => int8
	// 消息时间戳
  Timestamp => int64
	// 分区分配的可选消息密钥
  Key => bytes
	// 实际消息内容，可能是包含的一个消息集。
  Value => bytes

object Message {

  // crc 消息的校验码，防止消息错误（4B）
  val CrcOffset = 0
  val CrcLength = 4
  
  // magic 消息格式的版本号（1B）
  val MagicOffset = CrcOffset + CrcLength
  val MagicLength = 1
  
  // attributes 消息的属性，比如压缩类型，时间戳类型，创建时间/追加时间 (1B)
  val AttributesOffset = MagicOffset + MagicLength
  val AttributesLength = 1
  // timestamp 时间戳信息(8B)
  val TimestampOffset = AttributesOffset + AttributesLength
  val TimestampLength = 8
  // key （4B）
  val KeySizeOffset_V0 = AttributesOffset + AttributesLength
  val KeySizeOffset_V1 = TimestampOffset + TimestampLength
  val KeySizeLength = 4 
  val KeyOffset_V0 = KeySizeOffset_V0 + KeySizeLength
  val KeyOffset_V1 = KeySizeOffset_V1 + KeySizeLength
  // value （4B）
  val ValueSizeLength = 4
}

// 表示当前RecordBatch的起始位移 (8B)
baseOffset: int64 
// 计算partition leader epoch到headers之间的长度 (4B)
batchLength: int32
// 用来确保数据可靠性 4B
partitionLeaderEpoch: int32
// magic等于2 (1B)
magic: int8 (current magic value is 2)
// 4B
crc: int32
// 2B
attributes: int16
	bit 0~2:
		0: no compression
		1: gzip
		2: snappy
		3: lz4
	bit 3: timestampType
	bit 4: isTransactional (0 means not transactional)
	bit 5: isControlBatch (0 means not a control batch)
	bit 6~15: unused
// 4B
lastOffsetDelta: int32
// 8B
firstTimestamp: int64
maxTimestamp: int64
producerId: int64
producerEpoch: int16
baseSequence: int32
// batch records
records: [Record]

// 消息总长度 3B
length: varint
// 弃用
attributes: int8
	bit 0~7: unused
// 时间戳增量 3B
timestampDelta: varint
// 位移增量 3B
offsetDelta: varint
// key
keyLength: varint
key: byte[]
// value
valueLen: varint
value: byte[]
Headers => [Header]

headerKeyLength: varint
headerKey: String
headerValueLength: varint
Value: byte[]