kafka-coordinator-group-GroupMetadataManager.adoc

GroupMetadataManager

GroupMetadataManager is created exclusively when GroupCoordinator is created.

Table 1. GroupMetadataManager’s Internal Properties (e.g. Registries and Counters)

Name	Description
compressionType	CompressionType (default: NoCompressionCodec)
groupMetadataCache	GroupMetadata by group ID
groupMetadataTopicPartitionCount	Number of partitions for the __consumer_offsets consumer group metadata topic (default: 50)
scheduler	KafkaScheduler with 1 daemon thread with group-metadata-manager- prefix Started when GroupMetadataManager is requested to start up. Shut down when GroupMetadataManager is requested to shut down. Used when: startup (with enableMetadataExpiration flag on) that schedules the delete-expired-group-metadata task that cleanupGroupMetadata every 600000L millis scheduleLoadGroupAndOffsets that loads offsets and group metadata from __consumer_offsets topic removeGroupsForPartition that unloads offsets and group metadata from __consumer_offsets topic scheduleHandleTxnCompletion that schedules the handleTxnCompletion-[producerId] task that handleTxnCompletion

enableMetadataExpiration Method

enableMetadataExpiration(): Unit

enableMetadataExpiration requests KafkaScheduler to start.

enableMetadataExpiration schedules delete-expired-group-metadata task that cleanupGroupMetadata every offsetsRetentionCheckIntervalMs milliseconds.

Note	enableMetadataExpiration is used exclusively when GroupCoordinator is started.

cleanupGroupMetadata Internal Method

cleanupGroupMetadata(): Unit  // (1)
cleanupGroupMetadata(deletedTopicPartitions: Option[Seq[TopicPartition]]): Unit

1. Calls the other cleanupGroupMetadata with empty deletedTopicPartitions collection

cleanupGroupMetadata takes the current time (using time) and for every GroupMetadata in the cache does the following:

1. FIXME

In the end, cleanupGroupMetadata prints out the following INFO message to the logs:

Removed [offsetsRemoved] expired offsets in [duration] milliseconds

Note	cleanupGroupMetadata is used exclusively when GroupMetadataManager is requested to enableMetadataExpiration (as delete-expired-group-metadata task).

Getting Number of Partitions for __consumer_offsets Consumer Group Metadata Topic — getGroupMetadataTopicPartitionCount Internal Method

getGroupMetadataTopicPartitionCount: Int

getGroupMetadataTopicPartitionCount requests the KafkaZkClient for getTopicPartitionCount of __consumer_offsets consumer group metadata topic.

If not available, getGroupMetadataTopicPartitionCount uses the OffsetConfig for offsetsTopicNumPartitions (default: 50).

Note	getGroupMetadataTopicPartitionCount is used exclusively when GroupMetadataManager is requested for groupMetadataTopicPartitionCount.

shutdown Method

shutdown(): Unit

shutdown…​FIXME

Note	shutdown is used when…​FIXME

startup Method

startup(enableMetadataExpiration: Boolean): Unit

startup…​FIXME

Note	startup is used when…​FIXME

scheduleLoadGroupAndOffsets Method

scheduleLoadGroupAndOffsets(offsetsPartition: Int, onGroupLoaded: GroupMetadata => Unit): Unit

scheduleLoadGroupAndOffsets…​FIXME

Note	scheduleLoadGroupAndOffsets is used when…​FIXME

removeGroupsForPartition Method

removeGroupsForPartition(
  offsetsPartition: Int,
  onGroupUnloaded: GroupMetadata => Unit): Unit

removeGroupsForPartition…​FIXME

Note	removeGroupsForPartition is used when…​FIXME

scheduleHandleTxnCompletion Method

scheduleHandleTxnCompletion(
  producerId: Long,
  completedPartitions: Set[Int],
  isCommit: Boolean): Unit

scheduleHandleTxnCompletion…​FIXME

Note	scheduleHandleTxnCompletion is used exclusively when GroupCoordinator is requested to scheduleHandleTxnCompletion.

loadGroupsAndOffsets Method

loadGroupsAndOffsets(
  topicPartition: TopicPartition,
  onGroupLoaded: GroupMetadata => Unit): Unit

loadGroupsAndOffsets…​FIXME

Note	loadGroupsAndOffsets is used when…​FIXME

Requesting ReplicaManager to Append Records — appendForGroup Internal Method

appendForGroup(
  group: GroupMetadata,
  records: Map[TopicPartition, MemoryRecords],
  callback: Map[TopicPartition, PartitionResponse] => Unit): Unit

appendForGroup simply requests the ReplicaManager to append records.

Note	appendForGroup is used exclusively when GroupMetadataManager is requested to storeGroup and storeOffsets.

Getting Metadata of Group by Group ID — getGroup Method

getGroup(groupId: String): Option[GroupMetadata]

getGroup finds the GroupMetadata for the group ID in the groupMetadataCache internal registry.

getGroup returns None if the metadata could not be found.

Note

getGroup is used when: GroupCoordinator is requested to handleJoinGroup, handleSyncGroup, handleLeaveGroup, handleDeleteGroups, handleHeartbeat, handleTxnCommitOffsets, handleCommitOffsets, and handleDescribeGroup GroupMetadataManager is requested to groupNotExists (when GroupCoordinator is requested to handleDeleteGroups) and handleTxnCompletion (when GroupCoordinator is requested to scheduleHandleTxnCompletion)