nprasad2077/kafka
1
0 Fork 0
mirror of https://github.com/apache/kafka.git synced 2026-11-04 04:44:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

MINOR: Fix Javadoc and typos in group-coordinator (#24083)

Browse Source 
Many Javadoc comments in the group coordinator module are out of date or
wrong. They name the wrong return type or record, describe share group
code with consumer group wording, miss `@param` tags, wrap `@link` in
double braces or contain typos. The exception messages in the consumer
and streams `CurrentAssignmentBuilder` also say "a unknown state".

This patch fixes them. It only changes Javadoc, comments and these two
exception messages.

Reviewers: Sean Quah <squah@confluent.io>, Chia-Ping Tsai
<chia7712@gmail.com>

Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
David JacotandClaude Sonnet 5.5 authored and GitHub committed 2026-11-03 19:18:31 +08:00
1 parent 1fdea4a760
commit 72955baae9
43 files changed
+300 -234
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/ConsumerGroupMigrationPolicy.java
+5 -3
View File
@@ -63,10 +63,12 @@ public enum ConsumerGroupMigrationPolicy {
.collect(Collectors.toMap(policy -> policy.name.toLowerCase(Locale.ROOT), Function.identity()));
/**
* Parse a string into the corresponding {@code GroupProtocolMigrationPolicy} enum value, in a case-insensitive manner.
* Parse a string into the corresponding {@code ConsumerGroupMigrationPolicy} enum value, in a
* case-insensitive manner.
*
* @return The {{@link ConsumerGroupMigrationPolicy}} according to the string passed. None is returned if
* the string doesn't correspond to a valid policy.
* @param name The name of the policy.
* @return The {@link ConsumerGroupMigrationPolicy} according to the string passed.
* {@code DISABLED} is returned if the string doesn't correspond to a valid policy.
*/
public static ConsumerGroupMigrationPolicy parse(String name) {
if (name == null) {
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
+1 -1
View File
@@ -1314,7 +1314,7 @@ public class GroupCoordinatorConfig {
}
/**
* The consumer group heartbeat interval in milliseconds.
* The share group heartbeat interval in milliseconds.
*/
public int shareGroupHeartbeatIntervalMs() {
return shareGroupHeartbeatIntervalMs;
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java
+22 -21
View File
@@ -236,7 +236,7 @@ public class GroupCoordinatorRecordHelpers {
* Creates a ConsumerGroupTargetAssignmentMetadata record.
*
* @param groupId The consumer group id.
* @param assignmentEpoch The consumer group epoch.
* @param assignmentEpoch The target assignment epoch.
* @param assignmentTimestamp The time at which the target assignment calculation finished.
* @return The record.
*/
@@ -526,8 +526,8 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ShareGroupMemberMetadata record.
*
* @param groupId The consumer group id.
* @param member The consumer group member.
* @param groupId The share group id.
* @param member The share group member.
* @return The record.
*/
public static CoordinatorRecord newShareGroupMemberSubscriptionRecord(
@@ -572,9 +572,9 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ShareGroupMetadata record.
*
* @param groupId The group id.
* @param newGroupEpoch The group epoch.
* @param metadataHash The group metadata hash.
* @param groupId The share group id.
* @param newGroupEpoch The share group epoch.
* @param metadataHash The share group metadata hash.
* @return The record.
*/
public static CoordinatorRecord newShareGroupEpochRecord(
@@ -597,7 +597,7 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ShareGroupMetadata tombstone.
*
* @param groupId The group id.
* @param groupId The share group id.
* @return The record.
*/
public static CoordinatorRecord newShareGroupEpochTombstoneRecord(
@@ -612,8 +612,8 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ShareGroupTargetAssignmentMember record.
*
* @param groupId The group id.
* @param memberId The group member id.
* @param groupId The share group id.
* @param memberId The share group member id.
* @param partitions The target partitions of the member.
* @return The record.
*/
@@ -648,8 +648,8 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ShareGroupTargetAssignmentMember tombstone.
*
* @param groupId The group id.
* @param memberId The group member id.
* @param groupId The share group id.
* @param memberId The share group member id.
* @return The record.
*/
public static CoordinatorRecord newShareGroupTargetAssignmentTombstoneRecord(
@@ -666,8 +666,8 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ShareGroupTargetAssignmentMetadata record.
*
* @param groupId The group id.
* @param assignmentEpoch The group epoch.
* @param groupId The share group id.
* @param assignmentEpoch The target assignment epoch.
* @param assignmentTimestamp The time at which the target assignment calculation finished.
* @return The record.
*/
@@ -691,7 +691,7 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ShareGroupTargetAssignmentMetadata tombstone.
*
* @param groupId The group id.
* @param groupId The share group id.
* @return The record.
*/
public static CoordinatorRecord newShareGroupTargetAssignmentMetadataTombstoneRecord(
@@ -706,8 +706,8 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ShareGroupCurrentMemberAssignment record.
*
* @param groupId The group id.
* @param member The group member.
* @param groupId The share group id.
* @param member The share group member.
* @return The record.
*/
public static CoordinatorRecord newShareGroupCurrentAssignmentRecord(
@@ -730,10 +730,10 @@ public class GroupCoordinatorRecordHelpers {
}
/**
* Creates a ConsumerGroupCurrentMemberAssignment tombstone.
* Creates a ShareGroupCurrentMemberAssignment tombstone.
*
* @param groupId The consumer group id.
* @param memberId The consumer group member id.
* @param groupId The share group id.
* @param memberId The share group member id.
* @return The record.
*/
public static CoordinatorRecord newShareGroupCurrentAssignmentTombstoneRecord(
@@ -765,9 +765,10 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ShareGroupStatePartitionMetadata record.
*
* @param groupId The share group id.
* @param groupId The share group id.
* @param initializingTopics Topics which are being initialized.
* @param initializedTopics Topics which have been initialized.
* @param deletingTopics Topics which are being deleted.
* @param deletingTopics Topics which are being deleted.
* @return The record.
*/
public static CoordinatorRecord newShareGroupStatePartitionMetadataRecord(
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
+19 -14
View File
@@ -387,14 +387,19 @@ public class GroupCoordinatorService implements GroupCoordinator {
private volatile CoordinatorMetadataImage metadataImage = null;
/**
* Constructor.
*
* @param logContext The log context.
* @param config The group coordinator config.
* @param runtime The runtime.
* @param groupCoordinatorMetrics The group coordinator metrics.
* @param groupConfigManager The group config manager.
* @param persister The persister.
* @param timer The timer.
* @param logContext The log context.
* @param config The group coordinator config.
* @param runtime The runtime.
* @param groupCoordinatorMetrics The group coordinator metrics.
* @param groupConfigManager The group config manager.
* @param persister The persister.
* @param timer The timer.
* @param partitionMetadataClient The partition metadata client.
* @param streamsGroupTopologyDescriptionPlugin The streams group topology description plugin,
* if one is configured.
* @param time The time.
*/
GroupCoordinatorService(
LogContext logContext,
@@ -431,7 +436,7 @@ public class GroupCoordinatorService implements GroupCoordinator {
}
/**
* Throws CoordinatorNotAvailableException if the not active.
* Throws CoordinatorNotAvailableException if the coordinator is not active.
*/
private void throwIfNotActive() {
if (!isActive.get()) {
@@ -567,7 +572,6 @@ public class GroupCoordinatorService implements GroupCoordinator {
*
* @param request The request to validate.
* @throws InvalidRequestException if the request is not valid.
* @throws UnsupportedAssignorException if the assignor is not supported.
*/
private static void throwIfStreamsGroupHeartbeatRequestIsInvalid(
StreamsGroupHeartbeatRequestData request
@@ -904,7 +908,8 @@ public class GroupCoordinatorService implements GroupCoordinator {
return topicData;
})
.iterator()));
// don't uninitialized share group state here, as we regard this alter share group offsets request failed.
// don't uninitialize share group state here, as we regard this alter share group offsets
// request failed.
return data;
}
@@ -2195,10 +2200,10 @@ public class GroupCoordinatorService implements GroupCoordinator {
}
private boolean shouldComputeSharePartitionLag(PartitionStateSummaryData partitionData) {
// The share partition lag would be computed for a share partition ony if -
// The share partition lag would be computed for a share partition only if -
// 1. The read summary result for the partition is successful.
// 3. The start offset is initialized.
// 4. The delivery complete count is initialized.
// 2. The start offset is initialized.
// 3. The delivery complete count is initialized.
return partitionData.errorCode() == Errors.NONE.code() &&
partitionData.startOffset() != PartitionFactory.UNINITIALIZED_START_OFFSET &&
partitionData.deliveryCompleteCount() != PartitionFactory.UNINITIALIZED_DELIVERY_COMPLETE_COUNT;
@@ -2682,7 +2687,7 @@ public class GroupCoordinatorService implements GroupCoordinator {
* @param context The request context.
* @param request The OffsetFetchRequestGroup request.
* @param exception The exception to handle.
* @return The OffsetFetchRequestGroup response.
* @return The OffsetFetchResponseGroup response.
*/
private OffsetFetchResponseData.OffsetFetchResponseGroup handleOffsetFetchException(
String operationName,
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
+28 -19
View File
@@ -146,8 +146,8 @@ import java.util.concurrent.TimeUnit;
/**
* The group coordinator shard is a replicated state machine that manages the metadata of all
* classic and consumer groups. It holds the hard and the soft state of the groups. This class
* has two kinds of methods:
* classic, consumer, share and streams groups. It holds the hard and the soft state of the
* groups. This class has two kinds of methods:
* 1) The request handlers which handle the requests and generate a response and records to
* mutate the hard state. Those records will be written by the runtime and applied to the
* hard state via the replay methods.
@@ -382,7 +382,7 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
static final String GROUP_EXPIRATION_KEY = "expire-group-metadata";
/**
* The classic, consumer and streams group size counter key to schedule a timer task.
* The group size counter key to schedule a timer task.
*
* Visible for testing.
*/
@@ -439,6 +439,9 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
* @param logContext The log context.
* @param groupMetadataManager The group metadata manager.
* @param offsetMetadataManager The offset metadata manager.
* @param time The time.
* @param timer The coordinator timer.
* @param config The group coordinator config.
* @param coordinatorMetrics The coordinator metrics.
* @param metricsShard The coordinator metrics shard.
*/
@@ -583,11 +586,12 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
/**
* Handles a SyncGroup request.
*
* @param context The request context.
* @param request The actual SyncGroup request.
* @param context The request context.
* @param request The actual SyncGroup request.
* @param responseFuture The sync group response future.
*
* @return A Result containing the SyncGroup response and
* a list of records to update the state machine.
* @return A Result containing a list of records to update the state machine. The response is
* completed through the response future.
*/
public CoordinatorResult<Void, CoordinatorRecord> classicGroupSync(
AuthorizableRequestContext context,
@@ -771,7 +775,11 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
*
* @param groupId - The group ID
* @param topics - The set of topics which were deleted successfully by the persister
* @return the final response {@link DeleteShareGroupOffsetsResponseData} for the DeleteShareGroupOffsetsRequest
* @param errorTopicResponseList - The topic responses for the topics which were not deleted
* successfully
* @return A Result containing the final response {@link DeleteShareGroupOffsetsResponseData}
* for the DeleteShareGroupOffsetsRequest and a list of records to update the state
* machine.
*/
public CoordinatorResult<DeleteShareGroupOffsetsResponseData, CoordinatorRecord> completeDeleteShareGroupOffsets(
String groupId,
@@ -822,7 +830,7 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
* @param epoch The epoch (or offset) used to read from the
* timeline data structure.
*
* @return A List of OffsetFetchResponseTopics response.
* @return The OffsetFetchResponseGroup response.
*/
public OffsetFetchResponseData.OffsetFetchResponseGroup fetchOffsets(
OffsetFetchRequestData.OffsetFetchRequestGroup request,
@@ -852,7 +860,7 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
}
/**
* Handles an TxnOffsetCommit request.
* Handles a TxnOffsetCommit request.
*
* @param context The request context.
* @param request The actual TxnOffsetCommit request.
@@ -894,10 +902,10 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
/**
* Handles a ConsumerGroupDescribe request.
*
* @param groupIds The IDs of the groups to describe.
* @param groupIds The IDs of the groups to describe.
* @param committedOffset A specified committed offset corresponding to this shard.
*
* @return A list containing the ConsumerGroupDescribeResponseData.DescribedGroup.
*
*/
public List<ConsumerGroupDescribeResponseData.DescribedGroup> consumerGroupDescribe(
List<String> groupIds,
@@ -909,11 +917,11 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
/**
* Handles a StreamsGroupDescribe request.
*
* @param groupIds The IDs of the groups to describe.
* @param groupIds The IDs of the groups to describe.
* @param committedOffset A specified committed offset corresponding to this shard.
*
* @return A {@link StreamsGroupDescribeResult} containing the described groups and per-group
* stored description topology epoch (KIP-1331).
*
*/
public StreamsGroupDescribeResult streamsGroupDescribe(
List<String> groupIds,
@@ -1067,10 +1075,10 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
/**
* Handles a ShareGroupDescribe request.
*
* @param groupIds The IDs of the groups to describe.
* @param groupIds The IDs of the groups to describe.
* @param committedOffset A specified committed offset corresponding to this shard.
*
* @return A list containing the ShareGroupDescribeResponseData.DescribedGroup.
*
*/
public List<ShareGroupDescribeResponseData.DescribedGroup> shareGroupDescribe(
List<String> groupIds,
@@ -1113,7 +1121,7 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
}
/**
* Handles a OffsetDelete request.
* Handles an OffsetDelete request.
*
* @param context The request context.
* @param request The actual OffsetDelete request.
@@ -1205,7 +1213,8 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
}
/**
* Schedules (or reschedules) the group size counter for the classic/consumer groups.
* Schedules (or reschedules) the group size counter for the classic, consumer, share and
* streams groups.
*/
private void scheduleGroupSizeCounter() {
timer.schedule(
@@ -1222,7 +1231,7 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
}
/**
* Cancels the group size counter for the classic/consumer groups.
* Cancels the group size counter for the classic, consumer, share and streams groups.
*/
private void cancelGroupSizeCounter() {
timer.cancel(GROUP_SIZE_COUNTER_KEY);
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+76 -53
View File
@@ -268,8 +268,8 @@ import static org.apache.kafka.coordinator.group.streams.StreamsGroupMember.hasA
/**
* The GroupMetadataManager manages the metadata of all classic and consumer groups. It holds
* the hard and the soft state of the groups. This class has two kinds of methods:
* The GroupMetadataManager manages the metadata of all classic, consumer, share and streams groups.
* It holds the hard and the soft state of the groups. This class has two kinds of methods:
* 1) The request handlers which handle the requests and generate a response and records to
* mutate the hard state. Those records will be written by the runtime and applied to the
* hard state via the replay methods.
@@ -473,7 +473,7 @@ public class GroupMetadataManager {
private final CoordinatorTimer<CoordinatorRecord> timer;
/**
* The executor to executor asynchronous tasks.
* The executor to execute asynchronous tasks.
*/
private final CoordinatorExecutor<CoordinatorRecord> executor;
@@ -498,7 +498,7 @@ public class GroupMetadataManager {
private final ConsumerGroupPartitionAssignor defaultConsumerGroupAssignor;
/**
* The classic and consumer groups keyed by their name.
* The groups keyed by their name.
*/
private final TimelineHashMap<String, Group> groups;
@@ -1080,8 +1080,7 @@ public class GroupMetadataManager {
* created if it does not exist.
*
* @return A StreamsGroup.
* @throws GroupIdNotFoundException if the group does not exist and createIfNotExists is false or
* if the group is not a streams group.
* @throws GroupIdNotFoundException if the group does not exist and createIfNotExists is false.
* @throws IllegalStateException if the group does not have the expected type.
* Package private for testing.
*/
@@ -1767,7 +1766,7 @@ public class GroupMetadataManager {
}
/**
* Validates if the received instanceId has been released from the group
* Validates if the received instanceId has been fenced from the group
*
* @param member The consumer group member.
* @param groupId The consumer group id.
@@ -1835,7 +1834,7 @@ public class GroupMetadataManager {
}
/**
* Validates if the received instanceId has been released from the group
* Validates whether a static member exists for the given instanceId.
*
* @param staticMember The static member in the group.
* @param receivedInstanceId The instance id received in the request.
@@ -1866,7 +1865,8 @@ public class GroupMetadataManager {
* Checks whether the streams group can accept a new member or not based on the
* max group size defined.
*
* @param group The streams group.
* @param group The streams group.
* @param instanceId The instance id.
* @param memberId The member id.
*
* @throws GroupMaxSizeReachedException if the maximum capacity has been reached.
@@ -2104,10 +2104,13 @@ public class GroupMetadataManager {
* @param ownedActiveTasks The list of owned active tasks from the request or null.
* @param ownedStandbyTasks The list of owned standby tasks from the request or null.
* @param ownedWarmupTasks The list of owned warmup tasks from the request or null.
* @param taskOffsets The changelog offsets for tasks from the request or null.
* @param taskEndOffsets The changelog end offsets for tasks from the request or null.
* @param processId The process ID from the request or null.
* @param userEndpoint User-defined endpoint for Interactive Queries, or null.
* @param clientTags Used for rack-aware assignment algorithm, or null.
* @param shutdownApplication Whether all Streams clients in the group should shut down.
* @param memberEndpointEpoch The last endpoint information epoch seen be the group member.
* @param memberEndpointEpoch The last endpoint information epoch seen by the group member.
* @param requestApiVersion The api version of the StreamsGroupHeartbeat request.
* @return A result containing the StreamsGroupHeartbeat response and a list of records to update the state machine.
*/
@@ -2201,10 +2204,12 @@ public class GroupMetadataManager {
}
StreamsGroupMember updatedMember = updatedMemberBuilder.build();
// If the member is new or has changed, a StreamsGroupMemberMetadataValue record is written to the __consumer_offsets partition
// to persist the change, and bump the group epoch later. We track whether (and why) the group epoch is bumped:
// RECOMPUTE re-runs the assignor to produce a new final target assignment; REFINE only advances the assignment epoch to move
// the in-memory refinement (warm-up promotion) forward without recomputing the final target.
// If the member is new or has changed, a StreamsGroupMemberMetadataValue record is written
// to the __consumer_offsets partition to persist the change, and bump the group epoch
// later. We track whether (and why) the group epoch is bumped: RECOMPUTE re-runs the
// assignor to produce a new final target assignment; REFINED only advances the assignment
// epoch to move the in-memory refinement (warm-up promotion) forward without recomputing
// the final target.
AssignmentUpdate assignmentUpdate = hasStreamsMemberMetadataChanged(groupId, instanceId, member, updatedMember, records)
? AssignmentUpdate.RECOMPUTE
: AssignmentUpdate.NONE;
@@ -2510,6 +2515,9 @@ public class GroupMetadataManager {
* StreamsInvalidTopologyEpochException is thrown.
* - If the request defines a different topology with a higher epoch than the group, an InvalidRequestException
* is thrown, since topology updates are not supported yet.
* - If the request defines the same topology as the group, the group topology is returned.
* - If the request does not define a topology, the topology of the group is returned, or an
* IllegalStateException is thrown if the group does not have a topology yet.
*
* @param groupId The group ID.
* @param memberId The member ID.
@@ -2643,9 +2651,8 @@ public class GroupMetadataManager {
// 1. Create or update the member. If the member is new or has changed, a ConsumerGroupMemberMetadataValue
// record is written to the __consumer_offsets partition to persist the change. If the subscriptions have
// changed, the subscription metadata is updated and persisted by writing a ConsumerGroupPartitionMetadataValue
// record to the __consumer_offsets partition. Finally, the group epoch is bumped if the subscriptions have
// changed, and persisted by writing a ConsumerGroupMetadataValue record to the partition.
// changed, the group epoch is bumped and persisted by writing a ConsumerGroupMetadataValue
// record to the partition.
ConsumerGroupMember updatedMember = new ConsumerGroupMember.Builder(member)
.maybeUpdateInstanceId(Optional.ofNullable(instanceId))
.maybeUpdateRackId(Optional.ofNullable(rackId))
@@ -2837,9 +2844,8 @@ public class GroupMetadataManager {
// 1. Create or update the member. If the member is new or has changed, a ConsumerGroupMemberMetadataValue
// record is written to the __consumer_offsets partition to persist the change. If the subscriptions have
// changed, the subscription metadata is updated and persisted by writing a ConsumerGroupPartitionMetadataValue
// record to the __consumer_offsets partition. Finally, the group epoch is bumped if the subscriptions have
// changed, and persisted by writing a ConsumerGroupMetadataValue record to the partition.
// changed, the group epoch is bumped and persisted by writing a ConsumerGroupMetadataValue
// record to the partition.
ConsumerGroupMember updatedMember = new ConsumerGroupMember.Builder(member)
.maybeUpdateInstanceId(Optional.ofNullable(instanceId))
.maybeUpdateRackId(Utils.toOptional(subscription.rackId()))
@@ -3018,9 +3024,8 @@ public class GroupMetadataManager {
// 1. Create or update the member. If the member is new or has changed, a ShareGroupMemberMetadataValue
// record is written to the __consumer_offsets partition to persist the change. If the subscriptions have
// changed, the subscription metadata is updated and persisted by writing a ShareGroupPartitionMetadataValue
// record to the __consumer_offsets partition. Finally, the group epoch is bumped if the subscriptions have
// changed, and persisted by writing a ShareGroupMetadataValue record to the partition.
// changed, the group epoch is bumped and persisted by writing a ShareGroupMetadataValue
// record to the partition.
ShareGroupMember updatedMember = new ShareGroupMember.Builder(member)
.maybeUpdateRackId(Optional.ofNullable(rackId))
.maybeUpdateSubscribedTopicNames(Optional.ofNullable(subscribedTopicNames))
@@ -3231,6 +3236,7 @@ public class GroupMetadataManager {
* @param groupId The share group id for which partitions need to be initialized.
* @param groupEpoch The group epoch of the share group.
* @param subscriptionTopicNames The subscription topic names for the share group.
* @param records The list to accumulate any new records.
* @return An optional representing the persister initialize request.
*/
private Optional<InitializeShareGroupStateParameters> maybeCreateInitializeShareGroupStateRequest(
@@ -3822,7 +3828,7 @@ public class GroupMetadataManager {
// 3.3 Or the last refresh is older than the batch refresh max interval.
// 0. The group is subscribed to regular expressions. We also take the one
// that the current may have just introduced.
// that the current member may have just introduced.
if (!requireRefresh && group.subscribedRegularExpressions().isEmpty()) {
return updateRegularExpressionStatus;
}
@@ -4036,7 +4042,6 @@ public class GroupMetadataManager {
return false;
}
/**
/**
* Reconciles the current assignment of the member towards the target assignment if needed.
*
@@ -4166,7 +4171,7 @@ public class GroupMetadataManager {
* @param ownedActiveTasks The list of active tasks owned by the member.
* This is reported in the StreamsGroupHeartbeat API, and
* it could be null if not provided.
* @param ownedStandbyTasks The list of standby owned by the member.
* @param ownedStandbyTasks The list of standby tasks owned by the member.
* This is reported in the StreamsGroupHeartbeat API, and
* it could be null if not provided.
* @param ownedWarmupTasks The list of warmup tasks owned by the member.
@@ -4295,9 +4300,9 @@ public class GroupMetadataManager {
group.setMetadataRefreshDeadline(currentTimeMs + METADATA_REFRESH_INTERVAL_MS, groupEpoch);
// Before 4.0, the coordinator used subscription metadata to keep topic metadata.
// After 4.1, the subscription metadata is replaced by the metadata hash. If there is subscription metadata in log,
// add a tombstone record to remove it.
// The coordinator used to keep the topic metadata in subscription metadata records. The
// metadata hash replaces them. If there is a subscription metadata record in the log, add
// a tombstone record to remove it.
if (group.hasSubscriptionMetadataRecord()) {
records.add(newConsumerGroupSubscriptionMetadataTombstoneRecord(groupId));
}
@@ -4537,8 +4542,9 @@ public class GroupMetadataManager {
* <li>{@code NONE} - nothing changed; the group epoch is not bumped.</li>
* <li>{@code RECOMPUTE} - the group changed (membership, topology or configuration); bump the group epoch and re-run the
* assignor to produce a new final target assignment.</li>
* <li>{@code REFINE} - a warm-up task has caught up; bump the group epoch to advance the in-memory refinement one step
* WITHOUT re-running the assignor (the final target assignment is unchanged).</li>
* <li>{@code REFINED} - a warm-up task has caught up; bump the group epoch to advance the
* in-memory refinement one step WITHOUT re-running the assignor (the final target
* assignment is unchanged).</li>
* </ul>
*/
private enum AssignmentUpdate {
@@ -4823,7 +4829,9 @@ public class GroupMetadataManager {
/**
* Handles leave request from a consumer group member.
*
* @param groupId The group id from the request.
* @param instanceId The instance id from the request.
* @param memberId The member id from the request.
* @param memberEpoch The member epoch from the request.
*
@@ -4863,9 +4871,12 @@ public class GroupMetadataManager {
/**
* Handles leave request from a streams group member.
* @param groupId The group id from the request.
* @param memberId The member id from the request.
* @param memberEpoch The member epoch from the request.
*
* @param groupId The group id from the request.
* @param instanceId The instance id from the request.
* @param memberId The member id from the request.
* @param memberEpoch The member epoch from the request.
* @param shutdownApplication Whether all Streams clients in the group should shut down.
*
* @return A result containing the StreamsGroupHeartbeat response and
* a list of records to update the state machine.
@@ -4920,7 +4931,7 @@ public class GroupMetadataManager {
/**
* Handles the case when a static member decides to leave the group.
* The member is not actually fenced from the group, and instead it's
* The member is not actually fenced from the group, and instead its
* member epoch is updated to -2 to reflect that a member using the given
* instance id decided to leave the group and would be back within session
* timeout.
@@ -4953,7 +4964,7 @@ public class GroupMetadataManager {
/**
* Handles the case when a static member decides to leave the group.
* The member is not actually fenced from the group, and instead it's
* The member is not actually fenced from the group, and instead its
* member epoch is updated to -2 to reflect that a member using the given
* instance id decided to leave the group and would be back within session
* timeout.
@@ -5018,7 +5029,7 @@ public class GroupMetadataManager {
}
/**
* Fences a member from a consumer group and maybe downgrade the consumer group to a classic group.
* Fences a member from a consumer group and maybe downgrades the group to a classic group.
*
* @param group The group.
* @param member The member.
@@ -5035,7 +5046,7 @@ public class GroupMetadataManager {
}
/**
* Fences members from a consumer group and maybe downgrade the consumer group to a classic group.
* Fences members from a consumer group and maybe downgrades the group to a classic group.
*
* @param group The group.
* @param members The members.
@@ -5121,6 +5132,7 @@ public class GroupMetadataManager {
*
* @param group The group.
* @param member The member.
* @param response The response of the CoordinatorResult.
*
* @return A list of records to be applied to the state.
*/
@@ -5522,8 +5534,9 @@ public class GroupMetadataManager {
/**
* Schedules (or reschedules) the session timeout for the share group member.
*
* @param groupId The group id.
* @param memberId The member id.
* @param groupId The group id.
* @param memberId The member id.
* @param sessionTimeoutMs The session timeout.
*/
private void scheduleShareGroupSessionTimeout(
String groupId,
@@ -6598,7 +6611,7 @@ public class GroupMetadataManager {
/**
* Replays StreamsGroupTargetAssignmentMemberKey/Value to update the hard state of
* the consumer group.
* the streams group.
* It updates the target assignment of the member or deletes it.
*
* @param key A StreamsGroupTargetAssignmentMemberKey key.
@@ -6628,7 +6641,7 @@ public class GroupMetadataManager {
/**
* Replays StreamsGroupCurrentMemberAssignmentKey/Value to update the hard state of
* the consumer group.
* the streams group.
* It updates the assignment of a member or deletes it.
*
* @param key A StreamsGroupCurrentMemberAssignmentKey key.
@@ -6874,7 +6887,8 @@ public class GroupMetadataManager {
}
/**
* Counts and updates the number of classic and consumer groups in different states.
* Counts and updates the number of classic, consumer, share and streams groups in different
* states.
*/
public void updateGroupSizeCounter() {
Map<ClassicGroupState, Long> classicGroupSizeCounter = new HashMap<>();
@@ -7202,8 +7216,9 @@ public class GroupMetadataManager {
String memberId = request.memberId();
boolean isUnknownMember = memberId.equals(UNKNOWN_MEMBER_ID);
// Group is created if it does not exist and the member id is UNKNOWN. if member
// is specified but group does not exist, request is rejected with GROUP_ID_NOT_FOUND
// Group is created if it does not exist and the member id is UNKNOWN. If the member id
// is specified but the group does not exist, the request is rejected with
// UNKNOWN_MEMBER_ID.
ClassicGroup group;
if (maybeDeleteEmptyConsumerGroup(groupId, records)) {
log.info("[GroupId {}] Converted the empty consumer group to a classic group.", groupId);
@@ -7472,7 +7487,7 @@ public class GroupMetadataManager {
String groupInstanceId = request.groupInstanceId();
if (group.isInState(DEAD)) {
// If the group is marked as dead, it means the group was recently removed the group
// If the group is marked as dead, it means the group was recently removed
// from the coordinator metadata; it is likely that the group has migrated to some other
// coordinator OR the group is in a transient unstable phase. Let the member retry
// finding the correct coordinator and rejoin.
@@ -8020,7 +8035,9 @@ public class GroupMetadataManager {
* Try to complete the join phase of the initial rebalance.
* Otherwise, extend the rebalance.
*
* @param groupId The group under initial rebalance.
* @param groupId The group under initial rebalance.
* @param delayMs The delay of the initial rebalance.
* @param remainingMs The remaining time of the initial rebalance.
*
* @return The coordinator result that will be appended to the log.
*/
@@ -8122,7 +8139,7 @@ public class GroupMetadataManager {
// Reset the session timeout for members after propagating the member's assignment.
// This is because if any member's session expired while we were still awaiting either
// the leader sync group or the append future, its expiration will be ignored and no
// future heartbeat expectations will not be scheduled.
// future heartbeat expectation will be scheduled.
rescheduleClassicGroupMemberHeartbeat(group, member);
}
}
@@ -8595,7 +8612,7 @@ public class GroupMetadataManager {
) {
if (group.isInState(DEAD)) {
// If the group is marked as dead, it means some other thread has just removed the group
// from the coordinator metadata; this is likely that the group has migrated to some other
// from the coordinator metadata; it is likely that the group has migrated to some other
// coordinator OR the group is in a transient unstable phase. Let the member retry
// finding the correct coordinator and rejoin.
return Optional.of(COORDINATOR_NOT_AVAILABLE);
@@ -9326,7 +9343,7 @@ public class GroupMetadataManager {
// New members may timeout with a pending JoinGroup while the group is still rebalancing, so we have
// to invoke the callback before removing the member. We return UNKNOWN_MEMBER_ID so that the consumer
// will retry the JoinGroup request if is still active.
// will retry the JoinGroup request if it is still active.
group.completeJoinFuture(
member,
new JoinGroupResponseData()
@@ -9555,8 +9572,9 @@ public class GroupMetadataManager {
* @param groupId The group id from the request.
* @param topics The topic information for altering the share group's offsets from the request.
*
* @return A Result containing a pair of ShareGroupHeartbeat response and maybe InitializeShareGroupStateParameters
* and a list of records to update the state machine.
* @return A Result containing a pair of AlterShareGroupOffsets response and
* InitializeShareGroupStateParameters and a list of records to update the state
* machine.
*/
public CoordinatorResult<Map.Entry<AlterShareGroupOffsetsResponseData, InitializeShareGroupStateParameters>, CoordinatorRecord> alterShareGroupOffsets(
String groupId,
@@ -10044,8 +10062,10 @@ public class GroupMetadataManager {
* (the first entry of {@code group.streams.assignors}) is used. If the selected assignor is no
* longer registered on the broker, the coordinator falls back to the default.
*
* @param groupId The group id.
* @param maybeLogWarning Whether to warn about the fallback. Set to false on read-only paths such as
* describe, which would otherwise log on every request.
* @return The assignor of the streams group.
*/
// Visible for testing
TaskAssignor streamsGroupAssignor(String groupId, boolean maybeLogWarning) {
@@ -10067,6 +10087,9 @@ public class GroupMetadataManager {
/**
* Get the assignment configs of the provided streams group.
*
* @param groupId The group id.
* @return The assignment configs of the streams group.
*/
private AssignmentConfigsImpl streamsGroupAssignmentConfigs(String groupId) {
Optional<GroupConfig> groupConfig = groupConfigManager.groupConfig(groupId);
@@ -10167,7 +10190,7 @@ public class GroupMetadataManager {
* @param groupId The group id.
* @param memberId The member id.
*
* @return the sync key.
* @return the join key.
*/
static String consumerGroupJoinKey(String groupId, String memberId) {
return "join-" + groupId + "-" + memberId;
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetAndMetadata.java
+2 -2
View File
@@ -149,7 +149,7 @@ public class OffsetAndMetadata {
}
/**
* @return An OffsetAndMetadata created from a OffsetCommitValue record.
* @return An OffsetAndMetadata created from an OffsetCommitValue record.
*/
public static OffsetAndMetadata fromRecord(
long recordOffset,
@@ -187,7 +187,7 @@ public class OffsetAndMetadata {
}
/**
* @return An OffsetAndMetadata created from an OffsetCommitRequestPartition request.
* @return An OffsetAndMetadata created from a TxnOffsetCommitRequestPartition request.
*/
public static OffsetAndMetadata fromRequest(
Uuid topicId,
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java
+7 -7
View File
@@ -502,7 +502,7 @@ public class OffsetMetadataManager {
}
/**
* Validates an TxnOffsetCommit request.
* Validates a TxnOffsetCommit request.
*
* @param context The request context.
* @param request The actual request.
@@ -553,7 +553,7 @@ public class OffsetMetadataManager {
* Validates an OffsetFetch request.
*
* @param request The actual request.
* @param lastCommittedOffset The last committed offsets in the timeline.
* @param lastCommittedOffset The last committed offset in the timeline.
*/
private void validateOffsetFetch(
OffsetFetchRequestData.OffsetFetchRequestGroup request,
@@ -680,7 +680,7 @@ public class OffsetMetadataManager {
}
/**
* Handles an TxnOffsetCommit request.
* Handles a TxnOffsetCommit request.
*
* @param context The request context.
* @param request The TxnOffsetCommit request.
@@ -889,9 +889,9 @@ public class OffsetMetadataManager {
* Fetch offsets for a given Group.
*
* @param request The OffsetFetchRequestGroup request.
* @param lastCommittedOffset The last committed offsets in the timeline.
* @param lastCommittedOffset The last committed offset in the timeline.
*
* @return A List of OffsetFetchResponseTopics response.
* @return The OffsetFetchResponseGroup response.
*/
public OffsetFetchResponseData.OffsetFetchResponseGroup fetchOffsets(
OffsetFetchRequestData.OffsetFetchRequestGroup request,
@@ -973,9 +973,9 @@ public class OffsetMetadataManager {
* Fetch all offsets for a given Group.
*
* @param request The OffsetFetchRequestGroup request.
* @param lastCommittedOffset The last committed offsets in the timeline.
* @param lastCommittedOffset The last committed offset in the timeline.
*
* @return A List of OffsetFetchResponseTopics response.
* @return The OffsetFetchResponseGroup response.
*/
public OffsetFetchResponseData.OffsetFetchResponseGroup fetchAllOffsets(
OffsetFetchRequestData.OffsetFetchRequestGroup request,
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/ShareGroupAutoOffsetResetStrategy.java
+4 -3
View File
@@ -91,9 +91,10 @@ public class ShareGroupAutoOffsetResetStrategy {
/**
* Delegates the timestamp calculation to the base strategy.
* @return the timestamp for the OffsetResetStrategy,
* if the strategy is EARLIEST or LATEST or duration is provided
* else return Optional.empty()
*
* @return {@code ListOffsetsRequest.EARLIEST_TIMESTAMP} for the EARLIEST strategy,
* {@code ListOffsetsRequest.LATEST_TIMESTAMP} for the LATEST strategy, or the current
* time minus the duration for the BY_DURATION strategy.
*/
public Long timestamp() {
return delegate.timestamp().get();
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java
+1 -1
View File
@@ -190,7 +190,7 @@ public class Utils {
}
/**
* Converts a map of topic id and partition set to a ConsumerProtocolAssignment.
* Converts a ConsumerProtocolAssignment to a map of topic id and partition set.
*
* @param consumerProtocolAssignment The ConsumerProtocolAssignment.
* @param metadataImage The Metadata image.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/SimpleHeterogeneousAssignmentBuilder.java
+6 -4
View File
@@ -61,7 +61,7 @@ public class SimpleHeterogeneousAssignmentBuilder {
private final Set<Uuid> subscribedTopicIds;
/**
* The list of members in the consumer group.
* The list of members in the share group.
*/
private final List<String> memberIds;
@@ -249,9 +249,11 @@ public class SimpleHeterogeneousAssignmentBuilder {
private final Map<Integer, Set<Integer>> finalAssignmentByMember;
/**
* The desired sharing for each target partition.
* For entirely balanced assignment, we would expect (numTargetPartitions / numGroupMembers) partitions per member, rounded upwards.
* That can be expressed as: Math.ceil(numTargetPartitions / (double) numGroupMembers)
* The desired sharing for each target partition, which is the number of members that the
* partition is assigned to.
* For entirely balanced assignment, we would expect
* (numSubscribedMembers / numTargetPartitions) members per partition, rounded upwards.
* That can be expressed as: Math.ceil(numSubscribedMembers / (double) numTargetPartitions)
*/
private final Integer desiredSharing;
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/SimpleHomogeneousAssignmentBuilder.java
+6 -4
View File
@@ -56,7 +56,7 @@ public class SimpleHomogeneousAssignmentBuilder {
private final Set<Uuid> subscribedTopicIds;
/**
* The list of members in the consumer group.
* The list of members in the share group.
*/
private final List<String> memberIds;
@@ -76,9 +76,11 @@ public class SimpleHomogeneousAssignmentBuilder {
private final int numGroupMembers;
/**
* The desired sharing for each target partition.
* For entirely balanced assignment, we would expect (numTargetPartitions / numGroupMembers) partitions per member, rounded upwards.
* That can be expressed as: Math.ceil(numTargetPartitions / (double) numGroupMembers)
* The desired sharing for each target partition, which is the number of members that the
* partition is assigned to.
* For entirely balanced assignment, we would expect (numGroupMembers / numTargetPartitions)
* members per partition, rounded upwards.
* That can be expressed as: Math.ceil(numGroupMembers / (double) numTargetPartitions)
*/
private final int desiredSharing;
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java
+7 -6
View File
@@ -1172,9 +1172,10 @@ public class ClassicGroup implements Group {
try {
Set<String> allSubscribedTopics = new HashSet<>();
members.values().forEach(member -> {
// The consumer protocol is parsed with V0 which is the based prefix of all versions.
// This way the consumer group manager does not depend on any specific existing or
// future versions of the consumer protocol. VO must prefix all new versions.
// The consumer protocol is parsed with V0 which is the base prefix of all
// versions. This way the consumer group manager does not depend on any specific
// existing or future versions of the consumer protocol. V0 must prefix all new
// versions.
ByteBuffer buffer = ByteBuffer.wrap(member.metadata(protocolName.get()));
ConsumerProtocol.deserializeVersion(buffer);
allSubscribedTopics.addAll(new HashSet<>(
@@ -1240,7 +1241,7 @@ public class ClassicGroup implements Group {
}
/**
* Complete all the awaiting join future with the given error.
* Complete all the awaiting join futures with the given error.
*
* @param error the error to complete the future with.
*/
@@ -1270,7 +1271,7 @@ public class ClassicGroup implements Group {
}
/**
* Complete all the awaiting sync future with the give error.
* Complete all the awaiting sync futures with the given error.
*
* @param error the error to complete the future with.
*/
@@ -1416,7 +1417,7 @@ public class ClassicGroup implements Group {
// The serializing version should align with that of the member's JoinGroupRequestProtocol.
String memberId = classicGroupMember.memberId();
if (joiningMember != null && memberId.equals(joiningMember.memberId())) {
// If the downgraded is triggered by the joining static member replacing
// If the downgrade is triggered by the joining static member replacing
// the leaving static member, the joining member should take the assignment
// of the leaving one.
ConsumerGroupMember replacedMember = consumerGroup.staticMember(joiningMember.instanceId());
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShard.java
+4 -4
View File
@@ -50,22 +50,22 @@ public class GroupCoordinatorMetricsShard implements CoordinatorMetricsShard {
}
/**
* Classic group size gauge counters keyed by the metric name.
* Classic group size gauge counters keyed by the group state.
*/
private volatile Map<ClassicGroupState, Long> classicGroupGauges;
/**
* Consumer group size gauge counters keyed by the metric name.
* Consumer group size gauge counters keyed by the group state.
*/
private volatile Map<ConsumerGroupState, Long> consumerGroupGauges;
/**
* Share group size gauge counters keyed by the metric name.
* Share group size gauge counters keyed by the group state.
*/
private volatile Map<ShareGroupState, Long> shareGroupGauges;
/**
* Streams group size gauge counters keyed by the metric name.
* Streams group size gauge counters keyed by the group state.
*/
private volatile Map<StreamsGroupState, Long> streamsGroupGauges;
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/Assignment.java
+4 -4
View File
@@ -74,10 +74,10 @@ public class Assignment implements MemberAssignment {
}
/**
* Creates a {{@link Assignment}} from a {{@link ConsumerGroupTargetAssignmentMemberValue}}.
* Creates an {@link Assignment} from a {@link ConsumerGroupTargetAssignmentMemberValue}.
*
* @param record The record.
* @return A {{@link Assignment}}.
* @return An {@link Assignment}.
*/
public static Assignment fromRecord(
ConsumerGroupTargetAssignmentMemberValue record
@@ -90,10 +90,10 @@ public class Assignment implements MemberAssignment {
}
/**
* Creates a {{@link Assignment}} from a {{@link ShareGroupTargetAssignmentMemberValue}}.
* Creates an {@link Assignment} from a {@link ShareGroupTargetAssignmentMemberValue}.
*
* @param record The record.
* @return A {{@link Assignment}}.
* @return An {@link Assignment}.
*/
public static Assignment fromRecord(
ShareGroupTargetAssignmentMemberValue record
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/GroupSpecBuilder.java
+2 -2
View File
@@ -122,7 +122,7 @@ public abstract class GroupSpecBuilder<T extends ModernGroupMember, U extends Gr
private Map<String, T> members = Map.of();
/**
* The subscription type of the consumer group.
* The subscription type of the group.
*/
private SubscriptionType subscriptionType;
@@ -151,7 +151,7 @@ public abstract class GroupSpecBuilder<T extends ModernGroupMember, U extends Gr
/**
* Adds all the existing members.
*
* @param members The existing members in the consumer group.
* @param members The existing members in the group.
* @return This object.
*/
public U withMembers(
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/MemberState.java
+1 -1
View File
@@ -23,7 +23,7 @@ import java.util.Map;
/**
* The various states that a member can be in. For their definition,
* refer to the documentation of {{@link CurrentAssignmentBuilder}}.
* refer to the documentation of {@link CurrentAssignmentBuilder}.
*/
public enum MemberState {
/**
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/MemberSubscriptionAndAssignmentImpl.java
+1
View File
@@ -38,6 +38,7 @@ public class MemberSubscriptionAndAssignmentImpl implements MemberSubscription,
* Constructs a new {@code MemberSubscriptionAndAssignmentImpl}.
*
* @param rackId The rack Id.
* @param instanceId The instance id.
* @param subscribedTopicIds The set of subscribed topic Ids.
* @param memberAssignment The current member assignment.
*/
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/ModernGroup.java
+7 -5
View File
@@ -84,7 +84,7 @@ public abstract class ModernGroup<T extends ModernGroupMember> implements Group
protected final TimelineHashMap<String, SubscriptionCount> subscribedTopicNames;
/**
* The metadata hash which is computed based on the all subscribed topics.
* The metadata hash which is computed based on all the subscribed topics.
*/
protected final TimelineLong metadataHash;
@@ -247,7 +247,9 @@ public abstract class ModernGroup<T extends ModernGroupMember> implements Group
/**
* Returns the target assignment of the member.
*
* @return The ConsumerGroupMemberAssignment or an EMPTY one if it does not
* @param memberId The member id.
*
* @return The member's target assignment or an EMPTY one if it does not
* exist.
*/
public Assignment targetAssignment(String memberId) {
@@ -444,6 +446,7 @@ public abstract class ModernGroup<T extends ModernGroupMember> implements Group
* Compute the subscription type of the group.
*
* @param subscribedTopicNames A map of topic names to the count of members subscribed to each topic.
* @param numberOfMembers The number of members in the group.
*
* @return {@link SubscriptionType#HOMOGENEOUS} if all members are subscribed to exactly the same topics;
* otherwise, {@link SubscriptionType#HETEROGENEOUS}.
@@ -475,10 +478,9 @@ public abstract class ModernGroup<T extends ModernGroupMember> implements Group
* Gets or creates a member.
*
* @param memberId The member id.
* @param createIfNotExists Booleans indicating whether the member must be
* created if it does not exist.
* @param createIfNotExists Whether the member must be created if it does not exist.
*
* @return A ConsumerGroupMember.
* @return The member.
* @throws UnknownMemberIdException when the member does not exist and createIfNotExists is false.
*/
public abstract T getOrMaybeCreateMember(String memberId, boolean createIfNotExists) throws UnknownMemberIdException;
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/TopicIds.java
+3 -2
View File
@@ -31,7 +31,8 @@ import java.util.Set;
/**
* TopicIds is initialized with topic names (String) but exposes a Set of topic ids (Uuid) to the
* user and performs the conversion lazily with a TopicResolver backed by a TopicsImage.
* user and performs the conversion lazily with a TopicResolver backed by a
* CoordinatorMetadataImage.
*/
public class TopicIds implements Set<Uuid> {
/**
@@ -39,7 +40,7 @@ public class TopicIds implements Set<Uuid> {
*/
public interface TopicResolver {
/**
* @return The TopicsImage used by the resolver.
* @return The metadata image used by the resolver.
*/
CoordinatorMetadataImage image();
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java
+3 -5
View File
@@ -247,8 +247,7 @@ public class ConsumerGroup extends ModernGroup<ConsumerGroupMember> {
* is done via the {@link ConsumerGroup#updateMember(ConsumerGroupMember)} method.
*
* @param memberId The member id.
* @param createIfNotExists Booleans indicating whether the member must be
* created if it does not exist.
* @param createIfNotExists Whether the member must be created if it does not exist.
*
* @return A ConsumerGroupMember.
* @throws UnknownMemberIdException when the member does not exist and createIfNotExists is false.
@@ -598,7 +597,7 @@ public class ConsumerGroup extends ModernGroup<ConsumerGroupMember> {
/**
* Compute the preferred (server side) assignor for the group while
* taking into account the updated member. The computation relies
* on {{@link ConsumerGroup#serverAssignors}} persisted structure
* on {@link ConsumerGroup#serverAssignors} persisted structure
* but it does not update it.
*
* @param oldMember The old member.
@@ -641,8 +640,7 @@ public class ConsumerGroup extends ModernGroup<ConsumerGroupMember> {
* @param memberId The member id.
* @param groupInstanceId The group instance id.
* @param memberEpoch The member epoch.
* @param isTransactional Whether the offset commit is transactional or not. It has no
* impact when a consumer group is used.
* @param isTransactional Whether the offset commit is transactional or not.
* @param apiVersion The api version.
* @return A validator for per-partition validation.
* @throws UnknownMemberIdException If the member is not found.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java
+1 -1
View File
@@ -51,7 +51,7 @@ public class ConsumerGroupMember extends ModernGroupMember {
* A builder that facilitates the creation of a new member or the update of
* an existing one.
*
* Please refer to the javadoc of {{@link ConsumerGroupMember}} for the
* Please refer to the javadoc of {@link ConsumerGroupMember} for the
* definition of the fields.
*/
public static class Builder {
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/CurrentAssignmentBuilder.java
+1 -1
View File
@@ -248,7 +248,7 @@ public class CurrentAssignmentBuilder {
// best option is to fence the member to force it to rejoin the group
// without any partitions and to reconcile it again from scratch.
if (ownedTopicPartitions == null || !ownedTopicPartitions.isEmpty()) {
throw new FencedMemberEpochException("The consumer group member is in a unknown state. "
throw new FencedMemberEpochException("The consumer group member is in an unknown state. "
+ "The member must abandon all its partitions and rejoin.");
}
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java
+8 -4
View File
@@ -76,8 +76,13 @@ public class ShareGroup extends ModernGroup<ShareGroupMember> {
* A record class to hold the value representing ShareGroupStatePartitionMetadata for the TimelineHashmap
* keyed on share group id.
*
* @param initializedTopics Map of set of partition ids keyed on the topic id.
* @param deletingTopics Set of topic ids.
* @param initializingTopics Map of InitMapValue (topic name, partition ids and timestamp)
* keyed on the topic id, for the topics whose share group state is
* being initialized.
* @param initializedTopics Map of InitMapValue (topic name, partition ids and timestamp)
* keyed on the topic id, for the topics whose share group state is
* initialized.
* @param deletingTopics Set of topic ids.
*/
public record ShareGroupStatePartitionMetadataInfo(
Map<Uuid, InitMapValue> initializingTopics,
@@ -192,8 +197,7 @@ public class ShareGroup extends ModernGroup<ShareGroupMember> {
* Gets or creates a member.
*
* @param memberId The member id.
* @param createIfNotExists Booleans indicating whether the member must be
* created if it does not exist.
* @param createIfNotExists Whether the member must be created if it does not exist.
*
* @return A ShareGroupMember.
* @throws UnknownMemberIdException when the member does not exist and createIfNotExists is false.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupMember.java
+1 -1
View File
@@ -43,7 +43,7 @@ public class ShareGroupMember extends ModernGroupMember {
* A builder that facilitates the creation of a new member or the update of
* an existing one.
* <p>
* Please refer to the javadoc of {{@link ShareGroupMember}} for the
* Please refer to the javadoc of {@link ShareGroupMember} for the
* definition of the fields.
*/
public static class Builder {
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java
+3
View File
@@ -1425,6 +1425,9 @@ public class AssignmentRefinerImpl implements AssignmentRefiner {
* What this migration needs from the budget, decided once when the migration is classified. The funding
* order reads it, as does the fall-back: a {@link Warming#SIBLING_MOVE} that does not get a warm-up slot
* falls back to borrowing the standby where it sits, everything else parks.
* @param tier
* The round of funding this migration competes in, lowest first, as decided by
* {@link AssignmentRefinerImpl#fundingTier}.
*/
private record FundingCandidate(
TaskId task,
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilder.java
+1 -1
View File
@@ -209,7 +209,7 @@ public class CurrentAssignmentBuilder {
// without any tasks and to reconcile it again from scratch.
if ((ownedTasks.isEmpty() || !ownedTasks.get().isEmpty())) {
throw new FencedMemberEpochException(
"The streams group member is in a unknown state. "
"The streams group member is in an unknown state. "
+ "The member must abandon all its tasks and rejoin.");
}
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/MemberState.java
+1 -1
View File
@@ -21,7 +21,7 @@ import java.util.Map;
/**
* The various states that a member can be in. For their definition, refer to the documentation of
* {{@link org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder}}.
* {@link org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder}.
*/
public enum MemberState {
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsCoordinatorRecordHelpers.java
+2 -2
View File
@@ -357,7 +357,7 @@ public class StreamsCoordinatorRecordHelpers {
/**
* Creates a StreamsTopology record.
*
* @param groupId The consumer group id.
* @param groupId The streams group id.
* @param topology The new topology.
* @return The record.
*/
@@ -372,7 +372,7 @@ public class StreamsCoordinatorRecordHelpers {
/**
* Creates a StreamsTopology record.
*
* @param groupId The consumer group id.
* @param groupId The streams group id.
* @param value The encoded topology record value.
* @return The record.
*/
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
+17 -17
View File
@@ -193,7 +193,7 @@ public class StreamsGroup implements Group {
private final TimelineInteger failedDescriptionTopologyEpoch;
/**
* The metadata hash which is computed based on the all subscribed topics.
* The metadata hash which is computed based on all the subscribed topics.
*/
protected final TimelineLong metadataHash;
@@ -768,9 +768,9 @@ public class StreamsGroup implements Group {
* associated with that static member instance ID.
*
* @param memberId The member id.
* @param instanceId The instance id.
*
* @return The StreamsGroupMemberAssignment for the resolved member ID, or {@link TasksTuple#EMPTY}
* @param instanceId The instance id.
*
* @return The target assignment for the resolved member ID, or {@link TasksTuple#EMPTY}
* if no assignment exists or no static member exists for {@code instanceId}.
*/
public TasksTuple targetAssignment(String memberId, Optional<String> instanceId) {
@@ -814,7 +814,7 @@ public class StreamsGroup implements Group {
/**
* Returns the current process ID of a task or null if the task does not have one.
*
* @param subtopologyId The topic ID.
* @param subtopologyId The subtopology ID.
* @param taskId The task ID.
* @return The process ID or null.
*/
@@ -832,7 +832,7 @@ public class StreamsGroup implements Group {
/**
* Returns the current process IDs of a task or empty set if the task does not have one.
*
* @param subtopologyId The topic ID.
* @param subtopologyId The subtopology ID.
* @param taskId The task ID.
* @return The process IDs or empty set.
*/
@@ -848,11 +848,11 @@ public class StreamsGroup implements Group {
}
/**
* Returns the current process ID of a task or empty set if the task does not have one.
* Returns the current process IDs of a task or empty set if the task does not have one.
*
* @param subtopologyId The topic ID.
* @param taskId The process ID.
* @return The member IDs or empty set.
* @param subtopologyId The subtopology ID.
* @param taskId The task ID.
* @return The process IDs or empty set.
*/
public Set<String> currentWarmupTaskProcessIds(
String subtopologyId, int taskId
@@ -1306,9 +1306,9 @@ public class StreamsGroup implements Group {
/**
* Removes the task process IDs based on the provided assignment.
*
* @param assignment The assignment.
* @param expectedProcessId The expected process ID.
* package-private for testing.
* @param assignment The assignment.
* @param currentTasksProcessId The current process ID of each task, which is updated.
* @param expectedProcessId The expected process ID.
*/
private void removeTaskProcessIds(
Map<String, Map<Integer, Integer>> assignment,
@@ -1344,9 +1344,9 @@ public class StreamsGroup implements Group {
/**
* Removes the task process IDs based on the provided assignment.
*
* @param assignment The assignment.
* @param processIdToRemove The expected process ID.
* package-private for testing.
* @param assignment The assignment.
* @param currentTasksProcessId The current process IDs of each task, which is updated.
* @param processIdToRemove The process ID to remove.
*/
private void removeTaskProcessIdsFromSet(
Map<String, Set<Integer>> assignment,
@@ -1377,7 +1377,7 @@ public class StreamsGroup implements Group {
}
/**
* Adds the partition epoch based on the provided assignment.
* Adds the process ID of the tasks based on the provided assignment.
*
* @param tasks The assigned tasks.
* @param processId The process ID.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java
+1 -1
View File
@@ -80,7 +80,7 @@ public record StreamsGroupMember(String memberId,
/**
* A builder that facilitates the creation of a new member or the update of an existing one.
* <p>
* Please refer to the javadoc of {{@link StreamsGroupMember}} for the definition of the fields.
* Please refer to the javadoc of {@link StreamsGroupMember} for the definition of the fields.
*/
public static class Builder {
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoff.java
+2 -2
View File
@@ -27,8 +27,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
* description push. An entry is armed when the broker decides to set
* {@code TopologyDescriptionRequired=true} on a heartbeat or after a transient plugin
* failure; consecutive arms at the same topology epoch advance the attempt count and
* delegate to {@link ExponentialBackoff} for the next delay . Successful pushes, permanent plugin failures,
* and topology-epoch advances clear the entry.
* delegate to {@link ExponentialBackoff} for the next delay. Successful pushes, permanent plugin
* failures, and topology-epoch advances clear the entry.
*/
public class StreamsGroupTopologyDescriptionBackoff {
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
+9 -9
View File
@@ -63,7 +63,7 @@ import static org.apache.kafka.common.requests.StreamsGroupDescribeResponse.TOPO
* and the per-group re-solicitation back-off. The push-RPC chain (validate → convert →
* plugin → metadata write → back-off mutation, see {@link #pushTopology}) and the periodic
* cleanup cycle's body (list eligible groups → mark uncertain → plugin delete → finalize, see
* {@link #runCleanupCycle}) both live here now, reading and writing group state through
* {@link #runCleanupCycle}) both live here, reading and writing group state through
* {@link TopologyDescriptionRuntime} so the same chains run unchanged on any host that supplies
* an implementation of that interface. {@code GroupCoordinatorService} still owns the broker
* timer and starts the cycle via {@link #startCleanupCycle}, which wraps the supplied cycle body
@@ -603,11 +603,11 @@ public class StreamsGroupTopologyDescriptionManager implements AutoCloseable {
}
/**
* Drop the back-off entry for a group unconditionally. Currently called when a group is
* removed via explicit DeleteGroups and on a post-plugin write failing with
* GroupIdNotFoundException. NOTE: groups removed by other lifecycle paths (session expiry,
* partition unload, tombstone-via-replay) are not yet wired to this, so their back-off
* entries can leak until the group id is reused. Delegates to
* Drop the back-off entry for a group unconditionally. Called once the plugin delete has
* succeeded for the group, by the cleanup cycle, the DeleteGroups flow and the classic-join
* conversion. NOTE: groups removed by other lifecycle paths (session expiry, partition
* unload, tombstone-via-replay) are not yet wired to this, so their back-off entries can
* leak until the group id is reused. Delegates to
* {@link StreamsGroupTopologyDescriptionBackoff#clearGroup}.
*/
private void clearBackoffGroup(String groupId) {
@@ -625,9 +625,9 @@ public class StreamsGroupTopologyDescriptionManager implements AutoCloseable {
* exceptional future.
*
* <p>Pure plugin invocation: does not read group state, does not touch the back-off
* map, and does not record metrics. The service layer pre-filters the input, records
* delete-success / delete-error sensors on the returned failure count, and is
* responsible for invoking {@link #clearBackoffGroup} for the groups it chose to clear.
* map, and does not record metrics. The callers pre-filter the input, record the
* delete-success and delete-error sensors on the returned failure count, and invoke
* {@link #clearBackoffGroup} for the groups whose delete succeeded.
*/
private CompletableFuture<Map<String, ApiError>> invokeDeleteTopologies(Set<String> groupIds) {
if (plugin.isEmpty() || groupIds.isEmpty()) {
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java
+2 -2
View File
@@ -49,7 +49,7 @@ public class TargetAssignmentBuilder {
private final int groupEpoch;
/**
* The partition assignor used to compute the assignment.
* The task assignor used to compute the assignment.
*/
private final TaskAssignor assignor;
@@ -107,7 +107,7 @@ public class TargetAssignmentBuilder {
}
/**
* Adds the topology image.
* Adds the configured topology.
*
* @param topology The topology.
* @return This object.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TasksTuple.java
+3 -3
View File
@@ -106,11 +106,11 @@ public record TasksTuple(Map<String, Set<Integer>> activeTasks,
}
/**
* Creates a {{@link TasksTuple}} from a
* {{@link org.apache.kafka.coordinator.group.generated.StreamsGroupTargetAssignmentMemberValue}}.
* Creates a {@link TasksTuple} from a
* {@link org.apache.kafka.coordinator.group.generated.StreamsGroupTargetAssignmentMemberValue}.
*
* @param record The record.
* @return A {{@link TasksTuple}}.
* @return A {@link TasksTuple}.
*/
public static TasksTuple fromTargetAssignmentRecord(StreamsGroupTargetAssignmentMemberValue record) {
return new TasksTuple(
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TasksTupleWithEpochs.java
+7 -5
View File
@@ -115,11 +115,13 @@ public record TasksTupleWithEpochs(Map<String, Map<Integer, Integer>> activeTask
/**
* Creates a TasksTupleWithEpochs from a current assignment record.
*
* @param activeTasks The active tasks from the record.
* @param standbyTasks The standby tasks from the record.
* @param warmupTasks The warmup tasks from the record.
* @param memberEpoch The member epoch to use as default for tasks without explicit epochs.
* @return The TasksTupleWithEpochs
* @param log The logger.
* @param groupId The group id.
* @param activeTasks The active tasks from the record.
* @param standbyTasks The standby tasks from the record.
* @param warmupTasks The warmup tasks from the record.
* @param memberEpoch The member epoch to use as default for tasks without explicit epochs.
* @return The TasksTupleWithEpochs.
*/
public static TasksTupleWithEpochs fromCurrentAssignmentRecord(
Logger log,
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopologyMetadata.java
+3 -3
View File
@@ -28,10 +28,10 @@ import java.util.SortedMap;
/**
* The topology metadata class is used by the {@link org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignor} to get topic and
* partition metadata for the topology that the streams group using.
* partition metadata for the topology that the streams group is using.
*
* @param metadataImage The metadata image
* @param subtopologyMap The configured subtopologies
* @param metadataImage The metadata image.
* @param subtopologyMap The configured subtopologies.
*/
public record TopologyMetadata(CoordinatorMetadataImage metadataImage, SortedMap<String, ConfiguredSubtopology> subtopologyMap) implements TopologyDescriber {
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/ProcessState.java
+1 -1
View File
@@ -140,7 +140,7 @@ public class ProcessState {
}
/**
* Assigns a task to the least loaded member of this process
* Assigns a task to the least loaded member of this process.
*
* @param taskId The task to assign.
* @param isActive Whether the task is an active task (true) or a standby task (false).
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredSubtopology.java
+7 -5
View File
@@ -26,14 +26,16 @@ import java.util.Set;
/**
* Internal representation of a subtopology.
* <p>
* The subtopology is configured according to the number of partitions available in the source topics. It has regular expressions already
* resolved and defined exactly the information that is being used by streams groups assignment reconciliation.
* The subtopology is configured according to the number of partitions available in the source
* topics. It has regular expressions already resolved and defines exactly the information that is
* being used by streams groups assignment reconciliation.
* <p>
* Configured subtopologies may be recreated every time the input topics used by the subtopology are modified.
*
* @param numberOfTasks Precomputed number of tasks. Not that not every source topic may have a partition for
* every task, in cases where there are multiple source topics with an unequal number of
* partitions (e.g., one topic has 3 partitions and another has 5 and both are used in a
* @param numberOfTasks Precomputed number of tasks. Note that not every source topic may
* have a partition for every task, in cases where there are multiple
* source topics with an unequal number of partitions (e.g., one
* topic has 3 partitions and another has 5 and both are used in a
* merge).
* @param sourceTopics The source topics of the subtopology.
* @param repartitionSourceTopics The repartition source topics of the subtopology.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredTopology.java
+4 -3
View File
@@ -37,9 +37,10 @@ import java.util.SortedMap;
* failed.
* @param internalTopicsToBeCreated Contains a list of internal topics that need to be created. This is used to create the topics in the
* broker.
* @param topicConfigurationException If the topic configuration process failed, e.g. because expected topics are missing or have an
* incorrect number of partitions, this field will store the error that occurred, so that is can be
* reported back to the client.
* @param topicConfigurationException If the topic configuration process failed, e.g. because
* expected topics are missing or have an incorrect number of
* partitions, this field will store the error that occurred, so
* that it can be reported back to the client.
*/
public record ConfiguredTopology(int topologyEpoch,
long metadataHash,
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java
+5 -3
View File
@@ -44,9 +44,11 @@ public class CopartitionedTopicsEnforcer {
* The constructor for the class.
*
* @param log The logger.
* @param topicPartitionCountProvider Returns the number of partitions for a given topic, representing the current state of the broker
* as well as any partition number decisions that have already been made. In particular, we expect
* the number of partitions for all topics in all co-partitions groups to be defined.
* @param topicPartitionCountProvider Returns the number of partitions for a given topic,
* representing the current state of the broker as well as
* any partition number decisions that have already been
* made. In particular, we expect the number of partitions
* for all topics in all co-partition groups to be defined.
*/
public CopartitionedTopicsEnforcer(final Logger log,
final Function<String, OptionalInt> topicPartitionCountProvider) {
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java
+3 -2
View File
@@ -47,8 +47,9 @@ public class InternalTopicManager {
/**
* Configures the internal topics for the given topology. Given a topology and the metadata image, this method determines the number of
* partitions for all internal topics and returns a {@link ConfiguredTopology} object.
* Configures the internal topics for the given topology. Given a topology and the metadata
* image, this method determines the number of partitions for all internal topics and returns a
* {@link ConfiguredTopology} object.
*
* @param log The logger.
* @param groupId The group id.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/RepartitionTopics.java
+9 -6
View File
@@ -31,9 +31,10 @@ import java.util.Set;
import java.util.function.Function;
/**
* Responsible for configuring the number of partitions in repartitioning topics. It computes a fix-point iteration, deriving the number of
* partitions for each repartition topic based on the number of partitions of the source topics of the topology, if the number of
* partitions is not explicitly set in the topology.
* Responsible for configuring the number of partitions in repartitioning topics. It computes a
* fixed-point iteration, deriving the number of partitions for each repartition topic based on the
* number of partitions of the source topics of the topology, if the number of partitions is not
* explicitly set in the topology.
*/
public class RepartitionTopics {
@@ -46,8 +47,10 @@ public class RepartitionTopics {
*
* @param log The logger.
* @param subtopologies The subtopologies for the requested topology.
* @param topicPartitionCountProvider Returns the number of partitions for a given topic, representing the current state of the
* broker. This class requires the number of partition for all source topics to be defined.
* @param topicPartitionCountProvider Returns the number of partitions for a given topic,
* representing the current state of the broker. This class
* requires the number of partitions for all source topics
* to be defined.
*/
public RepartitionTopics(final Logger log,
final Collection<Subtopology> subtopologies,
@@ -58,7 +61,7 @@ public class RepartitionTopics {
}
/**
* Returns the set of the number of partitions for each repartition topic.
* Returns the number of partitions for each repartition topic.
*
* @return the map of repartition topics for the requested topology to their required number of partitions.
*