You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2026/06/10 11:06:33 UTC
(kafka) branch trunk updated: KAFKA-20169: Support static membership for Kafka Streams with the streams rebalance protocol (1/N) (#22245)
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new eeac46d3331 KAFKA-20169: Support static membership for Kafka Streams with the streams rebalance protocol (1/N) (#22245)
eeac46d3331 is described below
commit eeac46d333119d4d16f83661cfd5ad718545c397
Author: ChickenchickenLove <oj...@naver.com>
AuthorDate: Wed Jun 10 20:06:26 2026 +0900
KAFKA-20169: Support static membership for Kafka Streams with the streams rebalance protocol (1/N) (#22245)
### Introduction
This PR enables "static membership for Streams group heartbeat" and
aligns group-epoch behavior with the KIP-1071 intent for Streams
rebalancing.
### Notification
1. According to KIP-1071, the group epoch must be bumped only when one
of the following member attributes changes: topology epoch, rack ID,
client tags, or process ID. However, the previous implementation
determined whether to bump the group epoch by comparing the entire
member record. In this PR, the static membership path is updated so that
the group epoch is bumped only when at least one of topology epoch, rack
ID, client tags, or process ID differs.
2. When a static member leaves with member epoch -2 and later rejoins
with the same instance ID, we treat this as not constituting a member
join/leave for the purpose of group-epoch bumping. Accordingly, the
implementation does not bump the group epoch in this case, since it does
not satisfy the “member join/leave” condition described in KIP-1071.
```
### KIP-1071 says
The group epoch is bumped:
- When a member joins or leaves the group.
- When a member is fenced or removed from the group by the group
coordinator.
- When the partition metadata is updated. For instance when a new
partition is added or a new topic matching the - subscribed topics is
created.
- When a member with an assigned warm-up task reports a task changelog
offset and task changelog end offset whose difference is less than
acceptable.recovery.lag.
- When a member updates its topology metadata, rack ID, client tags or
process ID. Note: Typically, these do not change within the lifetime of
a Streams client, so this only happens when a member with static
membership rejoins with an updated configuration.
- When an assignment configuration for the group is updated.
```
### Changes
- Removed service-layer rejection of static membership in Streams
heartbeat.
- Added static-member validation paths in Streams coordinator logic:
- unreleased instance id
- fenced instance id
- unknown static member
- Added static Streams member subscribe/replace flow:
- supports join of new static member
- supports rejoin replacement with new `memberId` for same
`instanceId`
- writes replacement records (tombstone old + create new)
- Added static leave handling for Streams:
- `-2` (temporary leave): keep static identity and write
current-assignment epoch `-2`
- `-1` (actual leave): fence/remove member and bump group epoch
- Updated max-size handling:
- If `instanceId` already maps to an existing static member, skip
max-size rejection for replacement flow
- Updated assignment continuity for static rejoin:
- when needed, resolve target assignment from previous static member
id
- Updated epoch bump decision for static rejoin:
- do not bump on member-id change alone
- bump only on epoch-relevant metadata changes (topology epoch, rack
id, client tags, process id)
- Added helper builder support for cloning Streams member with a new
member id.
- Added utility method to check whether a static member mapping is
currently valid.
- Add test case.
### Scope
- Only server-side implementation is included.
Reviewers: Lucas Brutschy <lb...@confluent.io>, Sean Quah
<sq...@confluent.io>
---
.../coordinator/group/GroupCoordinatorService.java | 1 -
.../coordinator/group/GroupMetadataManager.java | 358 +++-
.../coordinator/group/streams/StreamsGroup.java | 43 +-
.../group/streams/StreamsGroupMember.java | 38 +-
.../group/GroupCoordinatorServiceTest.java | 17 +-
.../group/GroupMetadataManagerTest.java | 8 +-
.../group/GroupMetadataManagerTestContext.java | 25 +-
.../StreamsGroupMixedGroupMetadataManagerTest.java | 899 +++++++++
...sGroupStaticMemberGroupMetadataManagerTest.java | 1928 ++++++++++++++++++++
.../coordinator/group/StreamsGroupTestUtil.java | 230 +++
10 files changed, 3489 insertions(+), 58 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
index 93fd360be24..2911736a010 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
@@ -584,7 +584,6 @@ public class GroupCoordinatorService implements GroupCoordinator {
private static void throwIfStreamsGroupHeartbeatRequestIsUsingUnsupportedFeatures(
StreamsGroupHeartbeatRequestData request
) throws InvalidRequestException {
- throwIfNotNull(request.instanceId(), "Static membership is not yet supported.");
throwIfNotNull(request.taskOffsets(), "TaskOffsets are not supported yet.");
throwIfNotNull(request.taskEndOffsets(), "TaskEndOffsets are not supported yet.");
throwIfNotNullOrEmpty(request.warmupTasks(), "WarmupTasks are not supported yet.");
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index faa36fcb18e..ac4b38d53c6 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -76,6 +76,7 @@ import org.apache.kafka.common.requests.ConsumerGroupHeartbeatResponse;
import org.apache.kafka.common.requests.JoinGroupRequest;
import org.apache.kafka.common.requests.ShareGroupHeartbeatRequest;
import org.apache.kafka.common.requests.ShareGroupHeartbeatResponse;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.internals.LogContext;
@@ -150,6 +151,7 @@ import org.apache.kafka.coordinator.group.modern.share.ShareGroup.InitMapValue;
import org.apache.kafka.coordinator.group.modern.share.ShareGroup.ShareGroupStatePartitionMetadataInfo;
import org.apache.kafka.coordinator.group.modern.share.ShareGroupAssignmentBuilder;
import org.apache.kafka.coordinator.group.modern.share.ShareGroupMember;
+import org.apache.kafka.coordinator.group.streams.StreamsCoordinatorRecordHelpers;
import org.apache.kafka.coordinator.group.streams.StreamsGroup;
import org.apache.kafka.coordinator.group.streams.StreamsGroupDescribeResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
@@ -312,7 +314,7 @@ public class GroupMetadataManager {
if (member.isPresent()) {
return new UpdateTargetAssignmentResult<>(
group.assignmentEpoch(),
- group.targetAssignment(member.get().memberId())
+ group.targetAssignment(member.get().memberId(), member.get().instanceId())
);
} else {
return new UpdateTargetAssignmentResult<>(
@@ -1690,6 +1692,26 @@ public class GroupMetadataManager {
}
}
+ /**
+ * Validates if the received instanceId has been released from the group
+ *
+ * @param member The streams group member.
+ * @param groupId The streams group id.
+ * @param receivedMemberId The member id received in the request.
+ * @param receivedInstanceId The instance id received in the request.
+ *
+ * @throws UnreleasedInstanceIdException if the instance id received in the request is still in use by an existing static member.
+ */
+ private void throwIfInstanceIdIsUnreleased(StreamsGroupMember member, String groupId, String receivedMemberId, String receivedInstanceId) {
+ if (member.memberEpoch() != LEAVE_GROUP_STATIC_MEMBER_EPOCH) {
+ // The new member can't join.
+ log.info("[GroupId {}] Static member {} with instance id {} cannot join the group because the instance id is" +
+ " owned by member {}.", groupId, receivedMemberId, receivedInstanceId, member.memberId());
+ throw Errors.UNRELEASED_INSTANCE_ID.exception("Static member " + receivedMemberId + " with instance id "
+ + receivedInstanceId + " cannot join the group because the instance id is owned by " + member.memberId() + " member.");
+ }
+ }
+
/**
* Validates if the received instanceId has been released from the group
*
@@ -1709,6 +1731,25 @@ public class GroupMetadataManager {
}
}
+ /**
+ * Validates if the received instanceId has been fenced from the group
+ *
+ * @param member The streams group member.
+ * @param groupId The streams group id.
+ * @param receivedMemberId The member id received in the request.
+ * @param receivedInstanceId The instance id received in the request.
+ *
+ * @throws FencedInstanceIdException if the instance id provided is fenced because of another static member.
+ */
+ private void throwIfInstanceIdIsFenced(StreamsGroupMember member, String groupId, String receivedMemberId, String receivedInstanceId) {
+ if (!member.memberId().equals(receivedMemberId)) {
+ log.info("[GroupId {}] Static member {} with instance id {} is fenced by existing member {}.",
+ groupId, receivedMemberId, receivedInstanceId, member.memberId());
+ throw Errors.FENCED_INSTANCE_ID.exception("Static member " + receivedMemberId + " with instance id "
+ + receivedInstanceId + " was fenced by member " + member.memberId() + ".");
+ }
+ }
+
/**
* Validates if the received instanceId has been released from the group
*
@@ -1723,20 +1764,45 @@ public class GroupMetadataManager {
}
}
+ /**
+ * 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.
+ *
+ * @throws UnknownMemberIdException if no static member exists in the group against the provided instance id.
+ */
+ private void throwIfStaticMemberIsUnknown(StreamsGroupMember staticMember, String receivedInstanceId) {
+ if (staticMember == null) {
+ throw Errors.UNKNOWN_MEMBER_ID.exception("Instance id " + receivedInstanceId + " is unknown.");
+ }
+ }
+
/**
* 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 memberId The member id.
*
* @throws GroupMaxSizeReachedException if the maximum capacity has been reached.
*/
private void throwIfStreamsGroupIsFull(
- StreamsGroup group
+ StreamsGroup group,
+ String instanceId,
+ String memberId
) throws GroupMaxSizeReachedException {
+ // If a static member already exists, we do not enforce the maximum group size check.
+ // An existing static member will fall into one of the following two cases,
+ // and neither affects the group size:
+ // 1. The member is replaced due to the static member rejoining.
+ // 2. 'UnreleasedInstanceIdException' is raised due to an epoch mismatch.
+ if (group.hasStaticMember(instanceId))
+ return;
+
// If the streams group has reached its maximum capacity, the member is rejected if it is not
// already a member of the streams group.
- if (group.numMembers() >= config.streamsGroupMaxSize()) {
+ if (group.numMembers() >= config.streamsGroupMaxSize() && (memberId.isEmpty() || !group.hasMember(memberId))) {
throw new GroupMaxSizeReachedException("The streams group has reached its maximum capacity of "
+ config.streamsGroupMaxSize() + " members.");
}
@@ -1984,17 +2050,18 @@ public class GroupMetadataManager {
final List<StreamsGroupHeartbeatResponseData.Status> returnedStatus = new ArrayList<>();
// Get or create the streams group.
- boolean isJoining = memberEpoch == 0;
+ boolean isJoining = memberEpoch == StreamsGroupHeartbeatRequest.JOIN_GROUP_MEMBER_EPOCH;
StreamsGroup group;
if (isJoining) {
group = getOrCreateStreamsGroup(groupId, records);
- throwIfStreamsGroupIsFull(group);
+ throwIfStreamsGroupIsFull(group, instanceId, memberId);
} else {
group = getStreamsGroupOrThrow(groupId);
}
// Get or create the member.
StreamsGroupMember member;
+ StreamsGroupMember replaceStaticOldMember = null;
if (instanceId == null) {
member = getOrMaybeCreateDynamicStreamsGroupMember(
group,
@@ -2006,25 +2073,47 @@ public class GroupMetadataManager {
isJoining
);
} else {
- throw new UnsupportedOperationException("Static members are not supported yet.");
+ StreamsGroupMember maybeOldStaticMember = group.staticMember(instanceId);
+ if (maybeOldStaticMember != null && !maybeOldStaticMember.memberId().equals(memberId)) {
+ replaceStaticOldMember = maybeOldStaticMember;
+ }
+ member = getOrMaybeCreateStaticStreamsGroupMember(
+ group,
+ memberId,
+ memberEpoch,
+ instanceId,
+ ownedActiveTasks,
+ ownedStandbyTasks,
+ ownedWarmupTasks,
+ isJoining,
+ records
+ );
}
// 1. Create or update the member.
- StreamsGroupMember updatedMember = new StreamsGroupMember.Builder(member)
- .maybeUpdateInstanceId(Optional.empty())
+ StreamsGroupMember.Builder updatedMemberBuilder = new StreamsGroupMember.Builder(member)
+ .maybeUpdateInstanceId(Optional.ofNullable(instanceId))
.maybeUpdateRackId(Optional.ofNullable(rackId))
.maybeUpdateRebalanceTimeoutMs(ofSentinel(rebalanceTimeoutMs))
.maybeUpdateTopologyEpoch(topology != null ? OptionalInt.of(topology.epoch()) : OptionalInt.empty())
.setClientId(clientId)
.setClientHost(clientHost)
.maybeUpdateProcessId(Optional.ofNullable(processId))
- .maybeUpdateClientTags(Optional.ofNullable(clientTags).map(x -> x.stream().collect(Collectors.toMap(KeyValue::key, KeyValue::value))))
- .maybeUpdateUserEndpoint(Optional.ofNullable(userEndpoint).map(x -> new StreamsGroupMemberMetadataValue.Endpoint().setHost(x.host()).setPort(x.port())))
- .build();
+ .maybeUpdateClientTags(Optional.ofNullable(clientTags).map(x -> x.stream().collect(Collectors.toMap(KeyValue::key, KeyValue::value))));
+ if (isJoining) {
+ StreamsGroupMemberMetadataValue.Endpoint userEndpointMetadata = userEndpoint == null ? null :
+ new StreamsGroupMemberMetadataValue.Endpoint().setHost(userEndpoint.host()).setPort(userEndpoint.port());
+ updatedMemberBuilder.setUserEndpoint(userEndpointMetadata);
+ } else {
+ updatedMemberBuilder
+ .maybeUpdateUserEndpoint(Optional.ofNullable(userEndpoint).map(x -> new StreamsGroupMemberMetadataValue.Endpoint().setHost(x.host()).setPort(x.port())));
+ }
+ 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.
- boolean bumpGroupEpoch = hasStreamsMemberMetadataChanged(groupId, member, updatedMember, records);
+ boolean bumpGroupEpoch = hasStreamsMemberMetadataChanged(groupId, instanceId, member, updatedMember, records);
// 2. Initialize/Update the group topology.
// If the topology is new or has changed, a StreamsGroupTopologyValue record is written to the __consumer_offsets partition to persist
@@ -2165,20 +2254,31 @@ public class GroupMetadataManager {
// The assignment is only provided in the following cases:
// 1. The member is joining.
// 2. The member's assignment has been updated.
- if (memberEpoch == 0 || hasAssignedTasksChanged(member, updatedMember)) {
+ boolean assignedTaskChanged = hasAssignedTasksChanged(member, updatedMember);
+ boolean endpointChanged = hasUserEndpointChanged(member, updatedMember);
+
+ // Echo the assignment back when joining or the assignment changed.
+ if (isJoining || assignedTaskChanged) {
response.setActiveTasks(createStreamsGroupHeartbeatResponseTaskIdsFromEpochs(updatedMember.assignedTasks().activeTasksWithEpochs()));
response.setStandbyTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().standbyTasks()));
response.setWarmupTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().warmupTasks()));
+ }
+
+ // Drop stale per-member endpoint mappings.
+ if (isJoining || assignedTaskChanged || endpointChanged || replaceStaticOldMember != null) {
group.invalidateCachedEndpointToPartitions(updatedMember.memberId());
- if (updatedMember.userEndpoint().isPresent()) {
- // If no user endpoint is defined, there is no change in the endpoint information.
- // Otherwise, bump the endpoint information epoch
- group.setEndpointInformationEpoch(group.endpointInformationEpoch() + 1);
+ if (replaceStaticOldMember != null) {
+ group.invalidateCachedEndpointToPartitions(replaceStaticOldMember.memberId());
}
}
+ // Bump the group's endpoint epoch so peers refetch endpoint-to-partition mappings.
+ if (endpointChanged || (assignedTaskChanged && updatedMember.userEndpoint().isPresent())) {
+ group.setEndpointInformationEpoch(group.endpointInformationEpoch() + 1);
+ }
+
if (group.endpointInformationEpoch() != memberEndpointEpoch) {
- response.setPartitionsByUserEndpoint(group.buildEndpointToPartitions(updatedMember, metadataImage));
+ response.setPartitionsByUserEndpoint(group.buildEndpointToPartitions(updatedMember, metadataImage, replaceStaticOldMember));
}
if (groups.containsKey(group.groupId())) {
// If we just created the group, the endpoint information epoch will not be persisted, so return epoch 0.
@@ -3206,6 +3306,79 @@ public class GroupMetadataManager {
}
}
+ /**
+ * Gets an existing static Streams group member or creates/replaces one for static membership.
+ *
+ * If the member is joining:
+ * 1. Creates a new static member when no member exists for the instance ID.
+ * 2. Replaces the previous static member when the instance ID is released.
+ *
+ * If the member is not joining, validates static member identity and member epoch
+ * and returns the existing static member.
+ *
+ * @param group The streams group.
+ * @param memberId The member id from the request.
+ * @param memberEpoch The member epoch from the request.
+ * @param instanceId The static instance id from the request.
+ * @param ownedActiveTasks The owned active tasks from the request.
+ * @param ownedStandbyTasks The owned standby tasks from the request.
+ * @param ownedWarmupTasks The owned warmup tasks from the request.
+ * @param memberIsJoining Whether the member is joining (epoch 0).
+ * @param records The records accumulator used for member replacement.
+ *
+ * @return The resolved streams group member.
+ */
+ private StreamsGroupMember getOrMaybeCreateStaticStreamsGroupMember(
+ StreamsGroup group,
+ String memberId,
+ int memberEpoch,
+ String instanceId,
+ List<StreamsGroupHeartbeatRequestData.TaskIds> ownedActiveTasks,
+ List<StreamsGroupHeartbeatRequestData.TaskIds> ownedStandbyTasks,
+ List<StreamsGroupHeartbeatRequestData.TaskIds> ownedWarmupTasks,
+ boolean memberIsJoining,
+ List<CoordinatorRecord> records
+ ) {
+ StreamsGroupMember existingStaticMemberOrNull = group.staticMember(instanceId);
+ if (memberIsJoining) {
+ // A new static member joins or the existing static member rejoins.
+ if (existingStaticMemberOrNull == null) {
+ // New static member.
+ StreamsGroupMember newMember = group.getOrCreateDefaultMember(memberId);
+ log.info("[GroupId {}][MemberId {}] Static member {} with instance id {} joins the streams group.",
+ group.groupId(), memberId, memberId, instanceId);
+ return newMember;
+ } else {
+ throwIfInstanceIdIsUnreleased(existingStaticMemberOrNull, group.groupId(), memberId, instanceId);
+
+ // Copy the member but with its new member id.
+ StreamsGroupMember newMember = new StreamsGroupMember.Builder(existingStaticMemberOrNull, memberId)
+ .setMemberEpoch(0)
+ .setPreviousMemberEpoch(0)
+ .build();
+
+ replaceStreamsMember(records, group, existingStaticMemberOrNull, newMember);
+
+ log.info("[GroupId {}][MemberId {}] Static member with instance id {} re-joins the streams group " +
+ "using the streams protocol. Created a new member {} to replace the existing member {}.",
+ group.groupId(), memberId, instanceId, memberId, existingStaticMemberOrNull.memberId());
+
+ return newMember;
+ }
+ } else {
+ throwIfStaticMemberIsUnknown(existingStaticMemberOrNull, instanceId);
+ throwIfInstanceIdIsFenced(existingStaticMemberOrNull, group.groupId(), memberId, instanceId);
+ throwIfStreamsGroupMemberEpochIsInvalid(
+ existingStaticMemberOrNull,
+ memberEpoch,
+ ownedActiveTasks,
+ ownedStandbyTasks,
+ ownedWarmupTasks
+ );
+ return existingStaticMemberOrNull;
+ }
+ }
+
/**
* Gets or subscribes a new dynamic share group member.
*
@@ -3542,11 +3715,13 @@ public class GroupMetadataManager {
}
/**
- * Creates the member metadata record record if the updatedMember is different from
- * the old member. Returns true if the metadata has changed, which is always the case
- * when a member is first created.
+ * Creates the member metadata record if the updatedMember is different from the
+ * old member. Returns whether the group epoch should be bumped. Dynamic members
+ * bump the group epoch on any member metadata change, while static members bump
+ * it only when an epoch-relevant member configuration changes.
*
* @param groupId The group id.
+ * @param instanceId The instance id.
* @param member The old member.
* @param updatedMember The updated member.
* @param records The list to accumulate any new records.
@@ -3555,6 +3730,7 @@ public class GroupMetadataManager {
*/
private boolean hasStreamsMemberMetadataChanged(
String groupId,
+ String instanceId,
StreamsGroupMember member,
StreamsGroupMember updatedMember,
List<CoordinatorRecord> records
@@ -3564,8 +3740,17 @@ public class GroupMetadataManager {
records.add(newStreamsGroupMemberRecord(groupId, updatedMember));
log.info("[GroupId {}][MemberId {}] Member updated its member metadata to {}.",
groupId, memberId, updatedMember);
-
- return true;
+ if (instanceId == null) {
+ return true;
+ } else {
+ // When a static member rejoins, its member ID may change.
+ // According to KIP-1071, the group epoch should be bumped only
+ // when the topology metadata, rack ID, client tags, or process ID changes.
+ // Therefore, if we compare the old and new members using equals(),
+ // the group epoch can be bumped unintentionally,
+ // which may trigger unnecessary task rebalancing.
+ return hasEpochRelevantMemberConfigChanged(member, updatedMember);
+ }
}
return false;
}
@@ -4108,9 +4293,17 @@ public class GroupMetadataManager {
.withMetadataImage(metadataImage)
.withTargetAssignment(group.targetAssignment());
- updatedMember.ifPresent(member ->
- assignmentResultBuilder.addOrUpdateMember(member.memberId(), member)
- );
+ updatedMember.ifPresent(member -> {
+ assignmentResultBuilder.addOrUpdateMember(member.memberId(), member);
+ // If the instance id was associated to a different member, it means that the
+ // static member is replaced by the current member hence we remove the previous one.
+ member.instanceId().ifPresent(instanceId -> {
+ StreamsGroupMember previousMember = group.staticMember(instanceId);
+ if (previousMember != null && !member.memberId().equals(previousMember.memberId())) {
+ assignmentResultBuilder.removeMember(previousMember.memberId());
+ }
+ });
+ });
long startTimeMs = time.milliseconds();
org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult assignmentResult =
@@ -4256,7 +4449,18 @@ public class GroupMetadataManager {
log.info("[GroupId {}][MemberId {}] Member {} left the streams group.", groupId, memberId, memberId);
return streamsGroupFenceMember(group, member, new StreamsGroupHeartbeatResult(response, Map.of(), group.currentTopologyEpoch()));
} else {
- throw new UnsupportedOperationException("Static members are not supported in streams groups.");
+ StreamsGroupMember member = group.staticMember(instanceId);
+ throwIfStaticMemberIsUnknown(member, instanceId);
+ throwIfInstanceIdIsFenced(member, groupId, memberId, instanceId);
+ if (memberEpoch == LEAVE_GROUP_STATIC_MEMBER_EPOCH) {
+ log.info("[GroupId {}][MemberId {}] Static member {} with instance id {} temporarily left the streams group.",
+ group.groupId(), memberId, memberId, instanceId);
+ return streamsGroupStaticMemberGroupLeave(group, member);
+ } else {
+ log.info("[GroupId {}][MemberId {}] Static member {} with instance id {} left the streams group.",
+ group.groupId(), memberId, memberId, instanceId);
+ return streamsGroupFenceMember(group, member, new StreamsGroupHeartbeatResult(response, Map.of(), group.currentTopologyEpoch()));
+ }
}
}
@@ -4293,6 +4497,41 @@ 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
+ * 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.
+ *
+ * @param group The group.
+ * @param member The static member in the group for the instance id.
+ *
+ * @return A CoordinatorResult with a single record signifying that the static member is leaving.
+ */
+ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupStaticMemberGroupLeave(
+ StreamsGroup group,
+ StreamsGroupMember member
+ ) {
+ // TODO: https://issues.apache.org/jira/browse/KAFKA-20680
+ StreamsGroupMember leavingStaticMember = new StreamsGroupMember.Builder(member)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .resetAssignedTasksEpochsToZero()
+ .build();
+
+ CoordinatorRecord record = newStreamsGroupCurrentAssignmentRecord(group.groupId(), leavingStaticMember);
+ StreamsGroupHeartbeatResponseData response = new StreamsGroupHeartbeatResponseData()
+ .setMemberId(member.memberId())
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setStatus(List.of());
+
+ return new CoordinatorResult<>(
+ List.of(record),
+ new StreamsGroupHeartbeatResult(response, Map.of(), group.currentTopologyEpoch())
+ );
+ }
+
/**
* Handles leave request from a share group member.
* @param groupId The group id from the request.
@@ -4502,6 +4741,42 @@ public class GroupMetadataManager {
));
}
+ /**
+ * Write records to replace the old member by the new member.
+ *
+ * @param records The list of records to append to.
+ * @param group The streams group.
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ */
+ private void replaceStreamsMember(
+ List<CoordinatorRecord> records,
+ StreamsGroup group,
+ StreamsGroupMember oldMember,
+ StreamsGroupMember newMember
+ ) {
+ String groupId = group.groupId();
+
+ // Remove the member without canceling its timers in case the change is reverted. If the
+ // change is not reverted, the group validation will fail and the timer will do nothing.
+ records.addAll(removeStreamsMember(groupId, oldMember.memberId()));
+
+ // Generate records.
+ records.add(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(
+ groupId,
+ newMember
+ ));
+ records.add(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord(
+ groupId,
+ newMember.memberId(),
+ group.targetAssignment(oldMember.memberId(), Optional.empty())
+ ));
+ records.add(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(
+ groupId,
+ newMember
+ ));
+ }
+
/**
* Fences a member from a streams group.
*
@@ -9026,6 +9301,35 @@ public class GroupMetadataManager {
));
}
+ private static boolean hasUserEndpointChanged(StreamsGroupMember maybeOldMember, StreamsGroupMember updatedMember) {
+ boolean hasPreviousUserEndpoint = maybeOldMember != null && maybeOldMember.userEndpoint().isPresent();
+ boolean hasCurrentUserEndpoint = updatedMember.userEndpoint().isPresent();
+
+ // Both the previous and updated member have a user endpoint. This can happen for
+ // regular heartbeats after join, or for static member rejoins with an endpoint.
+ if (hasPreviousUserEndpoint && hasCurrentUserEndpoint) {
+ return !maybeOldMember.userEndpoint().get().equals(updatedMember.userEndpoint().get());
+ }
+
+ if (!hasPreviousUserEndpoint && !hasCurrentUserEndpoint) {
+ return false;
+ }
+
+ return true;
+ }
+
+ private static boolean hasEpochRelevantMemberConfigChanged(
+ StreamsGroupMember oldMember,
+ StreamsGroupMember newMember
+ ) {
+ // The group epoch is bumped: (KIP-1071)
+ // - When a member updates its topology metadata, rack ID, client tags or process ID.
+ return !Objects.equals(oldMember.topologyEpoch(), newMember.topologyEpoch())
+ || !Objects.equals(oldMember.rackId(), newMember.rackId())
+ || !Objects.equals(oldMember.clientTags(), newMember.clientTags())
+ || !Objects.equals(oldMember.processId(), newMember.processId());
+ }
+
/**
* Generate a classic group heartbeat key for the timer.
*
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
index 5f2878bba43..39caa8ddb9d 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
@@ -567,12 +567,40 @@ public class StreamsGroup implements Group {
}
/**
- * Returns the target assignment of the member.
+ * Returns true if the static member exists.
+ *
+ * @param instanceId The instance id.
*
- * @return The StreamsGroupMemberAssignment or an EMPTY one if it does not exist.
+ * @return A boolean indicating whether the member exists or not.
*/
- public TasksTuple targetAssignment(String memberId) {
- return targetAssignment.getOrDefault(memberId, TasksTuple.EMPTY);
+ public boolean hasStaticMember(String instanceId) {
+ if (instanceId == null) return false;
+ return staticMembers.containsKey(instanceId);
+ }
+
+ /**
+ * Returns the target assignment of the member.
+ * <p>
+ * If {@code instanceId} is empty, the assignment is looked up by {@code memberId}.
+ * If {@code instanceId} is present, the assignment is looked up by the member ID
+ * 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}
+ * if no assignment exists or no static member exists for {@code instanceId}.
+ */
+ public TasksTuple targetAssignment(String memberId, Optional<String> instanceId) {
+ if (instanceId.isEmpty()) {
+ return targetAssignment.getOrDefault(memberId, TasksTuple.EMPTY);
+ } else {
+ StreamsGroupMember previousMember = staticMember(instanceId.get());
+ if (previousMember != null) {
+ return targetAssignment.getOrDefault(previousMember.memberId(), TasksTuple.EMPTY);
+ }
+ }
+ return TasksTuple.EMPTY;
}
/**
@@ -1274,11 +1302,13 @@ public class StreamsGroup implements Group {
*
* @param updatedMember The member that was just updated (may have a stale entry in the members map).
* @param metadataImage The current metadata image for resolving topic partitions.
+ * @param maybeReplacedStaticMember The replaced static member. it can be null.
* @return The list of endpoint-to-partitions mappings for all members with endpoints.
*/
public List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> buildEndpointToPartitions(
StreamsGroupMember updatedMember,
- CoordinatorMetadataImage metadataImage
+ CoordinatorMetadataImage metadataImage,
+ StreamsGroupMember maybeReplacedStaticMember
) {
List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> endpointToPartitionsList = new ArrayList<>();
if (updatedMember == null) {
@@ -1290,6 +1320,9 @@ public class StreamsGroup implements Group {
if (entry.getKey().equals(updatedMember.memberId())) {
continue;
}
+ if (maybeReplacedStaticMember != null && entry.getKey().equals(maybeReplacedStaticMember.memberId())) {
+ continue;
+ }
getOrComputeEndpointToPartitions(entry.getValue(), metadataImage)
.ifPresent(endpointToPartitionsList::add);
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java
index a43431ce47e..1c042778bbe 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java
@@ -24,6 +24,7 @@ import org.slf4j.Logger;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -104,9 +105,13 @@ public record StreamsGroupMember(String memberId,
}
public Builder(StreamsGroupMember member) {
+ this(Objects.requireNonNull(member, "member cannot be null"), member.memberId);
+ }
+
+ public Builder(StreamsGroupMember member, String memberId) {
Objects.requireNonNull(member, "member cannot be null");
- this.memberId = member.memberId;
+ this.memberId = memberId;
this.memberEpoch = member.memberEpoch;
this.previousMemberEpoch = member.previousMemberEpoch;
this.instanceId = member.instanceId;
@@ -315,6 +320,37 @@ public record StreamsGroupMember(String memberId,
tasksPendingRevocation
);
}
+
+ /**
+ * Resets the assignment epochs to 0 for all assigned active tasks.
+ * Used when a static member leaves, so that the rejoining member's
+ * active tasks will be assigned from epoch 0 to the new member ID.
+ * All commits using the old member ID will be fenced.
+ */
+ public Builder resetAssignedTasksEpochsToZero() {
+ if (this.assignedTasks.isEmpty()) {
+ return this;
+ }
+
+ if (this.assignedTasks.activeTasksWithEpochs().isEmpty()) {
+ return this;
+ }
+
+ Map<String, Map<Integer, Integer>> resetActiveTasks = new HashMap<>();
+ for (Map.Entry<String, Map<Integer, Integer>> entry : this.assignedTasks.activeTasksWithEpochs().entrySet()) {
+ Map<Integer, Integer> resetActiveTaskEpochs = new HashMap<>();
+ for (Integer partitionId : entry.getValue().keySet()) {
+ resetActiveTaskEpochs.put(partitionId, 0);
+ }
+ resetActiveTasks.put(entry.getKey(), resetActiveTaskEpochs);
+ }
+ this.assignedTasks = new TasksTupleWithEpochs(
+ resetActiveTasks,
+ this.assignedTasks.standbyTasks(),
+ this.assignedTasks.warmupTasks()
+ );
+ return this;
+ }
}
/**
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
index a5061023788..23c12223af0 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
@@ -591,21 +591,6 @@ public class GroupCoordinatorServiceTest {
AuthorizableRequestContext context = mock(AuthorizableRequestContext.class);
when(context.requestVersion()).thenReturn((int) ApiKeys.STREAMS_GROUP_HEARTBEAT.latestVersion());
- assertEquals(
- new StreamsGroupHeartbeatResult(
- new StreamsGroupHeartbeatResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage("Static membership is not yet supported."),
- Map.of(),
- -1
- ),
- service.streamsGroupHeartbeat(
- context,
- new StreamsGroupHeartbeatRequestData()
- .setInstanceId(Uuid.randomUuid().toString())
- ).get(5, TimeUnit.SECONDS)
- );
-
assertEquals(
new StreamsGroupHeartbeatResult(
new StreamsGroupHeartbeatResponseData()
@@ -2188,7 +2173,7 @@ public class GroupCoordinatorServiceTest {
ArgumentMatchers.eq("streams-group-describe"),
ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
ArgumentMatchers.any()
- )).thenReturn(CompletableFuture.completedFuture(new StreamsGroupDescribeResult(List.of(describedGroup), Map.of())));
+ )).thenReturn(CompletableFuture.completedFuture(List.of(describedGroup)));
CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>> future =
service.streamsGroupDescribe(requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), Arrays.asList("", null));
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index 7500ad7e711..9c1d3a2b2bc 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -22710,9 +22710,9 @@ public class GroupMetadataManagerTest {
TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
);
context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord("foo", "m1", tasks));
- assertEquals(tasks.activeTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").activeTasks());
- assertEquals(tasks.standbyTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").standbyTasks());
- assertEquals(tasks.warmupTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").warmupTasks());
+ assertEquals(tasks.activeTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1", Optional.empty()).activeTasks());
+ assertEquals(tasks.standbyTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1", Optional.empty()).standbyTasks());
+ assertEquals(tasks.warmupTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1", Optional.empty()).warmupTasks());
}
@Test
@@ -22743,7 +22743,7 @@ public class GroupMetadataManagerTest {
context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord("foo", "m1"));
- assertTrue(context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").isEmpty());
+ assertTrue(context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1", Optional.empty()).isEmpty());
}
@Test
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
index a9f4c2998b0..6c81bcb03ec 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
@@ -734,24 +734,28 @@ public class GroupMetadataManagerTestContext {
}
public CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupHeartbeat(
- StreamsGroupHeartbeatRequestData request
+ StreamsGroupHeartbeatRequestData request,
+ String clientId,
+ InetAddress clientAddress
) {
- return streamsGroupHeartbeat(request, ApiKeys.STREAMS_GROUP_HEARTBEAT.latestVersion());
+ return streamsGroupHeartbeat(request, clientId, clientAddress, ApiKeys.STREAMS_GROUP_HEARTBEAT.latestVersion());
}
public CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupHeartbeat(
StreamsGroupHeartbeatRequestData request,
+ String clientId,
+ InetAddress clientAddress,
short version
) {
RequestContext context = new RequestContext(
new RequestHeader(
ApiKeys.STREAMS_GROUP_HEARTBEAT,
version,
- "client",
+ clientId,
0
),
"1",
- InetAddress.getLoopbackAddress(),
+ clientAddress,
KafkaPrincipal.ANONYMOUS,
ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT),
SecurityProtocol.PLAINTEXT,
@@ -770,6 +774,19 @@ public class GroupMetadataManagerTestContext {
return result;
}
+ public CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupHeartbeat(
+ StreamsGroupHeartbeatRequestData request
+ ) {
+ return streamsGroupHeartbeat(request, "client", InetAddress.getLoopbackAddress());
+ }
+
+ public CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupHeartbeat(
+ StreamsGroupHeartbeatRequestData request,
+ short version
+ ) {
+ return streamsGroupHeartbeat(request, "client", InetAddress.getLoopbackAddress(), version);
+ }
+
public List<MockCoordinatorTimer.ExpiredTimeout<CoordinatorRecord>> sleep(long ms) {
time.sleep(ms);
List<MockCoordinatorTimer.ExpiredTimeout<CoordinatorRecord>> timeouts = timer.poll();
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupMixedGroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupMixedGroupMetadataManagerTest.java
new file mode 100644
index 00000000000..b709426710b
--- /dev/null
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupMixedGroupMetadataManagerTest.java
@@ -0,0 +1,899 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.FencedInstanceIdException;
+import org.apache.kafka.common.errors.GroupMaxSizeReachedException;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorResult;
+import org.apache.kafka.coordinator.common.runtime.MetadataImageBuilder;
+import org.apache.kafka.coordinator.common.runtime.MockCoordinatorTimer;
+import org.apache.kafka.coordinator.group.streams.MockTaskAssignor;
+import org.apache.kafka.coordinator.group.streams.StreamsCoordinatorRecordHelpers;
+import org.apache.kafka.coordinator.group.streams.StreamsGroup;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupBuilder;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupMember;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil;
+import org.apache.kafka.coordinator.group.streams.TasksTuple;
+import org.apache.kafka.coordinator.group.streams.TasksTupleWithEpochs;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+import static org.apache.kafka.coordinator.group.Assertions.assertRecordsEquals;
+import static org.apache.kafka.coordinator.group.Assertions.assertResponseEquals;
+import static org.apache.kafka.coordinator.group.Assertions.assertUnorderedRecordsEquals;
+import static org.apache.kafka.coordinator.group.GroupMetadataManager.groupSessionTimeoutKey;
+import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.getDefaultAssignmentConfigs;
+import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.mkResponseTasks;
+import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.staticHeartbeat;
+import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.staticJoinHeartbeat;
+import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.streamsGroupMemberBuilderWithDefaults;
+import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.streamsTopicFixture;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.TaskRole;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTasksTupleWithEpochs;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTasksWithEpochs;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class StreamsGroupMixedGroupMetadataManagerTest {
+
+ private static final int DEFAULT_GROUP_EPOCH = 10;
+
+ @Test
+ public void testDynamicJoinFailsAtMaxSizeWhileStaticMemberIsTemporarilyLeftAndDynamicMemberStillExists() {
+ // STREAMS_GROUP_MAX_SIZE_CONFIG is 2.
+ // There are 2 members. (1 dynamic member, 1 static member)
+ // - one static member leaves temporarily with -2 epoch.
+ // - one dynamic member alive in group.
+ // Another dynamic member try to join.
+
+ String groupId = "fooup";
+ String staticMemberId = Uuid.randomUuid().toString();
+ String staticInstanceId = Uuid.randomUuid().toString();
+ String dynamicMemberId = Uuid.randomUuid().toString();
+ String newDynamicMemberId = Uuid.randomUuid().toString();
+ StreamsGroupHeartbeatRequestData.Topology topology = new StreamsGroupHeartbeatRequestData.Topology().setSubtopologies(List.of());
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withMetadataImage(new MetadataImageBuilder().buildCoordinatorMetadataImage())
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, 2)
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, 10)
+ .withMember(streamsGroupMemberBuilderWithDefaults(staticMemberId, staticInstanceId)
+ .setMemberEpoch(StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(10)
+ .build())
+ .withMember(streamsGroupMemberBuilderWithDefaults(dynamicMemberId)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .build())
+ .withTargetAssignmentEpoch(10)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topology)))
+ .build();
+
+ assertThrows(GroupMaxSizeReachedException.class, () ->
+ context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, newDynamicMemberId, null, "new-process-id")
+ .setRebalanceTimeoutMs(1500)
+ .setTopology(topology)
+ )
+ );
+ }
+
+ @Test
+ public void testStaticRejoinSucceedsAtMaxSizeWhileDynamicMemberStillExists() {
+ // Scenario
+ // STREAMS_GROUP_MAX_SIZE_CONFIG is 2.
+ // There are 2 members. (1 dynamic member, 1 static member)
+ // - static member leaves temporarily with -2 epoch.
+ // - dynamic member alive in group
+ // static member try to rejoin.
+
+ int groupEpoch = 10;
+ String groupId = "fooup";
+
+ String subtopologyId = "subtopology-1";
+ StreamsGroupTestUtil.StreamsTopicFixture topic = streamsTopicFixture(subtopologyId, "foo", 4);
+
+ String oldStaticMemberId = Uuid.randomUuid().toString();
+ String newStaticMemberId = Uuid.randomUuid().toString();
+ String staticInstanceId = Uuid.randomUuid().toString();
+ String dynamicMemberId = Uuid.randomUuid().toString();
+
+ TasksTupleWithEpochs staticAssignedTasks = topic.assignedTasks(groupEpoch, 0, 1);
+ TasksTuple staticTargetAssignment = topic.targetAssignment(0, 1);
+
+ TasksTupleWithEpochs dynamicAssignedTasks = topic.assignedTasks(groupEpoch, 2, 3);
+ TasksTuple dynamicTargetAssignment = topic.targetAssignment(2, 3);
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, 2)
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(oldStaticMemberId, staticInstanceId)
+ .setMemberEpoch(StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(10)
+ .setAssignedTasks(staticAssignedTasks)
+ .build())
+ .withMember(streamsGroupMemberBuilderWithDefaults(dynamicMemberId)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setAssignedTasks(dynamicAssignedTasks)
+ .build())
+ .withTargetAssignment(oldStaticMemberId, staticTargetAssignment)
+ .withTargetAssignment(dynamicMemberId, dynamicTargetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .build();
+
+
+ // static member rejoins.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> rejoinResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, newStaticMemberId, staticInstanceId, StreamsGroupHeartbeatRequest.JOIN_GROUP_MEMBER_EPOCH)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(newStaticMemberId)
+ .setMemberEpoch(groupEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(mkResponseTasks(subtopologyId, 0, 1))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ rejoinResult.response().data()
+ );
+
+ StreamsGroup group = context.groupMetadataManager.streamsGroup(groupId);
+ assertFalse(group.hasMember(oldStaticMemberId));
+ assertTrue(group.hasMember(newStaticMemberId));
+ assertEquals(newStaticMemberId, group.staticMember(staticInstanceId).memberId());
+
+ assertTrue(group.hasMember(dynamicMemberId));
+ assertEquals(dynamicAssignedTasks, group.getMemberOrThrow(dynamicMemberId).assignedTasks());
+ assertEquals(dynamicTargetAssignment, group.targetAssignment(dynamicMemberId, Optional.empty()));
+ assertEquals(groupEpoch, group.groupEpoch());
+ }
+
+ @Test
+ public void testDynamicJoinSucceedsAfterTemporarilyLeftStaticMemberSessionTimeoutInMixedGroup() {
+ // Scenario:
+ // STREAMS_GROUP_MAX_SIZE_CONFIG is 2.
+ // There are 2 members. (1 dynamic member, 1 static member)
+ // - static member leaves temporarily with -2 epoch.
+ // - dynamic member alive in group.
+ // After the static member session timeout expires, a new dynamic member can join.
+
+ int groupEpoch = 10;
+ int timeoutGroupEpoch = groupEpoch + 1;
+ int joinGroupEpoch = timeoutGroupEpoch + 1;
+
+ String groupId = "fooup";
+
+ String subtopologyId = "subtopology-1";
+ StreamsGroupTestUtil.StreamsTopicFixture topic = streamsTopicFixture(subtopologyId, "foo", 4);
+ StreamsGroupHeartbeatRequestData.Topology topology = topic.topology();
+
+ String staticMemberId = Uuid.randomUuid().toString();
+ String staticInstanceId = Uuid.randomUuid().toString();
+ String dynamicMemberId = Uuid.randomUuid().toString();
+ String newDynamicMemberId = Uuid.randomUuid().toString();
+
+ String staticProcessId = "static-process-id";
+ String dynamicProcessId = "dynamic-process-id";
+ String newDynamicProcessId = "new-dynamic-process-id";
+
+ TasksTupleWithEpochs staticAssignedTasks = topic.assignedTasks(groupEpoch, 0, 1);
+ TasksTuple staticTargetAssignment = topic.targetAssignment(0, 1);
+
+ TasksTupleWithEpochs dynamicAssignedTasks = topic.assignedTasks(groupEpoch, 2, 3);
+ TasksTuple dynamicTargetAssignment = topic.targetAssignment(2, 3);
+
+ long groupMetadataHash = topic.metadataHash();
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(topic.metadataImage())
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, 2)
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(staticMemberId, staticInstanceId)
+ .setProcessId(staticProcessId)
+ .setMemberEpoch(groupEpoch)
+ .setPreviousMemberEpoch(groupEpoch - 1)
+ .setAssignedTasks(staticAssignedTasks)
+ .build())
+ .withMember(streamsGroupMemberBuilderWithDefaults(dynamicMemberId)
+ .setProcessId(dynamicProcessId)
+ .setMemberEpoch(groupEpoch)
+ .setPreviousMemberEpoch(groupEpoch - 1)
+ .setAssignedTasks(dynamicAssignedTasks)
+ .build())
+ .withTargetAssignment(staticMemberId, staticTargetAssignment)
+ .withTargetAssignment(dynamicMemberId, dynamicTargetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topology))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(groupMetadataHash)
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .build();
+ context.onLoaded();
+
+ // static member leaves with epoch -2.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> leaveResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, staticMemberId, staticInstanceId, StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(staticMemberId)
+ .setMemberEpoch(StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setStatus(List.of())
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null),
+ leaveResult.response().data()
+ );
+
+ // To prevent session timeout from dynamic member.
+ // Sleep 1, and dynamic member send a heartbeat.
+ GroupMetadataManagerTestContext.assertNoOrEmptyResult(context.sleep(1));
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> dynamicHeartbeatResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, dynamicMemberId, null, groupEpoch)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(dynamicMemberId)
+ .setMemberEpoch(groupEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null),
+ dynamicHeartbeatResult.response().data());
+ assertTrue(dynamicHeartbeatResult.records().isEmpty());
+
+ // static member session timeout.
+ context.assertSessionTimeout(groupId, staticMemberId, 45000 - 1);
+ List<MockCoordinatorTimer.ExpiredTimeout<CoordinatorRecord>> timeouts = context.sleep(45000 - 1);
+
+ List<CoordinatorRecord> expectedTimeoutRecords = List.of(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentTombstoneRecord(groupId, staticMemberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord(groupId, staticMemberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord(groupId, staticMemberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ groupId, timeoutGroupEpoch, groupMetadataHash, 0, getDefaultAssignmentConfigs(), -1, -1
+ )
+ );
+
+ assertEquals(
+ List.of(new MockCoordinatorTimer.ExpiredTimeout<>(
+ groupSessionTimeoutKey(groupId, staticMemberId),
+ new CoordinatorResult<>(expectedTimeoutRecords)
+ )),
+ timeouts
+ );
+
+ StreamsGroup group = context.groupMetadataManager.streamsGroup(groupId);
+ assertFalse(group.hasMember(staticMemberId));
+ assertTrue(group.hasMember(dynamicMemberId));
+ assertEquals(timeoutGroupEpoch, group.groupEpoch());
+
+ assignor.prepareGroupAssignment(Map.of(
+ dynamicMemberId, dynamicTargetAssignment,
+ newDynamicMemberId, staticTargetAssignment
+ ));
+
+ // new dynamic member try to join.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> joinResult = context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, newDynamicMemberId, null, newDynamicProcessId)
+ .setRebalanceTimeoutMs(1500)
+ .setTopology(topology)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(newDynamicMemberId)
+ .setMemberEpoch(joinGroupEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(mkResponseTasks(subtopologyId, 0, 1))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ joinResult.response().data()
+ );
+
+ StreamsGroupMember expectedJoiningDynamicMember = streamsGroupMemberBuilderWithDefaults(newDynamicMemberId)
+ .setProcessId(newDynamicProcessId)
+ .setMemberEpoch(0)
+ .setPreviousMemberEpoch(0)
+ .build();
+
+ StreamsGroupMember expectedReconciledDynamicMember = streamsGroupMemberBuilderWithDefaults(newDynamicMemberId)
+ .setProcessId(newDynamicProcessId)
+ .setMemberEpoch(joinGroupEpoch)
+ .setPreviousMemberEpoch(0)
+ .setAssignedTasks(mkTasksTupleWithEpochs(
+ TaskRole.ACTIVE,
+ mkTasksWithEpochs(subtopologyId, Map.of(
+ 0, joinGroupEpoch,
+ 1, joinGroupEpoch
+ ))
+ ))
+ .build();
+
+ List<CoordinatorRecord> expectedJoinRecords = List.of(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(groupId, expectedJoiningDynamicMember),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ groupId,
+ joinGroupEpoch,
+ groupMetadataHash,
+ 0,
+ getDefaultAssignmentConfigs(),
+ -1,
+ -1
+ ),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord(groupId, newDynamicMemberId, staticTargetAssignment),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentMetadataRecord(groupId, joinGroupEpoch, context.time.milliseconds()),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, expectedReconciledDynamicMember)
+ );
+
+ assertRecordsEquals(expectedJoinRecords, joinResult.records());
+
+ assertTrue(group.hasMember(dynamicMemberId));
+ assertTrue(group.hasMember(newDynamicMemberId));
+ assertEquals(joinGroupEpoch, group.groupEpoch());
+ assertEquals(staticTargetAssignment, group.targetAssignment(newDynamicMemberId, Optional.empty()));
+ assertEquals(dynamicTargetAssignment, group.targetAssignment(dynamicMemberId, Optional.empty()));
+ }
+
+ @Test
+ public void testStaticTemporaryLeaveDoesNotTransferTasksToExistingDynamicMember() {
+ // Scenario
+ // STREAMS_GROUP_MAX_SIZE_CONFIG is 2.
+ // There are 2 members. (1 dynamic member, 1 static member)
+ // - static member leaves temporarily with -2 epoch.
+ // - dynamic member alive in group
+ // If dynamic send a heartbeat, there is no new assignment.
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+ String groupId = "fooup";
+ String staticMemberId = Uuid.randomUuid().toString();
+ String staticInstanceId = Uuid.randomUuid().toString();
+ String dynamicMemberId = Uuid.randomUuid().toString();
+
+
+ StreamsGroupTestUtil.StreamsTopicFixture topic = streamsTopicFixture("subtopology-1", "foo", 3);
+
+ TasksTupleWithEpochs staticAssignedTasks = topic.assignedTasks(groupEpoch, 0, 1);
+ TasksTuple staticTargetAssignment = topic.targetAssignment(0, 1);
+
+ TasksTupleWithEpochs dynamicAssignedTasks = topic.assignedTasks(groupEpoch, 2);
+ TasksTuple dynamicTargetAssignment = topic.targetAssignment(2);
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(staticMemberId, staticInstanceId)
+ .setMemberEpoch(groupEpoch)
+ .setPreviousMemberEpoch(groupEpoch - 1)
+ .setAssignedTasks(staticAssignedTasks)
+ .build())
+ .withMember(streamsGroupMemberBuilderWithDefaults(dynamicMemberId)
+ .setMemberEpoch(groupEpoch)
+ .setPreviousMemberEpoch(groupEpoch - 1)
+ .setAssignedTasks(dynamicAssignedTasks)
+ .build())
+ .withTargetAssignment(staticMemberId, staticTargetAssignment)
+ .withTargetAssignment(dynamicMemberId, dynamicTargetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs())
+ )
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ // static member leaves with epoch -2.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> leaveResult =
+ context.streamsGroupHeartbeat(staticHeartbeat(
+ groupId,
+ staticMemberId,
+ staticInstanceId,
+ LEAVE_GROUP_STATIC_MEMBER_EPOCH
+ ));
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(staticMemberId)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setStatus(List.of())
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null),
+ leaveResult.response().data()
+ );
+
+ // dynamic member send a heartbeat.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> dynamicHeartbeatResult =
+ context.streamsGroupHeartbeat(
+ new StreamsGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(dynamicMemberId)
+ .setMemberEpoch(groupEpoch)
+ );
+
+ // There is no new assignment.
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(dynamicMemberId)
+ .setMemberEpoch(groupEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null),
+ dynamicHeartbeatResult.response().data());
+ assertTrue(dynamicHeartbeatResult.records().isEmpty());
+
+ StreamsGroup group = context.groupMetadataManager.streamsGroup(groupId);
+ assertEquals(dynamicAssignedTasks, group.getMemberOrThrow(dynamicMemberId).assignedTasks());
+ assertEquals(dynamicTargetAssignment, group.targetAssignment(dynamicMemberId, Optional.empty()));
+ assertEquals(groupEpoch, group.groupEpoch());
+ }
+
+ @Test
+ public void testStaticRejoinWithUpdatedProcessIdRecomputesTargetAssignmentAndDynamicMemberReconcilesInMixedGroup() {
+ // Scenario:
+ // There are 2 members.
+ // - static member left temporarily with -2 epoch.
+ // - dynamic member alive in group.
+ // When the same static instance rejoins with a different processId, the coordinator
+ // bumps the group epoch and recomputes the target assignment.
+ // The dynamic member keeps its current assignment until its next heartbeat.
+ // When the dynamic member sends the next heartbeat, it reconciles to the recomputed target assignment.
+
+ String groupId = "fooup";
+ int groupEpoch = 10;
+ int bumpedGroupEpoch = groupEpoch + 1;
+
+ String subtopologyId = "subtopology-1";
+ StreamsGroupTestUtil.StreamsTopicFixture topic = streamsTopicFixture(subtopologyId, "foo", 4);
+
+ String oldStaticMemberId = Uuid.randomUuid().toString();
+ String rejoinStaticMemberId = Uuid.randomUuid().toString();
+ String staticInstanceId = Uuid.randomUuid().toString();
+ String dynamicMemberId = Uuid.randomUuid().toString();
+
+ String oldProcessId = "old-process-id";
+ String newProcessId = "new-process-id";
+
+ // Initial assignment
+ TasksTupleWithEpochs staticAssignedTasks = topic.assignedTasks(groupEpoch, 0, 1);
+ TasksTuple oldStaticTargetAssignment = topic.targetAssignment(0, 1);
+
+ TasksTupleWithEpochs dynamicAssignedTasks = topic.assignedTasks(groupEpoch, 2, 3);
+ TasksTuple oldDynamicTargetAssignment = topic.targetAssignment(2, 3);
+
+ // Recomputed target assignment after static member rejoins with updated processId
+ TasksTuple newStaticTargetAssignment = topic.targetAssignment(0);
+ TasksTuple newDynamicTargetAssignment = topic.targetAssignment(1, 2, 3);
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ assignor.prepareGroupAssignment(Map.of(
+ rejoinStaticMemberId, newStaticTargetAssignment,
+ dynamicMemberId, newDynamicTargetAssignment
+ ));
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(oldStaticMemberId, staticInstanceId)
+ .setMemberEpoch(StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(groupEpoch)
+ .setProcessId(oldProcessId)
+ .setAssignedTasks(staticAssignedTasks)
+ .build())
+ .withMember(streamsGroupMemberBuilderWithDefaults(dynamicMemberId)
+ .setMemberEpoch(groupEpoch)
+ .setPreviousMemberEpoch(groupEpoch - 1)
+ .setAssignedTasks(dynamicAssignedTasks)
+ .build())
+ .withTargetAssignment(oldStaticMemberId, oldStaticTargetAssignment)
+ .withTargetAssignment(dynamicMemberId, oldDynamicTargetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ // static member try to rejoin with new process id.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> rejoinResult = context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, rejoinStaticMemberId, staticInstanceId, newProcessId)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(rejoinStaticMemberId)
+ .setMemberEpoch(bumpedGroupEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(0))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ rejoinResult.response().data()
+ );
+
+ StreamsGroup group = context.groupMetadataManager.streamsGroup(groupId);
+
+ // group epoch should be bumped up.
+ assertEquals(bumpedGroupEpoch, group.groupEpoch());
+
+ // static member should be replaced.
+ assertFalse(group.hasMember(oldStaticMemberId));
+ assertTrue(group.hasMember(rejoinStaticMemberId));
+ assertEquals(rejoinStaticMemberId, group.staticMember(staticInstanceId).memberId());
+
+ // Because group epoch is bumped up, target assignment should be recomputed.
+ assertEquals(newStaticTargetAssignment, group.targetAssignment(rejoinStaticMemberId, Optional.of(staticInstanceId)));
+ assertEquals(newDynamicTargetAssignment, group.targetAssignment(dynamicMemberId, Optional.empty()));
+ assertEquals(dynamicAssignedTasks, group.getMemberOrThrow(dynamicMemberId).assignedTasks());
+ assertEquals(groupEpoch, group.getMemberOrThrow(dynamicMemberId).memberEpoch());
+
+ StreamsGroupMember expectedCopiedStaticMember = streamsGroupMemberBuilderWithDefaults(rejoinStaticMemberId, staticInstanceId)
+ .setProcessId(oldProcessId)
+ .setMemberEpoch(0)
+ .setPreviousMemberEpoch(0)
+ .setAssignedTasks(staticAssignedTasks)
+ .build();
+
+ StreamsGroupMember expectedUpdatedStaticMember = streamsGroupMemberBuilderWithDefaults(rejoinStaticMemberId, staticInstanceId)
+ .setProcessId(newProcessId)
+ .setMemberEpoch(0)
+ .setPreviousMemberEpoch(0)
+ .setAssignedTasks(staticAssignedTasks)
+ .build();
+
+ StreamsGroupMember expectedReconciledStaticMember = streamsGroupMemberBuilderWithDefaults(rejoinStaticMemberId, staticInstanceId)
+ .setProcessId(newProcessId)
+ .setMemberEpoch(bumpedGroupEpoch)
+ .setPreviousMemberEpoch(0)
+ .setAssignedTasks(mkTasksTupleWithEpochs(
+ TaskAssignmentTestUtil.TaskRole.ACTIVE,
+ mkTasksWithEpochs(subtopologyId, Map.of(0, groupEpoch))
+ ))
+ .build();
+
+ List<CoordinatorRecord> expectedRecordsBeforeRecomputedAssignments = List.of(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentTombstoneRecord(groupId, oldStaticMemberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord(groupId, oldStaticMemberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord(groupId, oldStaticMemberId),
+
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(groupId, expectedCopiedStaticMember),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord(groupId, rejoinStaticMemberId,
+ oldStaticTargetAssignment),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, expectedCopiedStaticMember),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(groupId, expectedUpdatedStaticMember),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ groupId,
+ bumpedGroupEpoch,
+ topic.metadataHash(),
+ 0,
+ getDefaultAssignmentConfigs(),
+ -1,
+ -1
+ )
+ );
+
+ List<CoordinatorRecord> expectedRecomputedTargetAssignmentRecords = List.of(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord(groupId, rejoinStaticMemberId,
+ newStaticTargetAssignment),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord(groupId, dynamicMemberId,
+ newDynamicTargetAssignment)
+ );
+
+ List<CoordinatorRecord> expectedRecordsAfterRecomputedAssignments = List.of(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentMetadataRecord(groupId, bumpedGroupEpoch,
+ context.time.milliseconds()),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, expectedReconciledStaticMember)
+ );
+
+ assertRecordsEquals(
+ expectedRecordsBeforeRecomputedAssignments,
+ rejoinResult.records().subList(0, 8)
+ );
+ assertUnorderedRecordsEquals(
+ List.of(expectedRecomputedTargetAssignmentRecords),
+ rejoinResult.records().subList(8, 10)
+ );
+ assertRecordsEquals(
+ expectedRecordsAfterRecomputedAssignments,
+ rejoinResult.records().subList(10, 12)
+ );
+
+ // dynamic member send a heartbeat and reconciles to the new target assignment.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> dynamicHeartbeatResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, dynamicMemberId, null, groupEpoch)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(dynamicMemberId)
+ .setMemberEpoch(bumpedGroupEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(1, 2, 3))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ dynamicHeartbeatResult.response().data()
+ );
+
+ StreamsGroupMember expectedUpdatedDynamicMember = streamsGroupMemberBuilderWithDefaults(dynamicMemberId)
+ .setMemberEpoch(bumpedGroupEpoch)
+ .setPreviousMemberEpoch(groupEpoch)
+ .setAssignedTasks(mkTasksTupleWithEpochs(
+ TaskAssignmentTestUtil.TaskRole.ACTIVE,
+ mkTasksWithEpochs(subtopologyId, Map.of(
+ 1, bumpedGroupEpoch,
+ 2, groupEpoch,
+ 3, groupEpoch
+ ))
+ ))
+ .build();
+
+ List<CoordinatorRecord> expectedRecordsAfterDynamicHeartbeat = List.of(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, expectedUpdatedDynamicMember)
+ );
+ assertRecordsEquals(expectedRecordsAfterDynamicHeartbeat, dynamicHeartbeatResult.records());
+
+ assertEquals(expectedUpdatedDynamicMember.assignedTasks(), group.getMemberOrThrow(dynamicMemberId).assignedTasks());
+ assertEquals(bumpedGroupEpoch, group.getMemberOrThrow(dynamicMemberId).memberEpoch());
+ assertEquals(newDynamicTargetAssignment, group.targetAssignment(dynamicMemberId, Optional.empty()));
+ }
+
+ @Test
+ public void testStaticRejoinWithSameProcessIdDoesNotBumpEpochAndDynamicHeartbeatRemainsNoOpInMixedGroup() {
+ // Scenario:
+ // There are 2 members.
+ // - static member left temporarily with -2 epoch.
+ // - dynamic member alive in group.
+ // When the same static instance rejoins with the same processId, the coordinator
+ // does not bump the group epoch or recompute the target assignment.
+ // When the dynamic member sends a heartbeat, there is still no new assignment.
+
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+ String groupId = "fooup";
+
+ String oldStaticMemberId = Uuid.randomUuid().toString();
+ String newStaticMemberId = Uuid.randomUuid().toString();
+ String staticInstanceId = Uuid.randomUuid().toString();
+ String dynamicMemberId = Uuid.randomUuid().toString();
+
+ String staticProcessId = "static-process-id";
+ String dynamicProcessId = "dynamic-process-id";
+
+ StreamsGroupTestUtil.StreamsTopicFixture topic = streamsTopicFixture("subtopology-1", "foo", 4);
+
+ TasksTupleWithEpochs staticAssignedTasks = topic.assignedTasks(groupEpoch, 0, 1);
+ TasksTuple staticTargetAssignment = topic.targetAssignment(0, 1);
+
+ TasksTupleWithEpochs dynamicAssignedTasks = topic.assignedTasks(groupEpoch, 2, 3);
+ TasksTuple dynamicTargetAssignment = topic.targetAssignment(2, 3);
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(oldStaticMemberId, staticInstanceId)
+ .setProcessId(staticProcessId)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(groupEpoch)
+ .setAssignedTasks(staticAssignedTasks)
+ .build())
+ .withMember(streamsGroupMemberBuilderWithDefaults(dynamicMemberId)
+ .setProcessId(dynamicProcessId)
+ .setMemberEpoch(groupEpoch)
+ .setPreviousMemberEpoch(groupEpoch - 1)
+ .setAssignedTasks(dynamicAssignedTasks)
+ .build())
+ .withTargetAssignment(oldStaticMemberId, staticTargetAssignment)
+ .withTargetAssignment(dynamicMemberId, dynamicTargetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ // static member try to rejoin with same process id.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> rejoinResult = context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, newStaticMemberId, staticInstanceId, staticProcessId)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(newStaticMemberId)
+ .setMemberEpoch(groupEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(0, 1))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ rejoinResult.response().data()
+ );
+
+ StreamsGroup group = context.groupMetadataManager.streamsGroup(groupId);
+
+ // no group epoch bump up.
+ assertEquals(groupEpoch, group.groupEpoch());
+ assertFalse(group.hasMember(oldStaticMemberId));
+ assertTrue(group.hasMember(newStaticMemberId));
+ assertTrue(group.hasMember(dynamicMemberId));
+ assertEquals(newStaticMemberId, group.staticMember(staticInstanceId).memberId());
+ assertEquals(staticTargetAssignment, group.targetAssignment(newStaticMemberId, Optional.of(staticInstanceId)));
+ assertEquals(dynamicTargetAssignment, group.targetAssignment(dynamicMemberId, Optional.empty()));
+
+ StreamsGroupMember expectedCopiedStaticMember = streamsGroupMemberBuilderWithDefaults(newStaticMemberId, staticInstanceId)
+ .setProcessId(staticProcessId)
+ .setMemberEpoch(0)
+ .setPreviousMemberEpoch(0)
+ .setAssignedTasks(staticAssignedTasks)
+ .build();
+
+ StreamsGroupMember expectedRejoinedStaticMember = streamsGroupMemberBuilderWithDefaults(newStaticMemberId, staticInstanceId)
+ .setProcessId(staticProcessId)
+ .setMemberEpoch(groupEpoch)
+ .setPreviousMemberEpoch(0)
+ .setAssignedTasks(staticAssignedTasks)
+ .build();
+
+ // no new target assignment.
+ List<CoordinatorRecord> expectedRejoinRecords = List.of(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentTombstoneRecord(groupId, oldStaticMemberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord(groupId, oldStaticMemberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord(groupId, oldStaticMemberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(groupId, expectedCopiedStaticMember),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord(groupId, newStaticMemberId, staticTargetAssignment),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, expectedCopiedStaticMember),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, expectedRejoinedStaticMember)
+ );
+
+ assertRecordsEquals(expectedRejoinRecords, rejoinResult.records());
+
+ // dynamic member send a heartbeat request
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> dynamicHeartbeatResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, dynamicMemberId, null, groupEpoch)
+ );
+
+ // no new target assignment.
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(dynamicMemberId)
+ .setMemberEpoch(groupEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null),
+ dynamicHeartbeatResult.response().data());
+ assertTrue(dynamicHeartbeatResult.records().isEmpty());
+
+ assertEquals(dynamicAssignedTasks, group.getMemberOrThrow(dynamicMemberId).assignedTasks());
+ assertEquals(groupEpoch, group.getMemberOrThrow(dynamicMemberId).memberEpoch());
+ assertEquals(dynamicTargetAssignment, group.targetAssignment(dynamicMemberId, Optional.empty()));
+ }
+
+ @Test
+ public void testOldStaticMemberIdIsFencedAfterReplacementInMixedGroup() {
+ // Scenario:
+ // There are 2 members.
+ // - static member left temporarily with -2 epoch.
+ // - dynamic member alive in group.
+ // When the static member rejoins with a new memberId, the old memberId is fenced.
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+ String groupId = "fooup";
+
+ String oldStaticMemberId = Uuid.randomUuid().toString();
+ String newStaticMemberId = Uuid.randomUuid().toString();
+ String staticInstanceId = Uuid.randomUuid().toString();
+ String dynamicMemberId = Uuid.randomUuid().toString();
+
+ String staticProcessId = "static-process-id";
+ String dynamicProcessId = "dynamic-process-id";
+
+ StreamsGroupTestUtil.StreamsTopicFixture topic = streamsTopicFixture("subtopology-1", "foo", 4);
+
+ TasksTupleWithEpochs staticAssignedTasks = topic.assignedTasks(groupEpoch, 0, 1);
+ TasksTuple staticTargetAssignment = topic.targetAssignment(0, 1);
+
+ TasksTupleWithEpochs dynamicAssignedTasks = topic.assignedTasks(groupEpoch, 2, 3);
+ TasksTuple dynamicTargetAssignment = topic.targetAssignment(2, 3);
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(oldStaticMemberId, staticInstanceId)
+ .setProcessId(staticProcessId)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(groupEpoch)
+ .setAssignedTasks(staticAssignedTasks)
+ .build())
+ .withMember(streamsGroupMemberBuilderWithDefaults(dynamicMemberId)
+ .setProcessId(dynamicProcessId)
+ .setMemberEpoch(groupEpoch)
+ .setPreviousMemberEpoch(groupEpoch - 1)
+ .setAssignedTasks(dynamicAssignedTasks)
+ .build())
+ .withTargetAssignment(oldStaticMemberId, staticTargetAssignment)
+ .withTargetAssignment(dynamicMemberId, dynamicTargetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ // static member try to rejoin with new member id.
+ context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, newStaticMemberId, staticInstanceId, staticProcessId)
+ );
+
+ // stale static member send a heartbeat with stale member id.
+ assertThrows(FencedInstanceIdException.class, () ->
+ context.streamsGroupHeartbeat(staticHeartbeat(groupId, oldStaticMemberId, staticInstanceId, groupEpoch))
+ );
+
+ StreamsGroup group = context.groupMetadataManager.streamsGroup(groupId);
+ assertFalse(group.hasMember(oldStaticMemberId));
+ assertTrue(group.hasMember(newStaticMemberId));
+ assertTrue(group.hasMember(dynamicMemberId));
+ assertEquals(newStaticMemberId, group.staticMember(staticInstanceId).memberId());
+ }
+}
\ No newline at end of file
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupStaticMemberGroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupStaticMemberGroupMetadataManagerTest.java
new file mode 100644
index 00000000000..073a613b5d5
--- /dev/null
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupStaticMemberGroupMetadataManagerTest.java
@@ -0,0 +1,1928 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.FencedInstanceIdException;
+import org.apache.kafka.common.errors.FencedMemberEpochException;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.errors.GroupMaxSizeReachedException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.errors.UnreleasedInstanceIdException;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorResult;
+import org.apache.kafka.coordinator.common.runtime.MetadataImageBuilder;
+import org.apache.kafka.coordinator.common.runtime.MockCoordinatorTimer;
+import org.apache.kafka.coordinator.group.StreamsGroupTestUtil.StreamsTopicFixture;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataKey;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupMetadataKey;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupMetadataValue;
+import org.apache.kafka.coordinator.group.streams.MemberState;
+import org.apache.kafka.coordinator.group.streams.MockTaskAssignor;
+import org.apache.kafka.coordinator.group.streams.StreamsCoordinatorRecordHelpers;
+import org.apache.kafka.coordinator.group.streams.StreamsGroup;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupBuilder;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupMember;
+import org.apache.kafka.coordinator.group.streams.StreamsTopology;
+import org.apache.kafka.coordinator.group.streams.TasksTuple;
+import org.apache.kafka.coordinator.group.streams.TasksTupleWithEpochs;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Stream;
+
+import static org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest.JOIN_GROUP_MEMBER_EPOCH;
+import static org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH;
+import static org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+import static org.apache.kafka.coordinator.group.Assertions.assertRecordsEquals;
+import static org.apache.kafka.coordinator.group.Assertions.assertResponseEquals;
+import static org.apache.kafka.coordinator.group.GroupMetadataManager.groupSessionTimeoutKey;
+import static org.apache.kafka.coordinator.group.GroupMetadataManagerTestContext.DEFAULT_CLIENT_ADDRESS;
+import static org.apache.kafka.coordinator.group.GroupMetadataManagerTestContext.DEFAULT_CLIENT_ID;
+import static org.apache.kafka.coordinator.group.GroupMetadataManagerTestContext.DEFAULT_PROCESS_ID;
+import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.getDefaultAssignmentConfigs;
+import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.resetAssignedTasksEpochsToZero;
+import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.staticHeartbeat;
+import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.staticJoinHeartbeat;
+import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.streamsGroupMemberBuilderWithDefaults;
+import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.streamsTopicFixture;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class StreamsGroupStaticMemberGroupMetadataManagerTest {
+
+ private static final int DEFAULT_MEMBER_EPOCH = 10;
+ private static final int DEFAULT_GROUP_EPOCH = 10;
+
+
+ @Test
+ public void testUnknownStaticMemberLeaveStreamsGroup() {
+ String groupId = "streams-group";
+
+ String activeMemberId = "active-member";
+ String activeInstanceId = "active-instance";
+
+ String unknownMemberId = "unknown-member-id";
+ String unknownInstanceId = "unknown-instance-id";
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 1);
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, DEFAULT_GROUP_EPOCH)
+ .withMember(streamsGroupMemberBuilderWithDefaults(activeMemberId, activeInstanceId)
+ .setMemberEpoch(DEFAULT_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1)
+ .build())
+ .withTargetAssignmentEpoch(DEFAULT_GROUP_EPOCH)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ assertThrows(
+ UnknownMemberIdException.class,
+ () -> context.streamsGroupHeartbeat(staticHeartbeat(
+ groupId,
+ unknownMemberId,
+ unknownInstanceId,
+ LEAVE_GROUP_STATIC_MEMBER_EPOCH
+ ))
+ );
+ }
+
+ @Test
+ public void testStreamsStaticJoinWithNewInstanceAtMaxSizeThrowsGroupMaxSizeReached() {
+ // With max.size=2 already reached,
+ // joining with a new static instanceId must throw GroupMaxSizeReachedException.
+ int streamsGroupMaxSize = 2;
+ String groupId = "streams-group";
+
+ String activeMemberId = "active-member";
+ String activeInstanceId = "active-instance";
+
+ String leftMemberId = "left-member";
+ String leftInstanceId = "left-instance";
+
+ String newMemberId = "new-member";
+ String newInstanceId = "new-instance";
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 1);
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ assignor.prepareGroupAssignment(Map.of(activeMemberId, TasksTuple.EMPTY));
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(topic.metadataImage())
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, streamsGroupMaxSize)
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, DEFAULT_GROUP_EPOCH)
+ .withMember(streamsGroupMemberBuilderWithDefaults(activeMemberId, activeInstanceId)
+ .setMemberEpoch(DEFAULT_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1)
+ .build())
+ .withMember(streamsGroupMemberBuilderWithDefaults(leftMemberId, leftInstanceId)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1)
+ .build())
+ .withTargetAssignmentEpoch(DEFAULT_GROUP_EPOCH)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withMetadataHash(topic.metadataHash())
+ .withValidatedTopologyEpoch(0)
+ )
+ .build();
+
+ assertThrows(
+ GroupMaxSizeReachedException.class,
+ () -> context.streamsGroupHeartbeat(staticJoinHeartbeat(
+ groupId,
+ newMemberId,
+ newInstanceId,
+ topic
+ ))
+ );
+
+ }
+
+ @Test
+ public void testStreamsStaticRejoinWithLeaveGroupStaticEpochAtMaxSizeSucceeds() {
+ // If a static member is in leave epoch (-2),
+ // rejoining with the same memberId/instanceId is allowed even at max size.
+ int streamsGroupMaxSize = 2;
+ String groupId = "streams-group";
+
+ String activeMemberId = "active-member";
+ String activeInstanceId = "active-instance";
+
+ String leftMemberId = "left-member";
+ String leftInstanceId = "left-instance";
+ String newJoinMemberId = "new-join-member";
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 1);
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ assignor.prepareGroupAssignment(Map.of(
+ activeMemberId, TasksTuple.EMPTY,
+ leftMemberId, TasksTuple.EMPTY
+ ));
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(topic.metadataImage())
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, streamsGroupMaxSize)
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, DEFAULT_GROUP_EPOCH)
+ .withMember(streamsGroupMemberBuilderWithDefaults(activeMemberId, activeInstanceId)
+ .setMemberEpoch(DEFAULT_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1)
+ .build())
+ .withMember(streamsGroupMemberBuilderWithDefaults(leftMemberId, leftInstanceId)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1)
+ .build())
+ .withTargetAssignmentEpoch(DEFAULT_GROUP_EPOCH)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withMetadataHash(topic.metadataHash())
+ .withValidatedTopologyEpoch(0)
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs())
+ )
+ .build();
+
+ assertDoesNotThrow(() -> context.streamsGroupHeartbeat(staticJoinHeartbeat(
+ groupId,
+ newJoinMemberId,
+ leftInstanceId,
+ topic
+ )));
+
+ }
+
+ @Test
+ public void testStreamsStaticJoinWithUnreleasedInstanceThrowsUnreleasedInstanceIdAtMaxSize() {
+ // If an active static member (epoch=10) still owns the instanceId,
+ // a different memberId joining with that instanceId must fail even at max size.
+ int streamsGroupMaxSize = 1;
+ String groupId = "streams-group";
+
+ String activeMemberId = "active-member";
+ String activeInstanceId = "active-instance";
+
+ String newMemberId = "new-member";
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 1);
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ assignor.prepareGroupAssignment(Map.of(activeMemberId, TasksTuple.EMPTY));
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(topic.metadataImage())
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, streamsGroupMaxSize)
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, DEFAULT_GROUP_EPOCH)
+ .withMember(streamsGroupMemberBuilderWithDefaults(activeMemberId, activeInstanceId)
+ .setMemberEpoch(DEFAULT_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1)
+ .build())
+ .withTargetAssignmentEpoch(DEFAULT_GROUP_EPOCH)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withMetadataHash(topic.metadataHash())
+ .withValidatedTopologyEpoch(0)
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs())
+ )
+ .build();
+
+ assertThrows(
+ UnreleasedInstanceIdException.class,
+ () -> context.streamsGroupHeartbeat(staticJoinHeartbeat(
+ groupId,
+ newMemberId,
+ activeInstanceId,
+ topic
+ ))
+ );
+ }
+
+ @ParameterizedTest
+ @MethodSource("staticMemberReusedInstanceErrorCases")
+ public void testStaticMemberSendHeartbeatWithVariousEpochThenThrowError(
+ int heartbeatMemberEpoch,
+ Class<? extends Exception> expectedException
+ ) {
+ // same instance id is reused with mismatched member identity/epoch.
+ String groupId = "streams-group";
+
+ String activeMemberId = "active-member";
+ String activeInstanceId = "active-instance";
+ String newMemberId = "new-member";
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 1);
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ assignor.prepareGroupAssignment(Map.of(activeMemberId, TasksTuple.EMPTY));
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(topic.metadataImage())
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, 2)
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, DEFAULT_GROUP_EPOCH)
+ .withMember(streamsGroupMemberBuilderWithDefaults(activeMemberId, activeInstanceId)
+ .setMemberEpoch(DEFAULT_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1)
+ .build())
+ .withTargetAssignmentEpoch(DEFAULT_GROUP_EPOCH)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withMetadataHash(topic.metadataHash())
+ .withValidatedTopologyEpoch(0)
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs())
+ )
+ .build();
+
+ assertThrows(
+ expectedException,
+ () -> context.streamsGroupHeartbeat(staticHeartbeat(
+ groupId,
+ newMemberId,
+ activeInstanceId,
+ heartbeatMemberEpoch
+ ))
+ );
+ }
+
+ private static Stream<Arguments> staticMemberReusedInstanceErrorCases() {
+ return Stream.of(
+ Arguments.of(0, UnreleasedInstanceIdException.class), // static member try to join when static member already existed, then throw UnreleasedInstanceIdException.
+ Arguments.of(1000, FencedInstanceIdException.class) // static member try to send bigger epoch when static member already existed, then throw FencedInstanceIdException.
+ );
+ }
+
+ @Test
+ public void testStaticMemberJoinThenRevokeAndReceiveTasks() {
+ int enoughMaxSize = 100;
+ testStaticMemberJoinThenRevokeAndReceiveTasksWith2Members(enoughMaxSize);
+ }
+
+ @Test
+ public void testStaticMemberJoinThenRevokeAndReceiveTasksInMaxSizeBoundary() {
+ int boundarySize = 2;
+ testStaticMemberJoinThenRevokeAndReceiveTasksWith2Members(boundarySize);
+ }
+
+ private void testStaticMemberJoinThenRevokeAndReceiveTasksWith2Members(int maxSize) {
+ String groupId = "fooup";
+
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+ String otherMemberId2 = Uuid.randomUuid().toString();
+
+ String instanceId1 = Uuid.randomUuid().toString();
+ String instanceId2 = Uuid.randomUuid().toString();
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4);
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, maxSize)
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, 10)
+ .withMember(StreamsGroupTestUtil.streamsGroupMemberBuilderWithDefaults(memberId1)
+ .setInstanceId(instanceId1)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setAssignedTasks(topic.assignedTasks(10, 0, 1, 2, 3))
+ .build())
+ .withTargetAssignment(memberId1, topic.targetAssignment(0, 1, 2, 3))
+ .withTargetAssignmentEpoch(10)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withMetadataHash(topic.metadataHash())
+ .withValidatedTopologyEpoch(0)
+ )
+ .build();
+
+ // Next target assignment after member2 joins.
+ assignor.prepareGroupAssignment(Map.of(
+ memberId1, topic.targetAssignment(0, 1),
+ memberId2, topic.targetAssignment(2, 3)
+ ));
+
+ // 1) Static member2 joins. It gets no active tasks yet because member1 still owns them.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> joinResult = context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, memberId2, instanceId2, topic)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId2)
+ .setMemberEpoch(11)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(List.of())
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ joinResult.response().data()
+ );
+
+ // 2) member1 receives revocation instruction: keep only [0,1].
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> revokeInstructionResult = context.streamsGroupHeartbeat(
+ new StreamsGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setInstanceId(instanceId1)
+ .setMemberId(memberId1)
+ .setMemberEpoch(10)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId1)
+ .setMemberEpoch(10)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(0, 1))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ revokeInstructionResult.response().data()
+ );
+
+ // 3) member1 acknowledges revocation by reporting owned active tasks [0,1].
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> revokeAckResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, memberId1, instanceId1, 10)
+ .setActiveTasks(List.of(
+ new StreamsGroupHeartbeatRequestData.TaskIds()
+ .setSubtopologyId("subtopology1")
+ .setPartitions(List.of(0, 1))
+ ))
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of())
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId1)
+ .setMemberEpoch(11)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setStatus(List.of()),
+ revokeAckResult.response().data()
+ );
+
+ // 4) member2 heartbeats again and now receives [2,3].
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> member2ReceiveResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, memberId2, instanceId2, 11)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId2)
+ .setMemberEpoch(11)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(2, 3))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ member2ReceiveResult.response().data()
+ );
+
+ // 5) member2 leave.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> member2LeaveResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, memberId2, instanceId2, LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setStatus(List.of())
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null)
+ .setHeartbeatIntervalMs(0),
+ member2LeaveResult.response().data()
+ );
+
+ // 6) member2 re-join with other memberId.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> member2rejoinResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, otherMemberId2, instanceId2, JOIN_GROUP_MEMBER_EPOCH)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(otherMemberId2)
+ .setMemberEpoch(11)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(2, 3))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ member2rejoinResult.response().data()
+ );
+ }
+
+ @Test
+ public void testStaticMemberRejoinWithUpdatedProcessIdBumpsStreamsGroupEpoch() {
+ String groupId = "fooup";
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+ int bumpedGroupEpoch = groupEpoch + 1;
+
+ String oldMemberId = Uuid.randomUuid().toString();
+ String rejoinMemberId = Uuid.randomUuid().toString();
+ String instanceId = Uuid.randomUuid().toString();
+
+ String oldProcessId = "old-process-id";
+ String newProcessId = "new-process-id";
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4);
+ TasksTupleWithEpochs assignedTasks = topic.assignedTasks(groupEpoch, 0, 1, 2, 3);
+ TasksTuple targetAssignment = topic.targetAssignment(0, 1, 2, 3);
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(oldMemberId, instanceId)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(groupEpoch)
+ .setProcessId(oldProcessId)
+ .setAssignedTasks(assignedTasks)
+ .build())
+ .withTargetAssignment(oldMemberId, targetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ assignor.prepareGroupAssignment(Map.of(rejoinMemberId, targetAssignment));
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, rejoinMemberId, instanceId, newProcessId)
+ );
+
+ assertEquals(rejoinMemberId, result.response().data().memberId());
+ assertEquals(bumpedGroupEpoch, result.response().data().memberEpoch());
+
+ CoordinatorRecord metadataRecord = result.records().stream()
+ .filter(record -> record.key() instanceof StreamsGroupMetadataKey)
+ .findFirst()
+ .orElse(null);
+
+ assertNotNull(metadataRecord, "Expected a StreamsGroupMetadata record when static member config changes.");
+ StreamsGroupMetadataValue metadataValue = (StreamsGroupMetadataValue) metadataRecord.value().message();
+ assertEquals(bumpedGroupEpoch, metadataValue.epoch());
+
+ assertTrue(result.records().contains(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ groupId,
+ bumpedGroupEpoch,
+ topic.metadataHash(),
+ 0,
+ getDefaultAssignmentConfigs(),
+ -1,
+ -1
+ )
+ ));
+ }
+
+ @Test
+ public void testStaticMemberRejoinWithSameProcessIdDoesNotBumpStreamsGroupEpoch() throws UnknownHostException {
+ String groupId = "fooup";
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+ String oldMemberId = Uuid.randomUuid().toString();
+ String rejoinMemberId = Uuid.randomUuid().toString();
+ String instanceId = Uuid.randomUuid().toString();
+
+ String processId = "process-id";
+ String newClientId = "new-client-id";
+ InetAddress newClientAddress = InetAddress.getByName("127.0.0.2");
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4);
+ TasksTupleWithEpochs assignedTasks = topic.assignedTasks(groupEpoch, 0, 1, 2, 3);
+ TasksTuple targetAssignment = topic.targetAssignment(0, 1, 2, 3);
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ assignor.prepareGroupAssignment(Map.of(rejoinMemberId, targetAssignment));
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(oldMemberId, instanceId)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(groupEpoch)
+ .setProcessId(processId)
+ .setAssignedTasks(assignedTasks)
+ .build())
+ .withTargetAssignment(oldMemberId, targetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, rejoinMemberId, instanceId, processId),
+ newClientId,
+ newClientAddress
+ );
+
+ assertEquals(rejoinMemberId, result.response().data().memberId());
+ assertEquals(groupEpoch, result.response().data().memberEpoch());
+
+ Optional<StreamsGroupMemberMetadataValue> updatedMemberMetadataValue = result.records().stream()
+ .filter(record -> record.key() instanceof StreamsGroupMemberMetadataKey)
+ .filter(record -> ((StreamsGroupMemberMetadataKey) record.key()).memberId().equals(rejoinMemberId))
+ .filter(record -> record.value() != null)
+ .map(record -> (StreamsGroupMemberMetadataValue) record.value().message())
+ .filter(value -> newClientId.equals(value.clientId()))
+ .filter(value -> newClientAddress.toString().equals(value.clientHost()))
+ .findFirst();
+
+ assertTrue(
+ updatedMemberMetadataValue.isPresent(),
+ "Expected a StreamsGroupMemberMetadata record with the updated client id/host."
+ );
+
+ assertTrue(
+ result.records().stream().noneMatch(record -> record.key() instanceof StreamsGroupMetadataKey),
+ "Expected no StreamsGroupMetadata record when only client id/host changes."
+ );
+
+ assertEquals(groupEpoch, result.response().data().memberEpoch());
+ }
+
+ @Test
+ public void testStaticMemberLeaveWithMinusOneFencesMemberAndBumpsStreamsGroupEpoch() {
+ String groupId = "fooup";
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+ int bumpedGroupEpoch = groupEpoch + 1;
+
+ String memberId = Uuid.randomUuid().toString();
+ String instanceId = Uuid.randomUuid().toString();
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4);
+ TasksTupleWithEpochs assignedTasks = topic.assignedTasks(groupEpoch, 0, 1, 2, 3);
+ TasksTuple targetAssignment = topic.targetAssignment(0, 1, 2, 3);
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(memberId, instanceId)
+ .setMemberEpoch(groupEpoch)
+ .setPreviousMemberEpoch(groupEpoch - 1)
+ .setAssignedTasks(assignedTasks)
+ .build())
+ .withTargetAssignment(memberId, targetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, memberId, instanceId, LEAVE_GROUP_MEMBER_EPOCH)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setStatus(List.of()),
+ result.response().data()
+ );
+
+ assertRecordsEquals(
+ List.of(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord(groupId, memberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord(groupId, memberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(groupId, bumpedGroupEpoch, topic.metadataHash(), 0, getDefaultAssignmentConfigs(), -1, -1),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentMetadataRecord(groupId, bumpedGroupEpoch, 0L)
+ ),
+ result.records()
+ );
+ }
+
+ @Test
+ public void testStaticMemberLeaveWithLeaveGroupStaticMemberEpoch() {
+ int leaveEpoch = LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+ int memberEpoch = DEFAULT_MEMBER_EPOCH;
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String instanceId = Uuid.randomUuid().toString();
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4);
+ TasksTupleWithEpochs assignedTasks = topic.assignedTasks(groupEpoch, 0, 1, 2, 3);
+ TasksTupleWithEpochs pendingRevocationTasks = topic.assignedTasks(groupEpoch, 2, 3);
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(memberId, instanceId)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setAssignedTasks(assignedTasks)
+ .setTasksPendingRevocation(pendingRevocationTasks)
+ .build())
+ .withTargetAssignment(memberId, topic.targetAssignment(0, 1, 2, 3))
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, memberId, instanceId, leaveEpoch)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId)
+ .setMemberEpoch(leaveEpoch)
+ .setStatus(List.of()),
+ result.response().data()
+ );
+
+ // No group epoch bump.
+ // Member epoch should be -2.
+ // task still remain.
+ // pendingRevocationTasks should be EMPTY.
+ StreamsGroupMember expectedMemberInResponse = streamsGroupMemberBuilderWithDefaults(memberId, instanceId)
+ .setMemberEpoch(leaveEpoch)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .setAssignedTasks(resetAssignedTasksEpochsToZero(assignedTasks))
+ .build();
+ assertRecordsEquals(
+ List.of(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, expectedMemberInResponse)),
+ result.records()
+ );
+ assertEquals(groupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+ }
+
+ @Test
+ public void testStaticMemberLeaveWithLeaveGroupStaticMemberEpochThenShouldBeIdempotence() {
+ int leaveEpoch = LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+ int memberEpoch = DEFAULT_MEMBER_EPOCH;
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String instanceId = Uuid.randomUuid().toString();
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4);
+ TasksTupleWithEpochs assignedTasks = topic.assignedTasks(groupEpoch, 0, 1, 2, 3);
+ TasksTuple targetAssignment = topic.targetAssignment(0, 1, 2, 3);
+
+ StreamsGroupMember alreadyLeftStaticMember = streamsGroupMemberBuilderWithDefaults(memberId, instanceId)
+ .setMemberEpoch(leaveEpoch)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setAssignedTasks(resetAssignedTasksEpochsToZero(assignedTasks))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(alreadyLeftStaticMember)
+ .withTargetAssignment(memberId, targetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(staticHeartbeat(groupId, memberId, instanceId, leaveEpoch));
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId)
+ .setMemberEpoch(leaveEpoch)
+ .setStatus(List.of()),
+ result.response().data()
+ );
+
+ assertRecordsEquals(
+ List.of(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, alreadyLeftStaticMember)),
+ result.records()
+ );
+ assertEquals(groupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+ }
+
+ @Test
+ public void testStaticMemberLeaveWithLeaveGroupStaticMemberEpochAndRejoinWithNewMemberId() {
+ String instanceId = Uuid.randomUuid().toString();
+ String oldMemberId = Uuid.randomUuid().toString();
+ String newMemberId = Uuid.randomUuid().toString();
+
+ verifyStaticMemberLeaveAndRejoinNoGroupBump(instanceId, oldMemberId, newMemberId);
+ }
+
+ @Test
+ public void testStaticMemberLeaveWithLeaveGroupStaticMemberEpochAndRejoinWithSameMemberId() {
+ String instanceId = Uuid.randomUuid().toString();
+ String memberId = Uuid.randomUuid().toString();
+
+ verifyStaticMemberLeaveAndRejoinNoGroupBump(instanceId, memberId, memberId);
+ }
+
+ private void verifyStaticMemberLeaveAndRejoinNoGroupBump(String instanceId, String oldMemberId, String newMemberId) {
+ /*
+ * Verifies:
+ * 1. leave(-2) does not bump group epoch.
+ * 2. rejoin restores member epoch and assignment.
+ * 3. replacement/tombstone records are written as expected.
+ */
+
+ int leaveEpoch = LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+ int memberEpoch = DEFAULT_MEMBER_EPOCH;
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+ String groupId = "fooup";
+
+ String subtopology1 = "subtopology1";
+ StreamsTopicFixture topic = streamsTopicFixture(subtopology1, "foo", 4);
+
+ TasksTupleWithEpochs givenAssignedTasks = topic.assignedTasks(memberEpoch, 0, 1, 2, 3);
+ TasksTuple givenTargetAssignment = topic.targetAssignment(0, 1, 2, 3);
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(oldMemberId, instanceId)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setAssignedTasks(givenAssignedTasks)
+ .build())
+ .withTargetAssignment(oldMemberId, givenTargetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> normalHeartbeatResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, oldMemberId, instanceId, memberEpoch)
+ );
+
+
+ // all tasks should be null because assigned tasks unchanged.
+ // Keep the group epoch.
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(oldMemberId)
+ .setMemberEpoch(memberEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null),
+ normalHeartbeatResult.response().data());
+ assertEquals(groupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+
+
+ // Stream member leaves with epoch -2.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> leaveResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, oldMemberId, instanceId, leaveEpoch)
+ );
+
+ // Keep the group epoch.
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(oldMemberId)
+ .setMemberEpoch(leaveEpoch)
+ .setStatus(List.of())
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null),
+ leaveResult.response().data()
+ );
+ assertEquals(groupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+
+ // Streams Member rejoin with other memberId
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> rejoinResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, newMemberId, instanceId, JOIN_GROUP_MEMBER_EPOCH)
+ );
+
+ // Inherit previous member's member epoch, and assigned tasks.
+ // Keep the member epoch bump.
+ // Keep the group epoch bump.
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(newMemberId)
+ .setMemberEpoch(memberEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(0, 1, 2, 3))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ rejoinResult.response().data()
+ );
+ assertEquals(groupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+
+ StreamsGroupMember newJoinStaticMember = streamsGroupMemberBuilderWithDefaults(newMemberId, instanceId)
+ .setMemberEpoch(JOIN_GROUP_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(JOIN_GROUP_MEMBER_EPOCH)
+ .setAssignedTasks(resetAssignedTasksEpochsToZero(givenAssignedTasks))
+ .build();
+
+ StreamsGroupMember withPrevMemberId = streamsGroupMemberBuilderWithDefaults(newMemberId, instanceId)
+ .setMemberEpoch(memberEpoch) // 0 -> 10
+ .setPreviousMemberEpoch(0) // 0 -> 0
+ .setAssignedTasks(resetAssignedTasksEpochsToZero(givenAssignedTasks))
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentTombstoneRecord(groupId, oldMemberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord(groupId, oldMemberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord(groupId, oldMemberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(groupId, newJoinStaticMember),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord(groupId, newJoinStaticMember.memberId(), givenTargetAssignment),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, newJoinStaticMember),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, withPrevMemberId)
+ ),
+ rejoinResult.records()
+ );
+ }
+
+ @Test
+ public void testStaticMemberLeaveWithLeaveGroupStaticMemberEpochFromUnrevokedState() {
+ int leaveEpoch = LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+ int memberEpoch = DEFAULT_MEMBER_EPOCH;
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+ String groupId = "fooup";
+ String instanceId = Uuid.randomUuid().toString();
+ String memberId = Uuid.randomUuid().toString();
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4);
+ TasksTupleWithEpochs assignedTasks = topic.assignedTasks(memberEpoch, 0, 1);
+ TasksTupleWithEpochs tasksPendingRevocation = topic.assignedTasks(memberEpoch, 2, 3);
+ TasksTuple targetAssignment = topic.targetAssignment(0, 1);
+
+ StreamsGroupMember unrevokedMember = streamsGroupMemberBuilderWithDefaults(memberId, instanceId)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setState(MemberState.UNREVOKED_TASKS)
+ .setAssignedTasks(assignedTasks)
+ .setTasksPendingRevocation(tasksPendingRevocation)
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(unrevokedMember)
+ .withTargetAssignment(memberId, targetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, memberId, instanceId, leaveEpoch)
+ );
+
+ StreamsGroupMember expectedMember = streamsGroupMemberBuilderWithDefaults(memberId, instanceId)
+ .setMemberEpoch(leaveEpoch)
+ .setState(MemberState.UNREVOKED_TASKS)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setAssignedTasks(resetAssignedTasksEpochsToZero(assignedTasks))
+ .build();
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId)
+ .setMemberEpoch(leaveEpoch)
+ .setStatus(List.of())
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null),
+ result.response().data()
+ );
+ assertRecordsEquals(
+ List.of(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, expectedMember)),
+ result.records()
+ );
+ assertEquals(MemberState.UNREVOKED_TASKS, context.streamsGroupMemberState(groupId, memberId));
+ assertEquals(groupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+ }
+
+ @Test
+ public void testStaticMemberRejoinsAfterTemporaryLeave() {
+ int leaveEpoch = LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+ int memberEpoch = DEFAULT_MEMBER_EPOCH;
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+ String groupId = "fooup";
+ String instanceId = Uuid.randomUuid().toString();
+ String oldMemberId = Uuid.randomUuid().toString();
+ String newMemberId = Uuid.randomUuid().toString();
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4);
+ TasksTupleWithEpochs assignedTasks = topic.assignedTasks(memberEpoch, 0, 1);
+ TasksTuple targetAssignment = topic.targetAssignment(0, 1);
+
+ StreamsGroupMember temporarilyLeftMember = streamsGroupMemberBuilderWithDefaults(oldMemberId, instanceId)
+ .setMemberEpoch(leaveEpoch)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setAssignedTasks(resetAssignedTasksEpochsToZero(assignedTasks))
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(temporarilyLeftMember)
+ .withTargetAssignment(oldMemberId, targetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, newMemberId, instanceId, JOIN_GROUP_MEMBER_EPOCH)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(newMemberId)
+ .setMemberEpoch(memberEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(0, 1))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ result.response().data()
+ );
+ assertEquals(MemberState.STABLE, context.streamsGroupMemberState(groupId, newMemberId));
+ assertEquals(groupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+ }
+
+ @Test
+ public void testStaticMemberLeaveWithLeaveGroupStaticMemberEpochFromUnreleasedState() {
+ int leaveEpoch = LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+ int memberEpoch = DEFAULT_MEMBER_EPOCH;
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+ String groupId = "fooup";
+ String instanceId = Uuid.randomUuid().toString();
+ String memberId = Uuid.randomUuid().toString();
+ String processId = Uuid.randomUuid().toString();
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 3);
+ TasksTupleWithEpochs assignedTasks = topic.assignedTasks(memberEpoch, 0, 1);
+ TasksTuple targetAssignment = topic.targetAssignment(0, 1, 2);
+
+ StreamsGroupMember unreleasedMember = streamsGroupMemberBuilderWithDefaults(memberId, instanceId)
+ .setProcessId(processId)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setState(MemberState.UNRELEASED_TASKS)
+ .setAssignedTasks(assignedTasks)
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(unreleasedMember)
+ .withTargetAssignment(memberId, targetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, memberId, instanceId, leaveEpoch)
+ );
+
+ StreamsGroupMember expectedMember = streamsGroupMemberBuilderWithDefaults(memberId, instanceId)
+ .setProcessId(processId)
+ .setMemberEpoch(leaveEpoch)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setState(MemberState.UNRELEASED_TASKS)
+ .setAssignedTasks(resetAssignedTasksEpochsToZero(assignedTasks))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build();
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId)
+ .setMemberEpoch(leaveEpoch)
+ .setStatus(List.of())
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null),
+ result.response().data()
+ );
+ assertRecordsEquals(
+ List.of(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, expectedMember)),
+ result.records()
+ );
+ assertEquals(MemberState.UNRELEASED_TASKS, context.streamsGroupMemberState(groupId, memberId));
+ assertEquals(groupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+ }
+
+ @Test
+ public void testStaticMemberRejoinsAfterTemporaryLeaveFromUnreleasedState() {
+ int leaveEpoch = LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+ int memberEpoch = DEFAULT_MEMBER_EPOCH;
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+ String groupId = "fooup";
+ String instanceId = Uuid.randomUuid().toString();
+ String oldMemberId = "old-member-id";
+ String newMemberId = "new-member-id";
+ String otherMemberId = "other-member-id";
+ String oldProcessId = "old-process-id";
+ String otherProcessId = "other-process-id";
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 3);
+ TasksTupleWithEpochs assignedTasks = topic.assignedTasks(memberEpoch, 0, 1);
+ TasksTupleWithEpochs otherTasksPendingRevocation = topic.assignedTasks(memberEpoch, 2);
+ TasksTuple targetAssignment = topic.targetAssignment(0, 1, 2);
+
+ StreamsGroupMember temporarilyLeftMember = StreamsGroupTestUtil.streamsGroupMemberBuilderWithDefaults(oldMemberId, instanceId)
+ .setProcessId(oldProcessId)
+ .setMemberEpoch(leaveEpoch)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setState(MemberState.UNRELEASED_TASKS)
+ .setAssignedTasks(assignedTasks)
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build();
+
+ StreamsGroupMember otherMember = StreamsGroupTestUtil.streamsGroupMemberBuilderWithDefaults(otherMemberId)
+ .setProcessId(otherProcessId)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setState(MemberState.UNREVOKED_TASKS)
+ .setTasksPendingRevocation(otherTasksPendingRevocation)
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(temporarilyLeftMember)
+ .withMember(otherMember)
+ .withTargetAssignment(oldMemberId, targetAssignment)
+ .withTargetAssignment(otherMemberId, TasksTuple.EMPTY)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, newMemberId, instanceId, JOIN_GROUP_MEMBER_EPOCH)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(newMemberId)
+ .setMemberEpoch(memberEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(0, 1))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ result.response().data()
+ );
+ assertEquals(MemberState.UNRELEASED_TASKS, context.streamsGroupMemberState(groupId, newMemberId));
+ assertEquals(groupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+ }
+
+
+ @Test
+ public void testStaticMemberLeaveWithLeaveGroupStaticMemberEpochFromUnrevokedStateAllowsUnreleasedMemberToProgress() {
+ int leaveEpoch = LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+ int memberEpoch = DEFAULT_MEMBER_EPOCH;
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+ String groupId = "fooup";
+ String instanceId = Uuid.randomUuid().toString();
+ String leavingMemberId = Uuid.randomUuid().toString();
+ String waitingMemberId = Uuid.randomUuid().toString();
+
+ String subtopology1 = "subtopology1";
+ StreamsTopicFixture topic = streamsTopicFixture(subtopology1, "foo", 3);
+
+ TasksTupleWithEpochs assignedTasks = topic.assignedTasks(memberEpoch, 0, 1);
+ TasksTupleWithEpochs tasksPendingRevocation = topic.assignedTasks(memberEpoch, 2);
+ TasksTuple leavingTargetAssignment = topic.targetAssignment(0, 1);
+ TasksTuple waitingTargetAssignment = topic.targetAssignment(2);
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(leavingMemberId, instanceId)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setState(MemberState.UNREVOKED_TASKS)
+ .setAssignedTasks(assignedTasks)
+ .setTasksPendingRevocation(tasksPendingRevocation)
+ .build())
+ .withMember(StreamsGroupTestUtil.streamsGroupMemberBuilderWithDefaults(waitingMemberId)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setState(MemberState.UNRELEASED_TASKS)
+ .build())
+ .withTargetAssignment(leavingMemberId, leavingTargetAssignment)
+ .withTargetAssignment(waitingMemberId, waitingTargetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> leaveResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, leavingMemberId, instanceId, leaveEpoch)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(leavingMemberId)
+ .setMemberEpoch(leaveEpoch)
+ .setStatus(List.of())
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null),
+ leaveResult.response().data()
+ );
+ List<CoordinatorRecord> expectedRecordsTriggeredByLeave = List.of(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId,
+ streamsGroupMemberBuilderWithDefaults(leavingMemberId, instanceId)
+ .setMemberEpoch(leaveEpoch)
+ .setState(MemberState.UNREVOKED_TASKS)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setAssignedTasks(resetAssignedTasksEpochsToZero(assignedTasks))
+ .build()));
+ assertRecordsEquals(expectedRecordsTriggeredByLeave, leaveResult.records());
+ assertEquals(MemberState.UNREVOKED_TASKS, context.streamsGroupMemberState(groupId, leavingMemberId));
+
+ // Waiting member send a heartbeat expecting get unreleased tasks.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> waitingMemberResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, waitingMemberId, null, memberEpoch)
+ );
+
+ StreamsGroupHeartbeatResponseData expectedWaitingMemberResponse = new StreamsGroupHeartbeatResponseData()
+ .setMemberId(waitingMemberId)
+ .setMemberEpoch(memberEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(2))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of());
+ assertResponseEquals(expectedWaitingMemberResponse, waitingMemberResult.response().data());
+
+ List<CoordinatorRecord> expectedRecordsTriggeredByWaitngMember = List.of(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId,
+ StreamsGroupTestUtil.streamsGroupMemberBuilderWithDefaults(waitingMemberId)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(topic.assignedTasks(memberEpoch, 2))
+ .build())
+ );
+ assertRecordsEquals(expectedRecordsTriggeredByWaitngMember, waitingMemberResult.records());
+
+ assertEquals(MemberState.STABLE, context.streamsGroupMemberState(groupId, waitingMemberId));
+ assertEquals(groupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+ }
+
+
+ @Test
+ public void testStaticMemberLeaveWithLeaveGroupStaticMemberEpochAndRejoinAndOtherRackIdThenGroupBumpOccur() {
+ int leaveEpoch = LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+ int memberEpoch = DEFAULT_MEMBER_EPOCH;
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+ String groupId = "fooup";
+ String rackId = Uuid.randomUuid().toString();
+ String memberId = Uuid.randomUuid().toString();
+ String instanceId = Uuid.randomUuid().toString();
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4);
+
+ TasksTupleWithEpochs givenAssignedTasks = topic.assignedTasks(memberEpoch, 0, 1, 2, 3);
+ TasksTuple givenTargetAssignment = topic.targetAssignment(0, 1, 2, 3);
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(memberId, instanceId)
+ .setMemberEpoch(memberEpoch)
+ .setRackId(rackId)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setAssignedTasks(givenAssignedTasks)
+ .build())
+ .withTargetAssignment(memberId, givenTargetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> normalHeartbeatResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, memberId, instanceId, memberEpoch)
+ .setRackId(rackId)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId)
+ .setMemberEpoch(memberEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null),
+ normalHeartbeatResult.response().data());
+ assertEquals(groupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> leaveResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, memberId, instanceId, leaveEpoch)
+ .setRackId(rackId)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId)
+ .setMemberEpoch(leaveEpoch)
+ .setStatus(List.of())
+ .setActiveTasks(null)
+ .setWarmupTasks(null)
+ .setStandbyTasks(null),
+ leaveResult.response().data()
+ );
+ assertEquals(groupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+
+ String newMemberId = Uuid.randomUuid().toString();
+ String newRackId = Uuid.randomUuid().toString();
+ assignor.prepareGroupAssignment(Map.of(newMemberId, givenTargetAssignment));
+
+ int bumpedGroupEpoch = groupEpoch + 1;
+ int bumpedMemberEpoch = memberEpoch + 1;
+
+ // Streams Member rejoin with other memberId and rackId
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> rejoinResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, newMemberId, instanceId, JOIN_GROUP_MEMBER_EPOCH)
+ .setRackId(newRackId)
+ );
+
+ // Inherit previous member's member epoch, and assigned tasks.
+ // member epoch should be bumped.
+ // group epoch should be bumped.
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(newMemberId)
+ .setMemberEpoch(bumpedMemberEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(0, 1, 2, 3))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of()),
+ rejoinResult.response().data()
+ );
+ assertEquals(bumpedGroupEpoch, context.groupMetadataManager.streamsGroup(groupId).groupEpoch());
+
+ StreamsGroupMember transationStaticInitMember = streamsGroupMemberBuilderWithDefaults(newMemberId, instanceId)
+ .setMemberEpoch(JOIN_GROUP_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(JOIN_GROUP_MEMBER_EPOCH)
+ .setRackId(rackId)
+ .setAssignedTasks(resetAssignedTasksEpochsToZero(givenAssignedTasks))
+ .build();
+
+ StreamsGroupMember newJoinStaticMember = streamsGroupMemberBuilderWithDefaults(newMemberId, instanceId)
+ .setMemberEpoch(JOIN_GROUP_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(JOIN_GROUP_MEMBER_EPOCH)
+ .setRackId(newRackId)
+ .setAssignedTasks(resetAssignedTasksEpochsToZero(givenAssignedTasks))
+ .build();
+
+ StreamsGroupMember reconciledMember = streamsGroupMemberBuilderWithDefaults(newMemberId, instanceId)
+ .setMemberEpoch(bumpedMemberEpoch)
+ .setPreviousMemberEpoch(0)
+ .setRackId(newRackId)
+ .setAssignedTasks(resetAssignedTasksEpochsToZero(givenAssignedTasks))
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ // From replaceStreamsMembers
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord(groupId, memberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord(groupId, memberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(groupId, transationStaticInitMember),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord(groupId, transationStaticInitMember.memberId(), givenTargetAssignment),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, transationStaticInitMember),
+
+ // From hasStreamsMemberMetadataChanged
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(groupId, newJoinStaticMember),
+
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(groupId, bumpedGroupEpoch, topic.metadataHash(), 0, getDefaultAssignmentConfigs(), -1, -1),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord(groupId, newJoinStaticMember.memberId(), givenTargetAssignment),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentMetadataRecord(groupId, bumpedGroupEpoch, context.time.milliseconds()),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, reconciledMember)
+ ),
+ rejoinResult.records()
+ );
+ }
+
+
+ @Test
+ public void testStaticMemberRejoinWritesReplacementRecordsInStreamsGroup() {
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+
+ String groupId = "fooup";
+ String oldMemberId = Uuid.randomUuid().toString();
+ String rejoinMemberId = Uuid.randomUuid().toString();
+ String instanceId = Uuid.randomUuid().toString();
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4);
+ TasksTuple oldTargetAssignment = topic.targetAssignment(0, 1, 2, 3);
+ TasksTupleWithEpochs assignedTasks = topic.assignedTasks(groupEpoch, 0, 1, 2, 3);
+
+ StreamsGroupMember oldMember = streamsGroupMemberBuilderWithDefaults(oldMemberId, instanceId)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(groupEpoch)
+ .setAssignedTasks(resetAssignedTasksEpochsToZero(assignedTasks))
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(oldMember)
+ .withTargetAssignment(oldMemberId, oldTargetAssignment)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, rejoinMemberId, instanceId, DEFAULT_PROCESS_ID)
+ );
+
+ assertEquals(rejoinMemberId, result.response().data().memberId());
+ assertEquals(groupEpoch, result.response().data().memberEpoch());
+
+ StreamsGroupMember expectedCopiedMember = new StreamsGroupMember.Builder(oldMember, rejoinMemberId)
+ .setMemberEpoch(JOIN_GROUP_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(JOIN_GROUP_MEMBER_EPOCH)
+ .build();
+
+ assertTrue(result.records().contains(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentTombstoneRecord(groupId, oldMemberId)
+ ));
+ assertTrue(result.records().contains(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord(groupId, oldMemberId)
+ ));
+ assertTrue(result.records().contains(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord(groupId, oldMemberId)
+ ));
+ assertTrue(result.records().contains(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(groupId, expectedCopiedMember)
+ ));
+ assertTrue(result.records().contains(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord(groupId, rejoinMemberId, oldTargetAssignment)
+ ));
+ assertTrue(result.records().contains(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, expectedCopiedMember)
+ ));
+ }
+
+ @Test
+ public void testStaticMemberLeaveWithMismatchedMemberIdThrowsFencedInstanceIdInStreamsGroup() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String differentMemberId = Uuid.randomUuid().toString();
+ String instanceId = Uuid.randomUuid().toString();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, 10)
+ .withMember(streamsGroupMemberBuilderWithDefaults(memberId, instanceId)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .build())
+ .withTargetAssignmentEpoch(10)
+ )
+ .build();
+
+ assertThrows(FencedInstanceIdException.class, () ->
+ context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, differentMemberId, instanceId, LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ )
+ );
+ }
+
+ @Test
+ public void testUnknownStaticMemberHeartbeatWithPositiveEpochThrowsUnknownMemberIdInStreamsGroup() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String instanceId = Uuid.randomUuid().toString();
+ String unknownInstanceId = Uuid.randomUuid().toString();
+ String unknownMemberId = Uuid.randomUuid().toString();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, 10)
+ .withMember(streamsGroupMemberBuilderWithDefaults(memberId, instanceId)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .build())
+ .withTargetAssignmentEpoch(10)
+ )
+ .build();
+
+ UnknownMemberIdException e = assertThrows(UnknownMemberIdException.class, () ->
+ context.streamsGroupHeartbeat(staticHeartbeat(groupId, unknownMemberId, unknownInstanceId, 1))
+ );
+ assertEquals(String.format("Instance id %s is unknown.", unknownInstanceId), e.getMessage());
+ }
+
+ @ParameterizedTest
+ @MethodSource("ownedActiveTasksAtPreviousEpochCases")
+ public void testStreamsStaticMemberHeartbeatWithPreviousEpochAndOwnedActiveTasks(
+ List<Integer> requestAssignedTaskIds, Class<? extends Exception> expectedException
+ ) {
+ Integer[] givenAssignedTasksIds = new Integer[]{0, 1, 2, 3};
+
+ verifyStreamsStaticMemberHeartbeatWithOwnedActiveTasksAtPreviousEpoch(
+ givenAssignedTasksIds,
+ requestAssignedTaskIds,
+ expectedException
+ );
+ }
+
+ private static Stream<Arguments> ownedActiveTasksAtPreviousEpochCases() {
+ return Stream.of(
+ Arguments.of(List.of(0, 1, 2), null), // Subset Owned Active Tasks
+ Arguments.of(List.of(0, 1, 2, 3), null), // Exact Owned Active Tasks
+ Arguments.of(List.of(0, 1, 2, 3, 4), FencedMemberEpochException.class) // Non Subset active tasks
+ );
+ }
+
+ private void verifyStreamsStaticMemberHeartbeatWithOwnedActiveTasksAtPreviousEpoch(
+ Integer[] givenTaskIds,
+ List<Integer> requestAssignedTaskIds,
+ Class<? extends Exception> expectedException) {
+ int groupEpoch = 10;
+ int partitionSize = 5;
+ int currentMemberEpoch = 10;
+ int previousMemberEpoch = 9;
+ int requestMemberEpoch = 9;
+
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String instanceId = Uuid.randomUuid().toString();
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", partitionSize);
+ TasksTupleWithEpochs givenAssignedTask = topic.assignedTasks(groupEpoch, givenTaskIds);
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withMember(streamsGroupMemberBuilderWithDefaults(memberId, instanceId)
+ .setMemberEpoch(currentMemberEpoch)
+ .setPreviousMemberEpoch(previousMemberEpoch)
+ .setAssignedTasks(givenAssignedTask)
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build())
+ .withTargetAssignment(memberId, topic.targetAssignment(givenTaskIds))
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT)
+ .build();
+
+ StreamsGroupHeartbeatRequestData requestData = staticHeartbeat(groupId, memberId, instanceId, requestMemberEpoch)
+ .setProcessId("process-id")
+ .setRebalanceTimeoutMs(1500)
+ .setTopology(topic.topology())
+ .setActiveTasks(topic.requestTasks(requestAssignedTaskIds))
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of());
+
+ if (expectedException != null) {
+ assertThrows(expectedException, () -> context.streamsGroupHeartbeat(requestData));
+ } else {
+ assertDoesNotThrow(() -> context.streamsGroupHeartbeat(requestData));
+ }
+ }
+
+ @Test
+ public void testStreamsStaticMemberTemporaryLeaveSessionTimeoutExpiration() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String instanceId = Uuid.randomUuid().toString();
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4);
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(topic.metadataImage())
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, 0)
+ .build();
+
+ assignor.prepareGroupAssignment(Map.of(memberId, topic.targetAssignment(0, 1, 2, 3)));
+
+ // static member joins (session timeout should be scheduled)
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> firstJoinResult = context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, memberId, instanceId, topic).setRebalanceTimeoutMs(90000)
+ );
+
+ // member epoch should be bumped up.
+ // session timeout should be 45000ms.
+ assertEquals(2, firstJoinResult.response().data().memberEpoch());
+ context.assertSessionTimeout(groupId, memberId, 45000);
+
+ // static member leaves temporarily.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> temporaryLeaveResult = context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, memberId, instanceId, LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ );
+
+ // member epoch should be -2.
+ // session timeout still 45000ms.
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(memberId)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setStatus(List.of()),
+ temporaryLeaveResult.response().data()
+ );
+ context.assertSessionTimeout(groupId, memberId, 45000);
+
+ // no rejoin, session timeout expires.
+ List<MockCoordinatorTimer.ExpiredTimeout<CoordinatorRecord>> timeouts = context.sleep(45000 + 1);
+
+ List<CoordinatorRecord> expectedRecords = List.of(
+ StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord(groupId, memberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord(groupId, memberId),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(groupId, 3, topic.metadataHash(), 0, getDefaultAssignmentConfigs(), -1, -1),
+ StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentMetadataRecord(groupId, 3, 0L)
+ );
+ assertEquals(
+ List.of(new MockCoordinatorTimer.ExpiredTimeout<>(
+ groupSessionTimeoutKey(groupId, memberId),
+ new CoordinatorResult<>(expectedRecords)
+ )),
+ timeouts
+ );
+ context.assertNoSessionTimeout(groupId, memberId);
+ context.assertNoRebalanceTimeout(groupId, memberId);
+ }
+
+ @Test
+ public void testStaticMemberJoinEmptyStreamsGroupRegistersStaticMember1() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String instanceId = Uuid.randomUuid().toString();
+
+ StreamsGroupHeartbeatRequestData.Topology topology = new StreamsGroupHeartbeatRequestData.Topology().setSubtopologies(List.of());
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ assignor.prepareGroupAssignment(Map.of(memberId, TasksTuple.EMPTY));
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withMetadataImage(new MetadataImageBuilder().buildCoordinatorMetadataImage())
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, 0)
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .build();
+
+ // There is no group at all.
+ assertThrows(GroupIdNotFoundException.class, () ->
+ context.groupMetadataManager.streamsGroup(groupId)
+ );
+
+ context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, memberId, instanceId, JOIN_GROUP_MEMBER_EPOCH)
+ .setProcessId(DEFAULT_PROCESS_ID)
+ .setRebalanceTimeoutMs(1500)
+ .setTopology(topology)
+ .setActiveTasks(List.of())
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of())
+ );
+
+ StreamsGroup group = context.groupMetadataManager.streamsGroup(groupId);
+ assertEquals(memberId, group.staticMember(instanceId).memberId());
+ assertEquals(Optional.of(instanceId), group.getMemberOrThrow(memberId).instanceId());
+ }
+
+ @ParameterizedTest
+ @MethodSource("userEndpointTestCases")
+ public void testStaticMemberRejoinUpdatesUserEndpointInformationEpoch(
+ StreamsGroupHeartbeatRequestData.Endpoint firstUserEndpoint,
+ int firstExpectedUserEndpointEpoch,
+ List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> firstExpectedPartitionsByUserEndpoint,
+ StreamsGroupMemberMetadataValue.Endpoint firstExpectedUserEndpointMetadata,
+
+ StreamsGroupHeartbeatRequestData.Endpoint secondUserEndpoint,
+ int secondExpectedUserEndpointEpoch,
+ List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> secondExpectedPartitionsByUserEndpoint,
+ StreamsGroupMemberMetadataValue.Endpoint secondExpectedUserEndpointMetadata
+ ) {
+ int memberEpoch = DEFAULT_MEMBER_EPOCH;
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+ int bumpedEpoch = memberEpoch + 1;
+
+ String groupId = "fooup";
+ String instanceId = Uuid.randomUuid().toString();
+ String oldMemberId = "old-member-id";
+ String rejoinMemberId = "new-member-id";
+
+ StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 3);
+ TasksTuple targetAssignment = topic.targetAssignment(0, 1, 2);
+
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ assignor.prepareGroupAssignment(Map.of(oldMemberId, topic.targetAssignment(0, 1, 2)));
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(topic.metadataImage())
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology()))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(topic.metadataHash())
+ .withTargetAssignment(oldMemberId, targetAssignment)
+ .withLastAssignmentConfigs(getDefaultAssignmentConfigs()))
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, 0)
+ .build();
+
+ assertEquals(0, context.groupMetadataManager.streamsGroup(groupId).endpointInformationEpoch());
+
+ // First Join -> First Input
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, oldMemberId, instanceId, topic)
+ .setUserEndpoint(firstUserEndpoint) // first input
+ );
+
+ // First Check
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(oldMemberId)
+ .setMemberEpoch(bumpedEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(0, 1, 2))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of())
+ .setEndpointInformationEpoch(firstExpectedUserEndpointEpoch) // first endpoint epoch
+ .setPartitionsByUserEndpoint(firstExpectedPartitionsByUserEndpoint), // first partitions by user endpoint
+ result.response().data()
+ );
+
+ if (firstExpectedUserEndpointMetadata != null) {
+ assertEquals(firstExpectedUserEndpointMetadata, context.groupMetadataManager.streamsGroup(groupId).getMemberOrThrow(oldMemberId).userEndpoint().get());
+ } else {
+ assertTrue(context.groupMetadataManager.streamsGroup(groupId).getMemberOrThrow(oldMemberId).userEndpoint().isEmpty());
+ }
+ assertEquals(firstExpectedUserEndpointEpoch, context.groupMetadataManager.streamsGroup(groupId).endpointInformationEpoch());
+
+ // static member leaves with epoch -2.
+ context.streamsGroupHeartbeat(
+ staticHeartbeat(groupId, oldMemberId, instanceId, StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ );
+
+ // static member rejoins.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> rejoinResult = context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, rejoinMemberId, instanceId, topic)
+ .setUserEndpoint(secondUserEndpoint)
+ );
+
+ assertResponseEquals(
+ new StreamsGroupHeartbeatResponseData()
+ .setMemberId(rejoinMemberId)
+ .setMemberEpoch(bumpedEpoch)
+ .setHeartbeatIntervalMs(5000)
+ .setTaskOffsetIntervalMs(60000)
+ .setAcceptableRecoveryLag(10000)
+ .setActiveTasks(topic.responseTasks(0, 1, 2))
+ .setWarmupTasks(List.of())
+ .setStandbyTasks(List.of())
+ .setEndpointInformationEpoch(secondExpectedUserEndpointEpoch)
+ .setPartitionsByUserEndpoint(secondExpectedPartitionsByUserEndpoint),
+ rejoinResult.response().data()
+ );
+
+ if (secondExpectedUserEndpointMetadata != null) {
+ assertEquals(secondExpectedUserEndpointMetadata, context.groupMetadataManager.streamsGroup(groupId).getMemberOrThrow(rejoinMemberId).userEndpoint().get());
+ } else {
+ assertTrue(context.groupMetadataManager.streamsGroup(groupId).getMemberOrThrow(rejoinMemberId).userEndpoint().isEmpty());
+ }
+ assertEquals(secondExpectedUserEndpointEpoch, context.groupMetadataManager.streamsGroup(groupId).endpointInformationEpoch());
+ }
+
+ private static Stream<Arguments> userEndpointTestCases() {
+ return Stream.of(
+ Arguments.of(
+ null, // firstInput
+ 0, // first endpoint Epoch
+ null, // first partitionsByUserEndpoint
+ null, // first group metadata userEndpoint
+ userEndpoint("bar.com", 8080), // second input
+ 1, // second endpoint epoch
+ buildEndpoints("bar.com", 8080, "foo", List.of(0, 1, 2)), // second partitionsByUserEndpoint
+ userEndpointForMetadata("bar.com", 8080)
+ ),
+ Arguments.of(
+ null, // firstInput
+ 0, // first endpoint Epoch
+ null, // first partitionsByUserEndpoint
+ null, // first group metadata userEndpoint
+ null, // second input
+ 0, // second endpoint epoch
+ null, // second partitionsByUserEndpoint
+ null
+ ),
+ Arguments.of(
+ userEndpoint("foo.com", 8080), // firstInput
+ 1, // first endpoint Epoch
+ buildEndpoints("foo.com", 8080, "foo", List.of(0, 1, 2)), // first partitionsByUserEndpoint
+ userEndpointForMetadata("foo.com", 8080), // first group metadata userEndpoint
+ null, // second input
+ 2, // second endpoint epoch
+ List.of(), // second partitionsByUserEndpoint
+ null
+ ),
+ Arguments.of(
+ userEndpoint("foo.com", 8080), // firstInput
+ 1, // first endpoint Epoch
+ buildEndpoints("foo.com", 8080, "foo", List.of(0, 1, 2)), // first partitionsByUserEndpoint
+ userEndpointForMetadata("foo.com", 8080), // first group metadata userEndpoint
+ userEndpoint("foo.com", 8080), // second input
+ 1, // second endpoint epoch
+ buildEndpoints("foo.com", 8080, "foo", List.of(0, 1, 2)), // second partitionsByUserEndpoint
+ userEndpointForMetadata("foo.com", 8080)
+ ),
+ Arguments.of(
+ userEndpoint("foo.com", 8080), // firstInput
+ 1, // first endpoint Epoch
+ buildEndpoints("foo.com", 8080, "foo", List.of(0, 1, 2)), // first partitionsByUserEndpoint
+ userEndpointForMetadata("foo.com", 8080), // first group metadata userEndpoint
+ userEndpoint("bar.com", 8080), // second input
+ 2, // second endpoint epoch
+ buildEndpoints("bar.com", 8080, "foo", List.of(0, 1, 2)), // second partitionsByUserEndpoint
+ userEndpointForMetadata("bar.com", 8080)
+ )
+ );
+ }
+
+ private static StreamsGroupHeartbeatRequestData.Endpoint userEndpoint(String host, int port) {
+ return new StreamsGroupHeartbeatRequestData.Endpoint()
+ .setHost(host)
+ .setPort(port);
+ }
+
+ private static List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> buildEndpoints(String host, int port, String topic, List<Integer> partitions) {
+ List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> endpoints = new ArrayList<>();
+ endpoints.add(new StreamsGroupHeartbeatResponseData.EndpointToPartitions()
+ .setUserEndpoint(new StreamsGroupHeartbeatResponseData.Endpoint()
+ .setHost(host)
+ .setPort(port))
+ .setActivePartitions(List.of(topicPartition(topic, partitions))));
+ return endpoints;
+ }
+
+ private static StreamsGroupMemberMetadataValue.Endpoint userEndpointForMetadata(String host, int port) {
+ return new StreamsGroupMemberMetadataValue.Endpoint().setHost(host).setPort(port);
+ }
+
+ private static StreamsGroupHeartbeatResponseData.TopicPartition topicPartition(String topic, List<Integer> partitions) {
+ return new StreamsGroupHeartbeatResponseData.TopicPartition().setTopic(topic).setPartitions(partitions);
+ }
+
+}
\ No newline at end of file
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupTestUtil.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupTestUtil.java
new file mode 100644
index 00000000000..5276ab6b00a
--- /dev/null
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupTestUtil.java
@@ -0,0 +1,230 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataImage;
+import org.apache.kafka.coordinator.common.runtime.MetadataImageBuilder;
+import org.apache.kafka.coordinator.group.streams.MemberState;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupMember;
+import org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil;
+import org.apache.kafka.coordinator.group.streams.TasksTuple;
+import org.apache.kafka.coordinator.group.streams.TasksTupleWithEpochs;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+
+import static org.apache.kafka.coordinator.group.GroupMetadataManagerTestContext.DEFAULT_CLIENT_ADDRESS;
+import static org.apache.kafka.coordinator.group.GroupMetadataManagerTestContext.DEFAULT_CLIENT_ID;
+import static org.apache.kafka.coordinator.group.GroupMetadataManagerTestContext.DEFAULT_PROCESS_ID;
+import static org.apache.kafka.coordinator.group.Utils.computeGroupHash;
+import static org.apache.kafka.coordinator.group.Utils.computeTopicHash;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTasksTupleWithCommonEpoch;
+
+class StreamsGroupTestUtil {
+
+ static StreamsGroupMember.Builder streamsGroupMemberBuilderWithDefaults(String memberId) {
+ return streamsGroupMemberBuilderWithDefaults(memberId, null);
+ }
+
+ static StreamsGroupMember.Builder streamsGroupMemberBuilderWithDefaults(String memberId, String instanceId) {
+ return new StreamsGroupMember.Builder(memberId)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setState(MemberState.STABLE)
+ .setRackId(null)
+ .setInstanceId(instanceId)
+ .setRebalanceTimeoutMs(1500)
+ .setAssignedTasks(TasksTupleWithEpochs.EMPTY)
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .setTopologyEpoch(0)
+ .setClientTags(Map.of())
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setProcessId(DEFAULT_PROCESS_ID)
+ .setUserEndpoint(null);
+ }
+
+ /**
+ * Returns the default assignment configurations that would be used by the system.
+ * This matches what streamsGroupAssignmentConfigs() would return.
+ */
+ static Map<String, String> getDefaultAssignmentConfigs() {
+ // Use the same default value as GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT
+ return new TreeMap<>(Map.of(
+ "num.standby.replicas", String.valueOf(GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT)
+ ));
+ }
+
+ static List<StreamsGroupHeartbeatResponseData.TaskIds> mkResponseTasks(
+ String subtopologyId,
+ Integer... partitions
+ ) {
+ return List.of(
+ new StreamsGroupHeartbeatResponseData.TaskIds()
+ .setSubtopologyId(subtopologyId)
+ .setPartitions(Arrays.asList(partitions))
+ );
+ }
+
+ static final int DEFAULT_REBALANCE_TIMEOUT_MS = 1500;
+
+ static StreamsTopicFixture streamsTopicFixture(String subtopologyId, String topicName, int partitions) {
+ return new StreamsTopicFixture(subtopologyId, topicName, partitions);
+ }
+
+ static StreamsGroupHeartbeatRequestData staticHeartbeat(String groupId, String memberId, String instanceId, int memberEpoch) {
+ return new StreamsGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setInstanceId(instanceId)
+ .setMemberId(memberId)
+ .setMemberEpoch(memberEpoch);
+ }
+
+ static StreamsGroupHeartbeatRequestData staticJoinHeartbeat(String groupId, String memberId, String instanceId, StreamsTopicFixture topic) {
+ return staticHeartbeat(groupId, memberId, instanceId, StreamsGroupHeartbeatRequest.JOIN_GROUP_MEMBER_EPOCH)
+ .setProcessId(DEFAULT_PROCESS_ID)
+ .setRebalanceTimeoutMs(DEFAULT_REBALANCE_TIMEOUT_MS)
+ .setTopology(topic.topology)
+ .setActiveTasks(List.of())
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of());
+ }
+
+ static class StreamsTopicFixture {
+ private final String subtopologyId;
+ private final String topicName;
+ private final Uuid topicId;
+ private final StreamsGroupHeartbeatRequestData.Topology topology;
+ private final CoordinatorMetadataImage metadataImage;
+ private final long metadataHash;
+
+ private StreamsTopicFixture(
+ String subtopologyId,
+ String topicName,
+ int partitions
+ ) {
+ this.subtopologyId = subtopologyId;
+ this.topicName = topicName;
+ this.topicId = Uuid.randomUuid();
+ this.topology = new StreamsGroupHeartbeatRequestData.Topology()
+ .setSubtopologies(List.of(
+ new StreamsGroupHeartbeatRequestData.Subtopology()
+ .setSubtopologyId(subtopologyId)
+ .setSourceTopics(List.of(topicName))
+ ));
+ this.metadataImage = new MetadataImageBuilder()
+ .addTopic(topicId, topicName, partitions)
+ .buildCoordinatorMetadataImage();
+ this.metadataHash = computeGroupHash(Map.of(
+ topicName,
+ computeTopicHash(topicName, metadataImage)
+ ));
+ }
+
+ public Map.Entry<String, Set<Integer>> tasks(Integer... partitions) {
+ return TaskAssignmentTestUtil.mkTasks(subtopologyId, partitions);
+ }
+
+ public TasksTuple targetAssignment(Integer... partitions) {
+ return TaskAssignmentTestUtil.mkTasksTuple(
+ TaskAssignmentTestUtil.TaskRole.ACTIVE,
+ tasks(partitions)
+ );
+ }
+
+ public TasksTupleWithEpochs assignedTasks(
+ int epoch,
+ Integer... partitions
+ ) {
+ return mkTasksTupleWithCommonEpoch(
+ TaskAssignmentTestUtil.TaskRole.ACTIVE,
+ epoch,
+ tasks(partitions)
+ );
+ }
+
+ public CoordinatorMetadataImage metadataImage() {
+ return metadataImage;
+ }
+
+ public long metadataHash() {
+ return metadataHash;
+ }
+
+ public StreamsGroupHeartbeatRequestData.Topology topology() {
+ return topology;
+ }
+
+ public List<StreamsGroupHeartbeatResponseData.TaskIds> responseTasks(Integer... partitions) {
+ return mkResponseTasks(subtopologyId, partitions);
+ }
+
+ public List<StreamsGroupHeartbeatRequestData.TaskIds> requestTasks(List<Integer> partitions) {
+ return List.of(
+ new StreamsGroupHeartbeatRequestData.TaskIds()
+ .setSubtopologyId(subtopologyId)
+ .setPartitions(partitions)
+ );
+ }
+ }
+
+ static StreamsGroupHeartbeatRequestData staticJoinHeartbeat(
+ String groupId,
+ String memberId,
+ String instanceId,
+ String processId
+ ) {
+ return staticHeartbeat(groupId, memberId, instanceId, StreamsGroupHeartbeatRequest.JOIN_GROUP_MEMBER_EPOCH)
+ .setProcessId(processId)
+ .setActiveTasks(List.of())
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of());
+ }
+
+ static TasksTupleWithEpochs resetAssignedTasksEpochsToZero(TasksTupleWithEpochs assignedTasks) {
+ if (assignedTasks.isEmpty()) {
+ return assignedTasks;
+ }
+
+ if (assignedTasks.activeTasksWithEpochs().isEmpty()) {
+ return assignedTasks;
+ }
+
+ Map<String, Map<Integer, Integer>> resetActiveTasks = new HashMap<>();
+ for (Map.Entry<String, Map<Integer, Integer>> entry : assignedTasks.activeTasksWithEpochs().entrySet()) {
+ Map<Integer, Integer> resetActiveTaskEpochs = new HashMap<>();
+ for (Integer partitionId : entry.getValue().keySet()) {
+ resetActiveTaskEpochs.put(partitionId, 0);
+ }
+ resetActiveTasks.put(entry.getKey(), resetActiveTaskEpochs);
+ }
+ return new TasksTupleWithEpochs(
+ resetActiveTasks,
+ assignedTasks.standbyTasks(),
+ assignedTasks.warmupTasks()
+ );
+ }
+
+}