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 2024/09/25 16:02:52 UTC
(kafka) 21/28: Rebase on AK trunk 2024-08-15
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch kip1071
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 4dc40157ada8ed8c7517e19ce003f8af9fc4f809
Author: Bruno Cadonna <br...@confluent.io>
AuthorDate: Fri Aug 16 19:58:35 2024 +0200
Rebase on AK trunk 2024-08-15
---
.../internals/AbstractMembershipManager.java | 1 +
.../consumer/internals/RequestManagers.java | 4 ++-
.../StreamsGroupHeartbeatRequestManager.java | 18 ++++++-------
.../events/ApplicationEventProcessor.java | 4 +--
.../StreamsGroupHeartbeatRequestManagerTest.java | 4 +--
.../events/ApplicationEventProcessorTest.java | 2 --
.../group/GroupCoordinatorRecordHelpers.java | 4 +--
.../coordinator/group/GroupMetadataManager.java | 2 +-
.../streams/CoordinatorStreamsRecordHelpers.java | 31 ++++++++++++----------
.../StreamsGroupCurrentMemberAssignmentKey.json | 6 ++---
.../message/StreamsGroupMemberMetadataKey.json | 6 ++---
.../common/message/StreamsGroupMetadataKey.json | 4 +--
.../message/StreamsGroupPartitionMetadataKey.json | 4 +--
.../StreamsGroupTargetAssignmentMemberKey.json | 6 ++---
.../StreamsGroupTargetAssignmentMetadataKey.json | 4 +--
.../common/message/StreamsGroupTopologyKey.json | 4 +--
.../group/GroupCoordinatorServiceTest.java | 30 ++++++++++++++-------
.../CoordinatorStreamsRecordHelpersTest.java | 2 +-
.../group/streams/StreamsGroupTest.java | 6 ++---
.../assignment/TaskAssignorConvergenceTest.java | 4 +--
20 files changed, 80 insertions(+), 66 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java
index 455e48a0c89..b1d0c8f112d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java
@@ -1215,6 +1215,7 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
protected void updateMemberEpoch(int newEpoch) {
boolean newEpochReceived = this.memberEpoch != newEpoch;
+ log.info("Updating member epoch to {}", newEpoch);
this.memberEpoch = newEpoch;
// Simply notify based on epoch change only, given that the member will never receive a
// new member ID without an epoch (member ID is only assigned when it joins the group).
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
index 3d6860a3db8..1cabe7e7ac9 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
@@ -68,7 +68,7 @@ public class RequestManagers implements Closeable {
Optional<CoordinatorRequestManager> coordinatorRequestManager,
Optional<CommitRequestManager> commitRequestManager,
Optional<ConsumerHeartbeatRequestManager> heartbeatRequestManager,
- Optional<ConsumerMembershipManager> membershipManager) {
+ Optional<ConsumerMembershipManager> membershipManager,
Optional<StreamsGroupHeartbeatRequestManager> streamsHeartbeatRequestManager,
Optional<StreamsInitializeRequestManager> streamsInitializeRequestManager) {
this.log = logContext.logger(RequestManagers.class);
@@ -111,6 +111,8 @@ public class RequestManagers implements Closeable {
this.shareHeartbeatRequestManager = shareHeartbeatRequestManager;
this.consumerMembershipManager = Optional.empty();
this.shareMembershipManager = shareMembershipManager;
+ this.streamsHeartbeatRequestManager = Optional.empty();
+ this.streamsInitializeRequestManager = Optional.empty();
this.offsetsRequestManager = null;
this.topicMetadataRequestManager = null;
this.fetchRequestManager = null;
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
index 7b3713b4131..2b3b7c8f248 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
@@ -61,7 +61,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
private final StreamsGroupHeartbeatRequestManager.HeartbeatState heartbeatState;
- private final MembershipManager membershipManager;
+ private final ConsumerMembershipManager membershipManager;
private final StreamsInitializeRequestManager streamsInitializeRequestManager;
@@ -83,7 +83,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
final ConsumerConfig config,
final CoordinatorRequestManager coordinatorRequestManager,
final StreamsInitializeRequestManager streamsInitializeRequestManager,
- final MembershipManager membershipManager,
+ final ConsumerMembershipManager membershipManager,
final BackgroundEventHandler backgroundEventHandler,
final Metrics metrics,
final StreamsAssignmentInterface streamsAssignmentInterface,
@@ -158,7 +158,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
final boolean ignoreResponse) {
NetworkClientDelegate.UnsentRequest request = makeHeartbeatRequest(ignoreResponse);
heartbeatRequestState.onSendAttempt(currentTimeMs);
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
metricsManager.recordHeartbeatSentMs(currentTimeMs);
return request;
}
@@ -335,7 +335,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
this.heartbeatState.reset();
this.heartbeatRequestState.onFailedAttempt(currentTimeMs);
- membershipManager.onHeartbeatFailure();
+ membershipManager.onHeartbeatFailure(false);
switch (error) {
case NOT_COORDINATOR:
@@ -503,7 +503,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
static class HeartbeatState {
- private final MembershipManager membershipManager;
+ private final ConsumerMembershipManager membershipManager;
private final int rebalanceTimeoutMs;
private final StreamsGroupHeartbeatRequestManager.HeartbeatState.SentFields sentFields;
@@ -514,7 +514,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
public HeartbeatState(
final StreamsAssignmentInterface streamsInterface,
- final MembershipManager membershipManager,
+ final ConsumerMembershipManager membershipManager,
final int rebalanceTimeoutMs) {
this.membershipManager = membershipManager;
this.rebalanceTimeoutMs = rebalanceTimeoutMs;
@@ -552,9 +552,9 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
// Immutable -- only sent when joining
if (joining) {
data.setProcessId(streamsInterface.processID().toString());
-// data.setActiveTasks(Collections.emptyList());
-// data.setStandbyTasks(Collections.emptyList());
-// data.setWarmupTasks(Collections.emptyList());
+ data.setActiveTasks(Collections.emptyList());
+ data.setStandbyTasks(Collections.emptyList());
+ data.setWarmupTasks(Collections.emptyList());
streamsInterface.endpoint().ifPresent(streamsEndpoint -> {
data.setUserEndpoint(new StreamsGroupHeartbeatRequestData.Endpoint()
.setHost(streamsEndpoint.host)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
index e0be87e866d..f08338e30bd 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
@@ -229,7 +229,7 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
* it is already a member.
*/
private void process(final SubscriptionChangeEvent ignored) {
- if (!requestManagers.membershipManager.isPresent()) {
+ if (!requestManagers.consumerMembershipManager.isPresent()) {
log.warn("Group membership manager not present when processing a subscribe event");
return;
}
@@ -277,7 +277,7 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
}
private void process(final ConsumerRebalanceListenerCallbackCompletedEvent event) {
- if (!requestManagers.membershipManager.isPresent()) {
+ if (!requestManagers.consumerMembershipManager.isPresent()) {
log.warn(
"An internal error occurred; the group membership manager was not present, so the notification of the {} callback execution could not be sent",
event.methodName()
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
index f4c67e1d9b9..aed1b4dafd7 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
@@ -55,7 +55,6 @@ import java.util.Properties;
import java.util.UUID;
import java.util.stream.Collectors;
-import static org.apache.kafka.clients.consumer.internals.ConsumerTestBuilder.DEFAULT_MAX_POLL_INTERVAL_MS;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.common.utils.Utils.mkSet;
@@ -74,6 +73,7 @@ class StreamsGroupHeartbeatRequestManagerTest {
public static final int TEST_MEMBER_EPOCH = 5;
public static final String TEST_INSTANCE_ID = "instanceId";
public static final int TEST_THROTTLE_TIME_MS = 5;
+ private static final int DEFAULT_MAX_POLL_INTERVAL_MS = 10000;
private StreamsGroupHeartbeatRequestManager heartbeatRequestManager;
private Time time;
@@ -89,7 +89,7 @@ class StreamsGroupHeartbeatRequestManagerTest {
private StreamsInitializeRequestManager streamsInitializeRequestManager;
@Mock
- private MembershipManager membershipManager;
+ private ConsumerMembershipManager membershipManager;
@Mock
private BackgroundEventHandler backgroundEventHandler;
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
index fb79b90d674..9799895dae1 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
@@ -84,8 +84,6 @@ public class ApplicationEventProcessorTest {
withGroupId ? Optional.of(heartbeatRequestManager) : Optional.empty(),
withGroupId ? Optional.of(membershipManager) : Optional.empty(),
Optional.empty(),
- Optional.empty(),
- Optional.empty(),
Optional.empty()
);
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java
index 4f3f11b09ed..4ddba5fea70 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java
@@ -268,7 +268,7 @@ public class GroupCoordinatorRecordHelpers {
* @param memberId The consumer group member id.
* @return The record.
*/
- public static CoordinatorRecord newTargetAssignmentTombstoneRecord(
+ public static CoordinatorRecord newConsumerGroupTargetAssignmentTombstoneRecord(
String groupId,
String memberId
) {
@@ -314,7 +314,7 @@ public class GroupCoordinatorRecordHelpers {
* @param groupId The consumer group id.
* @return The record.
*/
- public static CoordinatorRecord newConsumerTargetAssignmentEpochTombstoneRecord(
+ public static CoordinatorRecord newConsumerGroupTargetAssignmentEpochTombstoneRecord(
String groupId
) {
return new CoordinatorRecord(
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 2fb3bce4885..0dcc8af87bd 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
@@ -3151,7 +3151,7 @@ public class GroupMetadataManager {
if (!updatedMember.equals(member)) {
records.add(newStreamsGroupMemberRecord(groupId, updatedMember));
- if (updatedMember.topologyId().equals(member.topologyId())) {
+ if (!updatedMember.topologyId().equals(member.topologyId())) {
log.info("[GroupId {}] Member {} updated its topology ID to: {}.",
groupId, memberId, updatedMember.topologyId());
return true;
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpers.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpers.java
index f5cac3026aa..2f48151e137 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpers.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpers.java
@@ -58,7 +58,7 @@ public class CoordinatorStreamsRecordHelpers {
new StreamsGroupMemberMetadataKey()
.setGroupId(groupId)
.setMemberId(member.memberId()),
- (short) 14
+ (short) 17
),
new ApiMessageAndVersion(
new StreamsGroupMemberMetadataValue()
@@ -96,7 +96,7 @@ public class CoordinatorStreamsRecordHelpers {
new StreamsGroupMemberMetadataKey()
.setGroupId(groupId)
.setMemberId(memberId),
- (short) 14
+ (short) 17
),
null // Tombstone.
);
@@ -137,7 +137,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupPartitionMetadataKey()
.setGroupId(groupId),
- (short) 13
+ (short) 16
),
new ApiMessageAndVersion(
value,
@@ -159,7 +159,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupPartitionMetadataKey()
.setGroupId(groupId),
- (short) 13
+ (short) 16
),
null // Tombstone.
);
@@ -173,7 +173,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupMetadataKey()
.setGroupId(groupId),
- (short) 12
+ (short) 15
),
new ApiMessageAndVersion(
new StreamsGroupMetadataValue()
@@ -196,7 +196,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupMetadataKey()
.setGroupId(groupId),
- (short) 12
+ (short) 15
),
null // Tombstone.
);
@@ -239,7 +239,7 @@ public class CoordinatorStreamsRecordHelpers {
new StreamsGroupTargetAssignmentMemberKey()
.setGroupId(groupId)
.setMemberId(memberId),
- (short) 16
+ (short) 19
),
new ApiMessageAndVersion(
new StreamsGroupTargetAssignmentMemberValue()
@@ -267,7 +267,7 @@ public class CoordinatorStreamsRecordHelpers {
new StreamsGroupTargetAssignmentMemberKey()
.setGroupId(groupId)
.setMemberId(memberId),
- (short) 16
+ (short) 19
),
null // Tombstone.
);
@@ -282,7 +282,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupTargetAssignmentMetadataKey()
.setGroupId(groupId),
- (short) 15
+ (short) 18
),
new ApiMessageAndVersion(
new StreamsGroupTargetAssignmentMetadataValue()
@@ -305,7 +305,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupTargetAssignmentMetadataKey()
.setGroupId(groupId),
- (short) 15
+ (short) 18
),
null // Tombstone.
);
@@ -320,7 +320,7 @@ public class CoordinatorStreamsRecordHelpers {
new StreamsGroupCurrentMemberAssignmentKey()
.setGroupId(groupId)
.setMemberId(member.memberId()),
- (short) 17
+ (short) 20
),
new ApiMessageAndVersion(
new StreamsGroupCurrentMemberAssignmentValue()
@@ -352,7 +352,7 @@ public class CoordinatorStreamsRecordHelpers {
new StreamsGroupCurrentMemberAssignmentKey()
.setGroupId(groupId)
.setMemberId(memberId),
- (short) 17
+ (short) 20
),
null // Tombstone
);
@@ -402,7 +402,10 @@ public class CoordinatorStreamsRecordHelpers {
.setRepartitionSourceTopics(repartitionSourceTopics).setStateChangelogTopics(stateChangelogTopics));
});
- return new CoordinatorRecord(new ApiMessageAndVersion(new StreamsGroupTopologyKey().setGroupId(groupId), (short) 18),
+ return new CoordinatorRecord(new ApiMessageAndVersion(
+ new StreamsGroupTopologyKey()
+ .setGroupId(groupId),
+ (short) 21),
new ApiMessageAndVersion(value, (short) 0));
}
@@ -419,7 +422,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupTopologyKey()
.setGroupId(groupId),
- (short) 18
+ (short) 21
),
null // Tombstone
);
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupCurrentMemberAssignmentKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupCurrentMemberAssignmentKey.json
index 771f7324a9e..a99e32754c6 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupCurrentMemberAssignmentKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupCurrentMemberAssignmentKey.json
@@ -17,12 +17,12 @@
{
"type": "data",
"name": "StreamsGroupCurrentMemberAssignmentKey",
- "validVersions": "17",
+ "validVersions": "20",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "17",
+ { "name": "GroupId", "type": "string", "versions": "20",
"about": "The group id." },
- { "name": "MemberId", "type": "string", "versions": "17",
+ { "name": "MemberId", "type": "string", "versions": "20",
"about": "The member id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupMemberMetadataKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupMemberMetadataKey.json
index bac9ac247cf..8e0f66deba7 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupMemberMetadataKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupMemberMetadataKey.json
@@ -17,12 +17,12 @@
{
"type": "data",
"name": "StreamsGroupMemberMetadataKey",
- "validVersions": "14",
+ "validVersions": "17",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "14",
+ { "name": "GroupId", "type": "string", "versions": "17",
"about": "The group id." },
- { "name": "MemberId", "type": "string", "versions": "14",
+ { "name": "MemberId", "type": "string", "versions": "17",
"about": "The member id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataKey.json
index 26121bf2ba2..6c54fb8bd14 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataKey.json
@@ -17,10 +17,10 @@
{
"type": "data",
"name": "StreamsGroupMetadataKey",
- "validVersions": "12",
+ "validVersions": "15",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "12",
+ { "name": "GroupId", "type": "string", "versions": "15",
"about": "The group id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupPartitionMetadataKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupPartitionMetadataKey.json
index 546a8f80535..0d91a992d0c 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupPartitionMetadataKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupPartitionMetadataKey.json
@@ -17,10 +17,10 @@
{
"type": "data",
"name": "StreamsGroupPartitionMetadataKey",
- "validVersions": "13",
+ "validVersions": "16",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "13",
+ { "name": "GroupId", "type": "string", "versions": "16",
"about": "The group id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMemberKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMemberKey.json
index 4fc8231ec3d..f96f6f89e01 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMemberKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMemberKey.json
@@ -17,12 +17,12 @@
{
"type": "data",
"name": "StreamsGroupTargetAssignmentMemberKey",
- "validVersions": "16",
+ "validVersions": "19",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "16",
+ { "name": "GroupId", "type": "string", "versions": "19",
"about": "The group id." },
- { "name": "MemberId", "type": "string", "versions": "16",
+ { "name": "MemberId", "type": "string", "versions": "19",
"about": "The member id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMetadataKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMetadataKey.json
index 02b40f727c4..514885f8ad7 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMetadataKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMetadataKey.json
@@ -17,10 +17,10 @@
{
"type": "data",
"name": "StreamsGroupTargetAssignmentMetadataKey",
- "validVersions": "15",
+ "validVersions": "18",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "15",
+ { "name": "GroupId", "type": "string", "versions": "18",
"about": "The group id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupTopologyKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupTopologyKey.json
index 261b755cd51..2e51c117bd7 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupTopologyKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupTopologyKey.json
@@ -17,10 +17,10 @@
{
"type": "data",
"name": "StreamsGroupTopologyKey",
- "validVersions": "18",
+ "validVersions": "21",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "18",
+ { "name": "GroupId", "type": "string", "versions": "21",
"about": "The group id." }
]
}
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 38f742f6659..e0f0a8314a7 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
@@ -267,7 +267,8 @@ public class GroupCoordinatorServiceTest {
new LogContext(),
createConfig(),
runtime,
- new GroupCoordinatorMetrics()
+ new GroupCoordinatorMetrics(),
+ createConfigManager()
);
StreamsGroupInitializeRequestData request = new StreamsGroupInitializeRequestData()
@@ -292,7 +293,8 @@ public class GroupCoordinatorServiceTest {
new LogContext(),
createConfig(),
runtime,
- new GroupCoordinatorMetrics()
+ new GroupCoordinatorMetrics(),
+ createConfigManager()
);
StreamsGroupInitializeRequestData request = new StreamsGroupInitializeRequestData()
@@ -344,7 +346,8 @@ public class GroupCoordinatorServiceTest {
new LogContext(),
createConfig(),
runtime,
- new GroupCoordinatorMetrics()
+ new GroupCoordinatorMetrics(),
+ createConfigManager()
);
StreamsGroupInitializeRequestData request = new StreamsGroupInitializeRequestData()
@@ -379,7 +382,8 @@ public class GroupCoordinatorServiceTest {
new LogContext(),
createConfig(),
runtime,
- new GroupCoordinatorMetrics()
+ new GroupCoordinatorMetrics(),
+ createConfigManager()
);
StreamsGroupHeartbeatRequestData request = new StreamsGroupHeartbeatRequestData()
@@ -404,7 +408,8 @@ public class GroupCoordinatorServiceTest {
new LogContext(),
createConfig(),
runtime,
- new GroupCoordinatorMetrics()
+ new GroupCoordinatorMetrics(),
+ createConfigManager()
);
StreamsGroupHeartbeatRequestData request = new StreamsGroupHeartbeatRequestData()
@@ -455,7 +460,8 @@ public class GroupCoordinatorServiceTest {
new LogContext(),
createConfig(),
runtime,
- new GroupCoordinatorMetrics()
+ new GroupCoordinatorMetrics(),
+ createConfigManager()
);
StreamsGroupHeartbeatRequestData request = new StreamsGroupHeartbeatRequestData()
@@ -1873,7 +1879,8 @@ public class GroupCoordinatorServiceTest {
new LogContext(),
createConfig(),
runtime,
- new GroupCoordinatorMetrics()
+ new GroupCoordinatorMetrics(),
+ createConfigManager()
);
int partitionCount = 2;
service.startup(() -> partitionCount);
@@ -1915,7 +1922,8 @@ public class GroupCoordinatorServiceTest {
new LogContext(),
createConfig(),
runtime,
- new GroupCoordinatorMetrics()
+ new GroupCoordinatorMetrics(),
+ createConfigManager()
);
int partitionCount = 1;
service.startup(() -> partitionCount);
@@ -1949,7 +1957,8 @@ public class GroupCoordinatorServiceTest {
new LogContext(),
createConfig(),
runtime,
- new GroupCoordinatorMetrics()
+ new GroupCoordinatorMetrics(),
+ createConfigManager()
);
int partitionCount = 1;
service.startup(() -> partitionCount);
@@ -1981,7 +1990,8 @@ public class GroupCoordinatorServiceTest {
new LogContext(),
createConfig(),
runtime,
- new GroupCoordinatorMetrics()
+ new GroupCoordinatorMetrics(),
+ createConfigManager()
);
when(runtime.scheduleReadOperation(
ArgumentMatchers.eq("streams-group-describe"),
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpersTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpersTest.java
index 44b46421205..cf1a003cd84 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpersTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpersTest.java
@@ -96,7 +96,7 @@ class CoordinatorStreamsRecordHelpersTest {
new ApiMessageAndVersion(
new StreamsGroupTopologyKey()
.setGroupId("group-id"),
- (short) 18),
+ (short) 21),
new ApiMessageAndVersion(
new StreamsGroupTopologyValue()
.setTopology(expectedTopology),
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
index 3da224f3a95..2fd5d461c99 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
@@ -649,7 +649,7 @@ public class StreamsGroupTest {
group.validateOffsetFetch("member-id", 0, Long.MAX_VALUE));
// Create a member.
- snapshotRegistry.getOrCreateSnapshot(0);
+ snapshotRegistry.idempotentCreateSnapshot(0);
group.updateMember(new StreamsGroupMember.Builder("member-id").build());
// The member does not exist at last committed offset 0.
@@ -716,13 +716,13 @@ public class StreamsGroupTest {
new TopicPartition("__consumer_offsets", 0)
);
StreamsGroup group = new StreamsGroup(snapshotRegistry, "group-foo", metricsShard);
- snapshotRegistry.getOrCreateSnapshot(0);
+ snapshotRegistry.idempotentCreateSnapshot(0);
assertTrue(group.isInStates(Collections.singleton("empty"), 0));
assertFalse(group.isInStates(Collections.singleton("Empty"), 0));
group.updateMember(new StreamsGroupMember.Builder("member1")
.build());
- snapshotRegistry.getOrCreateSnapshot(1);
+ snapshotRegistry.idempotentCreateSnapshot((1));
assertTrue(group.isInStates(Collections.singleton("empty"), 0));
assertTrue(group.isInStates(Collections.singleton("stable"), 1));
assertFalse(group.isInStates(Collections.singleton("empty"), 1));
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/TaskAssignorConvergenceTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/TaskAssignorConvergenceTest.java
index dd703893e9a..e0e345a0ffc 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/TaskAssignorConvergenceTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/TaskAssignorConvergenceTest.java
@@ -472,8 +472,8 @@ public class TaskAssignorConvergenceTest {
@ParameterizedTest
@ValueSource(strings = {
- StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_NONE,
- StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_MIN_TRAFFIC,
+// StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_NONE,
+// StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_MIN_TRAFFIC,
StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_BALANCE_SUBTOPOLOGY
})
public void randomClusterPerturbationsShouldConverge(final String rackAwareStrategy) {