You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/12/01 18:26:36 UTC
(kafka) branch trunk updated: MINOR: Refactor configs in GroupMetadataManager (#17982)
This is an automated email from the ASF dual-hosted git repository.
chia7712 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 44cb90222c6 MINOR: Refactor configs in GroupMetadataManager (#17982)
44cb90222c6 is described below
commit 44cb90222c680b076f563004f8c0e7d8655d77af
Author: David Jacot <dj...@confluent.io>
AuthorDate: Sun Dec 1 19:26:28 2024 +0100
MINOR: Refactor configs in GroupMetadataManager (#17982)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../coordinator/group/GroupCoordinatorConfig.java | 44 +-
.../coordinator/group/GroupCoordinatorShard.java | 14 +-
.../coordinator/group/GroupMetadataManager.java | 280 ++-------
.../group/GroupMetadataManagerTest.java | 660 ++++++++++-----------
.../group/GroupMetadataManagerTestContext.java | 153 +++--
.../group/OffsetMetadataManagerTest.java | 3 +-
.../kafka/server/config/AbstractKafkaConfig.java | 54 +-
7 files changed, 503 insertions(+), 705 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
index f94b2cc706b..5dc576a5dd3 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
@@ -204,13 +204,13 @@ public class GroupCoordinatorConfig {
public static final String OFFSET_COMMIT_TIMEOUT_MS_DOC = "Offset commit will be delayed until all replicas for the offsets topic receive the commit " +
"or this timeout is reached. This is similar to the producer request timeout.";
- public static final ConfigDef GROUP_COORDINATOR_CONFIG_DEF = new ConfigDef()
+ public static final ConfigDef GROUP_COORDINATOR_CONFIG_DEF = new ConfigDef()
.define(GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, INT, GROUP_MIN_SESSION_TIMEOUT_MS_DEFAULT, MEDIUM, GROUP_MIN_SESSION_TIMEOUT_MS_DOC)
.define(GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, INT, GROUP_MAX_SESSION_TIMEOUT_MS_DEFAULT, MEDIUM, GROUP_MAX_SESSION_TIMEOUT_MS_DOC)
.define(GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, INT, GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT, MEDIUM, GROUP_INITIAL_REBALANCE_DELAY_MS_DOC)
.define(GROUP_MAX_SIZE_CONFIG, INT, GROUP_MAX_SIZE_DEFAULT, atLeast(1), MEDIUM, GROUP_MAX_SIZE_DOC);
- public static final ConfigDef NEW_GROUP_CONFIG_DEF = new ConfigDef()
+ public static final ConfigDef NEW_GROUP_CONFIG_DEF = new ConfigDef()
.define(GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, LIST, GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DEFAULT,
ConfigDef.ValidList.in(Group.GroupType.documentValidValues()), MEDIUM, GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DOC)
.define(GROUP_COORDINATOR_NUM_THREADS_CONFIG, INT, GROUP_COORDINATOR_NUM_THREADS_DEFAULT, atLeast(1), MEDIUM, GROUP_COORDINATOR_NUM_THREADS_DOC)
@@ -218,7 +218,7 @@ public class GroupCoordinatorConfig {
// Internal configuration used by integration and system tests.
.defineInternal(NEW_GROUP_COORDINATOR_ENABLE_CONFIG, BOOLEAN, NEW_GROUP_COORDINATOR_ENABLE_DEFAULT, null, MEDIUM, NEW_GROUP_COORDINATOR_ENABLE_DOC);
- public static final ConfigDef OFFSET_MANAGEMENT_CONFIG_DEF = new ConfigDef()
+ public static final ConfigDef OFFSET_MANAGEMENT_CONFIG_DEF = new ConfigDef()
.define(OFFSET_METADATA_MAX_SIZE_CONFIG, INT, OFFSET_METADATA_MAX_SIZE_DEFAULT, HIGH, OFFSET_METADATA_MAX_SIZE_DOC)
.define(OFFSETS_LOAD_BUFFER_SIZE_CONFIG, INT, OFFSETS_LOAD_BUFFER_SIZE_DEFAULT, atLeast(1), HIGH, OFFSETS_LOAD_BUFFER_SIZE_DOC)
.define(OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, SHORT, OFFSETS_TOPIC_REPLICATION_FACTOR_DEFAULT, atLeast(1), HIGH, OFFSETS_TOPIC_REPLICATION_FACTOR_DOC)
@@ -229,7 +229,7 @@ public class GroupCoordinatorConfig {
.define(OFFSETS_RETENTION_CHECK_INTERVAL_MS_CONFIG, LONG, OFFSETS_RETENTION_CHECK_INTERVAL_MS_DEFAULT, atLeast(1), HIGH, OFFSETS_RETENTION_CHECK_INTERVAL_MS_DOC)
.define(OFFSET_COMMIT_TIMEOUT_MS_CONFIG, INT, OFFSET_COMMIT_TIMEOUT_MS_DEFAULT, atLeast(1), HIGH, OFFSET_COMMIT_TIMEOUT_MS_DOC);
- public static final ConfigDef CONSUMER_GROUP_CONFIG_DEF = new ConfigDef()
+ public static final ConfigDef CONSUMER_GROUP_CONFIG_DEF = new ConfigDef()
.define(CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG, INT, CONSUMER_GROUP_SESSION_TIMEOUT_MS_DEFAULT, atLeast(1), MEDIUM, CONSUMER_GROUP_SESSION_TIMEOUT_MS_DOC)
.define(CONSUMER_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, INT, CONSUMER_GROUP_MIN_SESSION_TIMEOUT_MS_DEFAULT, atLeast(1), MEDIUM, CONSUMER_GROUP_MIN_SESSION_TIMEOUT_MS_DOC)
.define(CONSUMER_GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, INT, CONSUMER_GROUP_MAX_SESSION_TIMEOUT_MS_DEFAULT, atLeast(1), MEDIUM, CONSUMER_GROUP_MAX_SESSION_TIMEOUT_MS_DOC)
@@ -240,14 +240,14 @@ public class GroupCoordinatorConfig {
.define(CONSUMER_GROUP_ASSIGNORS_CONFIG, LIST, CONSUMER_GROUP_ASSIGNORS_DEFAULT, null, MEDIUM, CONSUMER_GROUP_ASSIGNORS_DOC)
.define(CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, STRING, CONSUMER_GROUP_MIGRATION_POLICY_DEFAULT, ConfigDef.CaseInsensitiveValidString.in(Utils.enumOptions(ConsumerGroupMigrationPolicy.class)), MEDIUM, CONSUMER_GROUP_MIGRATION_POLICY_DOC);
- public static final ConfigDef SHARE_GROUP_CONFIG_DEF = new ConfigDef()
+ public static final ConfigDef SHARE_GROUP_CONFIG_DEF = new ConfigDef()
.define(SHARE_GROUP_SESSION_TIMEOUT_MS_CONFIG, INT, SHARE_GROUP_SESSION_TIMEOUT_MS_DEFAULT, atLeast(1), MEDIUM, SHARE_GROUP_SESSION_TIMEOUT_MS_DOC)
.define(SHARE_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, INT, SHARE_GROUP_MIN_SESSION_TIMEOUT_MS_DEFAULT, atLeast(1), MEDIUM, SHARE_GROUP_MIN_SESSION_TIMEOUT_MS_DOC)
.define(SHARE_GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, INT, SHARE_GROUP_MAX_SESSION_TIMEOUT_MS_DEFAULT, atLeast(1), MEDIUM, SHARE_GROUP_MAX_SESSION_TIMEOUT_MS_DOC)
.define(SHARE_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG, INT, SHARE_GROUP_HEARTBEAT_INTERVAL_MS_DEFAULT, atLeast(1), MEDIUM, SHARE_GROUP_HEARTBEAT_INTERVAL_MS_DOC)
.define(SHARE_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, INT, SHARE_GROUP_MIN_HEARTBEAT_INTERVAL_MS_DEFAULT, atLeast(1), MEDIUM, SHARE_GROUP_MIN_HEARTBEAT_INTERVAL_MS_DOC)
.define(SHARE_GROUP_MAX_HEARTBEAT_INTERVAL_MS_CONFIG, INT, SHARE_GROUP_MAX_HEARTBEAT_INTERVAL_MS_DEFAULT, atLeast(1), MEDIUM, SHARE_GROUP_MAX_HEARTBEAT_INTERVAL_MS_DOC)
- .define(SHARE_GROUP_MAX_SIZE_CONFIG, INT, SHARE_GROUP_MAX_SIZE_DEFAULT, between(10, 1000), MEDIUM, SHARE_GROUP_MAX_SIZE_DOC);
+ .define(SHARE_GROUP_MAX_SIZE_CONFIG, INT, SHARE_GROUP_MAX_SIZE_DEFAULT, between(1, 1000), MEDIUM, SHARE_GROUP_MAX_SIZE_DOC);
/**
* The timeout used to wait for a new member in milliseconds.
@@ -287,14 +287,14 @@ public class GroupCoordinatorConfig {
private final int shareGroupMinHeartbeatIntervalMs;
private final int shareGroupMaxHeartbeatIntervalMs;
+ @SuppressWarnings("this-escape")
public GroupCoordinatorConfig(AbstractConfig config) {
this.numThreads = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_NUM_THREADS_CONFIG);
this.appendLingerMs = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG);
this.consumerGroupSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG);
this.consumerGroupHeartbeatIntervalMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG);
this.consumerGroupMaxSize = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG);
- this.consumerGroupAssignors = Collections.unmodifiableList(
- config.getConfiguredInstances(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, ConsumerGroupPartitionAssignor.class));
+ this.consumerGroupAssignors = consumerGroupAssignors(config);
this.offsetsTopicSegmentBytes = config.getInt(GroupCoordinatorConfig.OFFSETS_TOPIC_SEGMENT_BYTES_CONFIG);
this.offsetMetadataMaxSize = config.getInt(GroupCoordinatorConfig.OFFSET_METADATA_MAX_SIZE_CONFIG);
this.classicGroupMaxSize = config.getInt(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG);
@@ -367,6 +367,34 @@ public class GroupCoordinatorConfig {
SHARE_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG, SHARE_GROUP_SESSION_TIMEOUT_MS_CONFIG));
}
+ public static GroupCoordinatorConfig fromProps(
+ Map<?, ?> props
+ ) {
+ return new GroupCoordinatorConfig(
+ new AbstractConfig(
+ Utils.mergeConfigs(List.of(
+ GroupCoordinatorConfig.GROUP_COORDINATOR_CONFIG_DEF,
+ GroupCoordinatorConfig.NEW_GROUP_CONFIG_DEF,
+ GroupCoordinatorConfig.OFFSET_MANAGEMENT_CONFIG_DEF,
+ GroupCoordinatorConfig.CONSUMER_GROUP_CONFIG_DEF,
+ GroupCoordinatorConfig.SHARE_GROUP_CONFIG_DEF
+ )),
+ props
+ )
+ );
+ }
+
+ protected List<ConsumerGroupPartitionAssignor> consumerGroupAssignors(
+ AbstractConfig config
+ ) {
+ return Collections.unmodifiableList(
+ config.getConfiguredInstances(
+ GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG,
+ ConsumerGroupPartitionAssignor.class
+ )
+ );
+ }
+
/**
* Copy the subset of properties that are relevant to consumer group and share group.
*/
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
index 099755710b9..72a4ae58553 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
@@ -218,20 +218,8 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
.withTime(time)
.withTimer(timer)
.withExecutor(executor)
+ .withConfig(config)
.withGroupConfigManager(groupConfigManager)
- .withConsumerGroupAssignors(config.consumerGroupAssignors())
- .withConsumerGroupMaxSize(config.consumerGroupMaxSize())
- .withConsumerGroupSessionTimeout(config.consumerGroupSessionTimeoutMs())
- .withConsumerGroupHeartbeatInterval(config.consumerGroupHeartbeatIntervalMs())
- .withClassicGroupMaxSize(config.classicGroupMaxSize())
- .withClassicGroupInitialRebalanceDelayMs(config.classicGroupInitialRebalanceDelayMs())
- .withClassicGroupNewMemberJoinTimeoutMs(config.classicGroupNewMemberJoinTimeoutMs())
- .withClassicGroupMinSessionTimeoutMs(config.classicGroupMinSessionTimeoutMs())
- .withClassicGroupMaxSessionTimeoutMs(config.classicGroupMaxSessionTimeoutMs())
- .withConsumerGroupMigrationPolicy(config.consumerGroupMigrationPolicy())
- .withShareGroupMaxSize(config.shareGroupMaxSize())
- .withShareGroupSessionTimeout(config.shareGroupSessionTimeoutMs())
- .withShareGroupHeartbeatInterval(config.shareGroupHeartbeatIntervalMs())
.withGroupCoordinatorMetricsShard(metricsShard)
.build();
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 5a7e98ea3cc..e1fa86ec907 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
@@ -196,6 +196,7 @@ import static org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroupMe
* handling as well as during the initial loading of the records from the partitions.
*/
public class GroupMetadataManager {
+ private static final int METADATA_REFRESH_INTERVAL_MS = Integer.MAX_VALUE;
public static class Builder {
private LogContext logContext = null;
@@ -203,24 +204,10 @@ public class GroupMetadataManager {
private Time time = null;
private CoordinatorTimer<Void, CoordinatorRecord> timer = null;
private CoordinatorExecutor<CoordinatorRecord> executor = null;
- private List<ConsumerGroupPartitionAssignor> consumerGroupAssignors = null;
+ private GroupCoordinatorConfig config = null;
private GroupConfigManager groupConfigManager = null;
- private int consumerGroupMaxSize = Integer.MAX_VALUE;
- private int consumerGroupHeartbeatIntervalMs = 5000;
- private int consumerGroupMetadataRefreshIntervalMs = Integer.MAX_VALUE;
private MetadataImage metadataImage = null;
- private int consumerGroupSessionTimeoutMs = 45000;
- private int classicGroupMaxSize = Integer.MAX_VALUE;
- private int classicGroupInitialRebalanceDelayMs = 3000;
- private int classicGroupNewMemberJoinTimeoutMs = 5 * 60 * 1000;
- private int classicGroupMinSessionTimeoutMs;
- private int classicGroupMaxSessionTimeoutMs;
- private ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy;
private ShareGroupPartitionAssignor shareGroupAssignor = null;
- private int shareGroupMaxSize = Integer.MAX_VALUE;
- private int shareGroupHeartbeatIntervalMs = 5 * 1000;
- private int shareGroupSessionTimeoutMs = 45 * 1000;
- private int shareGroupMetadataRefreshIntervalMs = Integer.MAX_VALUE;
private GroupCoordinatorMetricsShard metrics;
Builder withLogContext(LogContext logContext) {
@@ -248,8 +235,8 @@ public class GroupMetadataManager {
return this;
}
- Builder withConsumerGroupAssignors(List<ConsumerGroupPartitionAssignor> consumerGroupAssignors) {
- this.consumerGroupAssignors = consumerGroupAssignors;
+ Builder withConfig(GroupCoordinatorConfig config) {
+ this.config = config;
return this;
}
@@ -258,61 +245,11 @@ public class GroupMetadataManager {
return this;
}
- Builder withConsumerGroupMaxSize(int consumerGroupMaxSize) {
- this.consumerGroupMaxSize = consumerGroupMaxSize;
- return this;
- }
-
- Builder withConsumerGroupSessionTimeout(int consumerGroupSessionTimeoutMs) {
- this.consumerGroupSessionTimeoutMs = consumerGroupSessionTimeoutMs;
- return this;
- }
-
- Builder withConsumerGroupHeartbeatInterval(int consumerGroupHeartbeatIntervalMs) {
- this.consumerGroupHeartbeatIntervalMs = consumerGroupHeartbeatIntervalMs;
- return this;
- }
-
- Builder withConsumerGroupMetadataRefreshIntervalMs(int consumerGroupMetadataRefreshIntervalMs) {
- this.consumerGroupMetadataRefreshIntervalMs = consumerGroupMetadataRefreshIntervalMs;
- return this;
- }
-
Builder withMetadataImage(MetadataImage metadataImage) {
this.metadataImage = metadataImage;
return this;
}
- Builder withClassicGroupMaxSize(int classicGroupMaxSize) {
- this.classicGroupMaxSize = classicGroupMaxSize;
- return this;
- }
-
- Builder withClassicGroupInitialRebalanceDelayMs(int classicGroupInitialRebalanceDelayMs) {
- this.classicGroupInitialRebalanceDelayMs = classicGroupInitialRebalanceDelayMs;
- return this;
- }
-
- Builder withClassicGroupNewMemberJoinTimeoutMs(int classicGroupNewMemberJoinTimeoutMs) {
- this.classicGroupNewMemberJoinTimeoutMs = classicGroupNewMemberJoinTimeoutMs;
- return this;
- }
-
- Builder withClassicGroupMinSessionTimeoutMs(int classicGroupMinSessionTimeoutMs) {
- this.classicGroupMinSessionTimeoutMs = classicGroupMinSessionTimeoutMs;
- return this;
- }
-
- Builder withClassicGroupMaxSessionTimeoutMs(int classicGroupMaxSessionTimeoutMs) {
- this.classicGroupMaxSessionTimeoutMs = classicGroupMaxSessionTimeoutMs;
- return this;
- }
-
- Builder withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy) {
- this.consumerGroupMigrationPolicy = consumerGroupMigrationPolicy;
- return this;
- }
-
Builder withGroupCoordinatorMetricsShard(GroupCoordinatorMetricsShard metrics) {
this.metrics = metrics;
return this;
@@ -323,26 +260,6 @@ public class GroupMetadataManager {
return this;
}
- public Builder withShareGroupMaxSize(int shareGroupMaxSize) {
- this.shareGroupMaxSize = shareGroupMaxSize;
- return this;
- }
-
- Builder withShareGroupSessionTimeout(int shareGroupSessionTimeoutMs) {
- this.shareGroupSessionTimeoutMs = shareGroupSessionTimeoutMs;
- return this;
- }
-
- Builder withShareGroupHeartbeatInterval(int shareGroupHeartbeatIntervalMs) {
- this.shareGroupHeartbeatIntervalMs = shareGroupHeartbeatIntervalMs;
- return this;
- }
-
- Builder withShareGroupMetadataRefreshIntervalMs(int shareGroupMetadataRefreshIntervalMs) {
- this.shareGroupMetadataRefreshIntervalMs = shareGroupMetadataRefreshIntervalMs;
- return this;
- }
-
GroupMetadataManager build() {
if (logContext == null) logContext = new LogContext();
if (snapshotRegistry == null) snapshotRegistry = new SnapshotRegistry(logContext);
@@ -353,8 +270,8 @@ public class GroupMetadataManager {
throw new IllegalArgumentException("Timer must be set.");
if (executor == null)
throw new IllegalArgumentException("Executor must be set.");
- if (consumerGroupAssignors == null || consumerGroupAssignors.isEmpty())
- throw new IllegalArgumentException("Assignors must be set before building.");
+ if (config == null)
+ throw new IllegalArgumentException("Config must be set.");
if (shareGroupAssignor == null)
shareGroupAssignor = new SimpleAssignor();
if (metrics == null)
@@ -369,24 +286,10 @@ public class GroupMetadataManager {
timer,
executor,
metrics,
- consumerGroupAssignors,
metadataImage,
+ config,
groupConfigManager,
- consumerGroupMaxSize,
- consumerGroupSessionTimeoutMs,
- consumerGroupHeartbeatIntervalMs,
- consumerGroupMetadataRefreshIntervalMs,
- classicGroupMaxSize,
- classicGroupInitialRebalanceDelayMs,
- classicGroupNewMemberJoinTimeoutMs,
- classicGroupMinSessionTimeoutMs,
- classicGroupMaxSessionTimeoutMs,
- consumerGroupMigrationPolicy,
- shareGroupAssignor,
- shareGroupMaxSize,
- shareGroupSessionTimeoutMs,
- shareGroupHeartbeatIntervalMs,
- shareGroupMetadataRefreshIntervalMs
+ shareGroupAssignor
);
}
}
@@ -432,6 +335,11 @@ public class GroupMetadataManager {
*/
private final GroupCoordinatorMetricsShard metrics;
+ /**
+ * The group coordinator config.
+ */
+ private final GroupCoordinatorConfig config;
+
/**
* The supported consumer group partition assignors keyed by their name.
*/
@@ -457,26 +365,6 @@ public class GroupMetadataManager {
*/
private final GroupConfigManager groupConfigManager;
- /**
- * The maximum number of members allowed in a single consumer group.
- */
- private final int consumerGroupMaxSize;
-
- /**
- * The default heartbeat interval for consumer groups.
- */
- private final int consumerGroupHeartbeatIntervalMs;
-
- /**
- * The default session timeout for consumer groups.
- */
- private final int consumerGroupSessionTimeoutMs;
-
- /**
- * The metadata refresh interval.
- */
- private final int consumerGroupMetadataRefreshIntervalMs;
-
/**
* The metadata image.
*/
@@ -497,61 +385,11 @@ public class GroupMetadataManager {
static final CoordinatorResult<Void, CoordinatorRecord> EMPTY_RESULT =
new CoordinatorResult<>(Collections.emptyList(), CompletableFuture.completedFuture(null), false);
- /**
- * The maximum number of members allowed in a single classic group.
- */
- private final int classicGroupMaxSize;
-
- /**
- * Initial rebalance delay for members joining a classic group.
- */
- private final int classicGroupInitialRebalanceDelayMs;
-
- /**
- * The timeout used to wait for a new member in milliseconds.
- */
- private final int classicGroupNewMemberJoinTimeoutMs;
-
- /**
- * The group minimum session timeout.
- */
- private final int classicGroupMinSessionTimeoutMs;
-
- /**
- * The group maximum session timeout.
- */
- private final int classicGroupMaxSessionTimeoutMs;
-
/**
* The share group partition assignor.
*/
private final ShareGroupPartitionAssignor shareGroupAssignor;
- /**
- * The maximum number of members allowed in a single share group.
- */
- private final int shareGroupMaxSize;
-
- /**
- * The heartbeat interval for share groups.
- */
- private final int shareGroupHeartbeatIntervalMs;
-
- /**
- * The session timeout for share groups.
- */
- private final int shareGroupSessionTimeoutMs;
-
- /**
- * The share group metadata refresh interval.
- */
- private final int shareGroupMetadataRefreshIntervalMs;
-
- /**
- * The config indicating whether group protocol upgrade/downgrade is allowed.
- */
- private final ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy;
-
private GroupMetadataManager(
SnapshotRegistry snapshotRegistry,
LogContext logContext,
@@ -559,24 +397,10 @@ public class GroupMetadataManager {
CoordinatorTimer<Void, CoordinatorRecord> timer,
CoordinatorExecutor<CoordinatorRecord> executor,
GroupCoordinatorMetricsShard metrics,
- List<ConsumerGroupPartitionAssignor> consumerGroupAssignors,
MetadataImage metadataImage,
+ GroupCoordinatorConfig config,
GroupConfigManager groupConfigManager,
- int consumerGroupMaxSize,
- int consumerGroupSessionTimeoutMs,
- int consumerGroupHeartbeatIntervalMs,
- int consumerGroupMetadataRefreshIntervalMs,
- int classicGroupMaxSize,
- int classicGroupInitialRebalanceDelayMs,
- int classicGroupNewMemberJoinTimeoutMs,
- int classicGroupMinSessionTimeoutMs,
- int classicGroupMaxSessionTimeoutMs,
- ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy,
- ShareGroupPartitionAssignor shareGroupAssignor,
- int shareGroupMaxSize,
- int shareGroupSessionTimeoutMs,
- int shareGroupHeartbeatIntervalMs,
- int shareGroupMetadataRefreshIntervalMs
+ ShareGroupPartitionAssignor shareGroupAssignor
) {
this.logContext = logContext;
this.log = logContext.logger(GroupMetadataManager.class);
@@ -585,27 +409,17 @@ public class GroupMetadataManager {
this.timer = timer;
this.executor = executor;
this.metrics = metrics;
+ this.config = config;
this.metadataImage = metadataImage;
- this.consumerGroupAssignors = consumerGroupAssignors.stream().collect(Collectors.toMap(ConsumerGroupPartitionAssignor::name, Function.identity()));
- this.defaultConsumerGroupAssignor = consumerGroupAssignors.get(0);
+ this.consumerGroupAssignors = config
+ .consumerGroupAssignors()
+ .stream()
+ .collect(Collectors.toMap(ConsumerGroupPartitionAssignor::name, Function.identity()));
+ this.defaultConsumerGroupAssignor = config.consumerGroupAssignors().get(0);
this.groups = new TimelineHashMap<>(snapshotRegistry, 0);
this.groupsByTopics = new TimelineHashMap<>(snapshotRegistry, 0);
this.groupConfigManager = groupConfigManager;
- this.consumerGroupMaxSize = consumerGroupMaxSize;
- this.consumerGroupSessionTimeoutMs = consumerGroupSessionTimeoutMs;
- this.consumerGroupHeartbeatIntervalMs = consumerGroupHeartbeatIntervalMs;
- this.consumerGroupMetadataRefreshIntervalMs = consumerGroupMetadataRefreshIntervalMs;
- this.classicGroupMaxSize = classicGroupMaxSize;
- this.classicGroupInitialRebalanceDelayMs = classicGroupInitialRebalanceDelayMs;
- this.classicGroupNewMemberJoinTimeoutMs = classicGroupNewMemberJoinTimeoutMs;
- this.classicGroupMinSessionTimeoutMs = classicGroupMinSessionTimeoutMs;
- this.classicGroupMaxSessionTimeoutMs = classicGroupMaxSessionTimeoutMs;
- this.consumerGroupMigrationPolicy = consumerGroupMigrationPolicy;
this.shareGroupAssignor = shareGroupAssignor;
- this.shareGroupMaxSize = shareGroupMaxSize;
- this.shareGroupSessionTimeoutMs = shareGroupSessionTimeoutMs;
- this.shareGroupHeartbeatIntervalMs = shareGroupHeartbeatIntervalMs;
- this.shareGroupMetadataRefreshIntervalMs = shareGroupMetadataRefreshIntervalMs;
}
/**
@@ -1093,11 +907,11 @@ public class GroupMetadataManager {
log.debug("Skip downgrading the consumer group {} to classic group because it's empty.",
consumerGroup.groupId());
return false;
- } else if (!consumerGroupMigrationPolicy.isDowngradeEnabled()) {
+ } else if (!config.consumerGroupMigrationPolicy().isDowngradeEnabled()) {
log.info("Cannot downgrade consumer group {} to classic group because the online downgrade is disabled.",
consumerGroup.groupId());
return false;
- } else if (consumerGroup.numMembers() - 1 > classicGroupMaxSize) {
+ } else if (consumerGroup.numMembers() - 1 > config.classicGroupMaxSize()) {
log.info("Cannot downgrade consumer group {} to classic group because its group size is greater than classic group max size.",
consumerGroup.groupId());
return false;
@@ -1120,11 +934,11 @@ public class GroupMetadataManager {
) {
if (!consumerGroup.allMembersUseClassicProtocolExcept(replacedMemberId)) {
return false;
- } else if (!consumerGroupMigrationPolicy.isDowngradeEnabled()) {
+ } else if (!config.consumerGroupMigrationPolicy().isDowngradeEnabled()) {
log.info("Cannot downgrade consumer group {} to classic group because the online downgrade is disabled.",
consumerGroup.groupId());
return false;
- } else if (consumerGroup.numMembers() > classicGroupMaxSize) {
+ } else if (consumerGroup.numMembers() > config.classicGroupMaxSize()) {
log.info("Cannot downgrade consumer group {} to classic group because its group size is greater than classic group max size.",
consumerGroup.groupId());
return false;
@@ -1191,7 +1005,7 @@ public class GroupMetadataManager {
* @return A boolean indicating whether it's valid to online upgrade the classic group.
*/
private boolean validateOnlineUpgrade(ClassicGroup classicGroup) {
- if (!consumerGroupMigrationPolicy.isUpgradeEnabled()) {
+ if (!config.consumerGroupMigrationPolicy().isUpgradeEnabled()) {
log.info("Cannot upgrade classic group {} to consumer group because the online upgrade is disabled.",
classicGroup.groupId());
return false;
@@ -1199,7 +1013,7 @@ public class GroupMetadataManager {
log.info("Cannot upgrade classic group {} to consumer group because the group does not use the consumer embedded protocol.",
classicGroup.groupId());
return false;
- } else if (classicGroup.numMembers() > consumerGroupMaxSize) {
+ } else if (classicGroup.numMembers() > config.consumerGroupMaxSize()) {
log.info("Cannot upgrade classic group {} to consumer group because the group size exceeds the consumer group maximum size.",
classicGroup.groupId());
return false;
@@ -1442,9 +1256,9 @@ public class GroupMetadataManager {
) throws GroupMaxSizeReachedException {
// If the consumer group has reached its maximum capacity, the member is rejected if it is not
// already a member of the consumer group.
- if (group.numMembers() >= consumerGroupMaxSize && (memberId.isEmpty() || !group.hasMember(memberId))) {
+ if (group.numMembers() >= config.consumerGroupMaxSize() && (memberId.isEmpty() || !group.hasMember(memberId))) {
throw new GroupMaxSizeReachedException("The consumer group has reached its maximum capacity of "
- + consumerGroupMaxSize + " members.");
+ + config.consumerGroupMaxSize() + " members.");
}
}
@@ -1463,9 +1277,9 @@ public class GroupMetadataManager {
) throws GroupMaxSizeReachedException {
// The member is rejected, if the share group has reached its maximum capacity, or it is not
// a member of the share group.
- if (group.numMembers() >= shareGroupMaxSize && (memberId.isEmpty() || !group.hasMember(memberId))) {
+ if (group.numMembers() >= config.shareGroupMaxSize() && (memberId.isEmpty() || !group.hasMember(memberId))) {
throw new GroupMaxSizeReachedException("The share group has reached its maximum capacity of "
- + shareGroupMaxSize + " members.");
+ + config.shareGroupMaxSize() + " members.");
}
}
@@ -1905,7 +1719,7 @@ public class GroupMetadataManager {
metrics.record(CONSUMER_GROUP_REBALANCES_SENSOR_NAME);
}
- group.setMetadataRefreshDeadline(currentTimeMs + consumerGroupMetadataRefreshIntervalMs, groupEpoch);
+ group.setMetadataRefreshDeadline(currentTimeMs + METADATA_REFRESH_INTERVAL_MS, groupEpoch);
}
// 2. Update the target assignment if the group epoch is larger than the target assignment epoch. The delta between
@@ -2098,7 +1912,7 @@ public class GroupMetadataManager {
metrics.record(CONSUMER_GROUP_REBALANCES_SENSOR_NAME);
}
- group.setMetadataRefreshDeadline(currentTimeMs + consumerGroupMetadataRefreshIntervalMs, groupEpoch);
+ group.setMetadataRefreshDeadline(currentTimeMs + METADATA_REFRESH_INTERVAL_MS, groupEpoch);
}
// 2. Update the target assignment if the group epoch is larger than the target assignment epoch. The delta between
@@ -2274,7 +2088,7 @@ public class GroupMetadataManager {
log.info("[GroupId {}] Bumped group epoch to {}.", groupId, groupEpoch);
}
- group.setMetadataRefreshDeadline(currentTimeMs + shareGroupMetadataRefreshIntervalMs, groupEpoch);
+ group.setMetadataRefreshDeadline(currentTimeMs + METADATA_REFRESH_INTERVAL_MS, groupEpoch);
}
// 2. Update the target assignment if the group epoch is larger than the target assignment epoch. The delta between
@@ -2735,7 +2549,7 @@ public class GroupMetadataManager {
log.info("[GroupId {}] Bumped group epoch to {}.", groupId, groupEpoch);
metrics.record(CONSUMER_GROUP_REBALANCES_SENSOR_NAME);
group.setMetadataRefreshDeadline(
- time.milliseconds() + consumerGroupMetadataRefreshIntervalMs,
+ time.milliseconds() + METADATA_REFRESH_INTERVAL_MS,
groupEpoch
);
}
@@ -4216,10 +4030,10 @@ public class GroupMetadataManager {
rescheduleClassicGroupMemberHeartbeat(classicGroup, member);
});
- if (classicGroup.numMembers() > classicGroupMaxSize) {
+ if (classicGroup.numMembers() > config.classicGroupMaxSize()) {
// In case the max size config has changed.
prepareRebalance(classicGroup, "Freshly-loaded group " + groupId +
- " (size " + classicGroup.numMembers() + ") is over capacity " + classicGroupMaxSize +
+ " (size " + classicGroup.numMembers() + ") is over capacity " + config.classicGroupMaxSize() +
". Rebalancing in order to give a chance for consumers to commit offsets");
}
@@ -5131,7 +4945,7 @@ public class GroupMetadataManager {
// timeout during a long rebalance), they may simply retry which will lead to a lot of defunct
// members in the rebalance. To prevent this going on indefinitely, we time out JoinGroup requests
// for new members. If the new member is still there, we expect it to retry.
- rescheduleClassicGroupMemberHeartbeat(group, member, classicGroupNewMemberJoinTimeoutMs);
+ rescheduleClassicGroupMemberHeartbeat(group, member, config.classicGroupNewMemberJoinTimeoutMs());
return maybePrepareRebalanceOrCompleteJoin(group, "Adding new member " + memberId + " with group instance id " +
request.groupInstanceId() + "; client reason: " + JoinGroupRequest.joinReason(request));
@@ -5182,8 +4996,8 @@ public class GroupMetadataManager {
boolean isInitialRebalance = group.isInState(EMPTY);
if (isInitialRebalance) {
// The group is new. Provide more time for the members to join.
- int delayMs = classicGroupInitialRebalanceDelayMs;
- int remainingMs = Math.max(group.rebalanceTimeoutMs() - classicGroupInitialRebalanceDelayMs, 0);
+ int delayMs = config.classicGroupInitialRebalanceDelayMs();
+ int remainingMs = Math.max(group.rebalanceTimeoutMs() - config.classicGroupInitialRebalanceDelayMs(), 0);
timer.schedule(
classicGroupJoinKey(group.groupId()),
@@ -5252,7 +5066,7 @@ public class GroupMetadataManager {
if (group.newMemberAdded() && remainingMs != 0) {
// A new member was added. Extend the delay.
group.setNewMemberAdded(false);
- int newDelayMs = Math.min(classicGroupInitialRebalanceDelayMs, remainingMs);
+ int newDelayMs = Math.min(config.classicGroupInitialRebalanceDelayMs(), remainingMs);
int newRemainingMs = Math.max(remainingMs - delayMs, 0);
timer.schedule(
@@ -5454,13 +5268,13 @@ public class GroupMetadataManager {
// 2) using the number of awaiting members allows to kick out the last rejoining
// members of the group.
return (group.hasMember(memberId) && group.member(memberId).isAwaitingJoin()) ||
- group.numAwaitingJoinResponse() < classicGroupMaxSize;
+ group.numAwaitingJoinResponse() < config.classicGroupMaxSize();
case COMPLETING_REBALANCE:
case STABLE:
// An existing member is accepted. New members are accepted up to the max group size.
// Note that the group size is used here. When the group transitions to CompletingRebalance,
// members who haven't rejoined are removed.
- return group.hasMember(memberId) || group.numMembers() < classicGroupMaxSize;
+ return group.hasMember(memberId) || group.numMembers() < config.classicGroupMaxSize();
default:
throw new IllegalStateException("Unknown group state: " + group.stateAsString());
}
@@ -6453,7 +6267,7 @@ public class GroupMetadataManager {
private int consumerGroupSessionTimeoutMs(String groupId) {
Optional<GroupConfig> groupConfig = groupConfigManager.groupConfig(groupId);
return groupConfig.map(GroupConfig::consumerSessionTimeoutMs)
- .orElse(consumerGroupSessionTimeoutMs);
+ .orElse(config.consumerGroupSessionTimeoutMs());
}
/**
@@ -6462,7 +6276,7 @@ public class GroupMetadataManager {
private int consumerGroupHeartbeatIntervalMs(String groupId) {
Optional<GroupConfig> groupConfig = groupConfigManager.groupConfig(groupId);
return groupConfig.map(GroupConfig::consumerHeartbeatIntervalMs)
- .orElse(consumerGroupHeartbeatIntervalMs);
+ .orElse(config.consumerGroupHeartbeatIntervalMs());
}
/**
@@ -6471,7 +6285,7 @@ public class GroupMetadataManager {
private int shareGroupSessionTimeoutMs(String groupId) {
Optional<GroupConfig> groupConfig = groupConfigManager.groupConfig(groupId);
return groupConfig.map(GroupConfig::shareSessionTimeoutMs)
- .orElse(shareGroupSessionTimeoutMs);
+ .orElse(config.shareGroupSessionTimeoutMs());
}
/**
@@ -6480,7 +6294,7 @@ public class GroupMetadataManager {
private int shareGroupHeartbeatIntervalMs(String groupId) {
Optional<GroupConfig> groupConfig = groupConfigManager.groupConfig(groupId);
return groupConfig.map(GroupConfig::shareHeartbeatIntervalMs)
- .orElse(shareGroupHeartbeatIntervalMs);
+ .orElse(config.shareGroupHeartbeatIntervalMs());
}
/**
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 c5bb26d4ba7..4cddd7db91e 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
@@ -168,7 +168,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
String memberId = Uuid.randomUuid().toString();
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.build();
Exception ex;
@@ -262,7 +262,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
assignor.prepareGroupAssignment(new GroupAssignment(Collections.emptyMap()));
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.build();
// Subscribing with an invalid regular expression fails.
@@ -306,7 +306,7 @@ public class GroupMetadataManagerTest {
.setMemberId(memberId)
.setMemberEpoch(1)
.setRebalanceTimeoutMs(5000)
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setTopicPartitions(Collections.emptyList()));
assertEquals(2, result.response().memberEpoch());
}
@@ -315,7 +315,7 @@ public class GroupMetadataManagerTest {
public void testMemberIdGeneration() {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(MetadataImage.EMPTY)
.build();
@@ -362,7 +362,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.build();
assertThrows(GroupIdNotFoundException.class, () ->
@@ -383,7 +383,7 @@ public class GroupMetadataManagerTest {
String memberId = Uuid.randomUuid().toString();
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new NoOpPartitionAssignor()))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(new NoOpPartitionAssignor()))
.build();
// A first member joins to create the group.
@@ -419,7 +419,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.build();
ConsumerGroupMember member = new ConsumerGroupMember.Builder(memberId)
@@ -486,7 +486,7 @@ public class GroupMetadataManagerTest {
.setMemberEpoch(99)
.setRebalanceTimeoutMs(5000)
.setSubscribedTopicNames(List.of("foo", "bar"))
- .setTopicPartitions(Collections.singletonList(new ConsumerGroupHeartbeatRequestData.TopicPartitions()
+ .setTopicPartitions(List.of(new ConsumerGroupHeartbeatRequestData.TopicPartitions()
.setTopicId(fooTopicId)
.setPartitions(List.of(1, 2)))));
assertEquals(100, result.response().memberEpoch());
@@ -505,7 +505,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -514,7 +514,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
- Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
+ Map.of(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5),
mkTopicAssignment(barTopicId, 0, 1, 2)
)))
@@ -595,7 +595,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -608,7 +608,7 @@ public class GroupMetadataManagerTest {
.setPreviousMemberEpoch(9)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setServerAssignorName("range")
.setAssignedPartitions(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)))
@@ -619,7 +619,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
- Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
+ Map.of(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5),
mkTopicAssignment(barTopicId, 0, 1, 2)
)))
@@ -695,7 +695,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -825,7 +825,7 @@ public class GroupMetadataManagerTest {
// Consumer group with two members.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -916,7 +916,7 @@ public class GroupMetadataManagerTest {
// Consumer group with two static members.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -1076,7 +1076,7 @@ public class GroupMetadataManagerTest {
// Consumer group with two static members.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -1148,7 +1148,7 @@ public class GroupMetadataManagerTest {
.setPartitions(List.of(3, 4, 5)),
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(barTopicId)
- .setPartitions(Collections.singletonList(2))
+ .setPartitions(List.of(2))
))),
rejoinResult.response()
);
@@ -1228,7 +1228,7 @@ public class GroupMetadataManagerTest {
.setRebalanceTimeoutMs(5000)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setServerAssignorName("range")
.setAssignedPartitions(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2)))
@@ -1241,7 +1241,7 @@ public class GroupMetadataManagerTest {
.setRebalanceTimeoutMs(5000)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setServerAssignorName("range")
.setAssignedPartitions(mkAssignment(
mkTopicAssignment(fooTopicId, 3, 4, 5)))
@@ -1249,7 +1249,7 @@ public class GroupMetadataManagerTest {
// Consumer group with two static members.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -1339,7 +1339,7 @@ public class GroupMetadataManagerTest {
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
.setRebalanceTimeoutMs(5000)
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setServerAssignorName("range")
.setAssignedPartitions(mkAssignment(
mkTopicAssignment(fooTopicId, 3, 4, 5)))
@@ -1437,7 +1437,7 @@ public class GroupMetadataManagerTest {
// Consumer group with two static members.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -1501,7 +1501,7 @@ public class GroupMetadataManagerTest {
// Consumer group with two static members.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -1592,7 +1592,7 @@ public class GroupMetadataManagerTest {
// Consumer group with one static member.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addRacks()
@@ -1642,7 +1642,7 @@ public class GroupMetadataManagerTest {
// Consumer group with one static member.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.build())
@@ -1690,7 +1690,7 @@ public class GroupMetadataManagerTest {
// Consumer group with one static member.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.build())
@@ -1732,7 +1732,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.build();
ConsumerGroupMember member = new ConsumerGroupMember.Builder(memberId)
@@ -1804,7 +1804,7 @@ public class GroupMetadataManagerTest {
.setMemberEpoch(99)
.setRebalanceTimeoutMs(5000)
.setSubscribedTopicNames(List.of("foo", "bar"))
- .setTopicPartitions(Collections.singletonList(new ConsumerGroupHeartbeatRequestData.TopicPartitions()
+ .setTopicPartitions(List.of(new ConsumerGroupHeartbeatRequestData.TopicPartitions()
.setTopicId(fooTopicId)
.setPartitions(List.of(1, 2)))));
assertEquals(100, result.response().memberEpoch());
@@ -1823,7 +1823,7 @@ public class GroupMetadataManagerTest {
// Consumer group with one static member.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.build())
@@ -1869,7 +1869,7 @@ public class GroupMetadataManagerTest {
// Consumer group with one static member.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.build())
@@ -1913,7 +1913,7 @@ public class GroupMetadataManagerTest {
// Create a context with an empty consumer group.
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 2)
.addRacks()
@@ -1944,7 +1944,7 @@ public class GroupMetadataManagerTest {
.setMemberEpoch(1)
.setHeartbeatIntervalMs(5000)
.setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
- .setTopicPartitions(Collections.singletonList(
+ .setTopicPartitions(List.of(
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(fooTopicId)
.setPartitions(List.of(0, 1))))),
@@ -1984,7 +1984,7 @@ public class GroupMetadataManagerTest {
.setMemberEpoch(1)
.setHeartbeatIntervalMs(5000)
.setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
- .setTopicPartitions(Collections.singletonList(
+ .setTopicPartitions(List.of(
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(fooTopicId)
.setPartitions(List.of(0, 1))))),
@@ -2008,7 +2008,7 @@ public class GroupMetadataManagerTest {
// Create a context with one consumer group containing two members.
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -2130,12 +2130,12 @@ public class GroupMetadataManagerTest {
.setPartitions(List.of(0, 1)),
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(barTopicId)
- .setPartitions(Collections.singletonList(0))
+ .setPartitions(List.of(0))
))),
result.response()
);
- assertRecordsEquals(Collections.singletonList(
+ assertRecordsEquals(List.of(
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, new ConsumerGroupMember.Builder(memberId1)
.setState(MemberState.UNREVOKED_PARTITIONS)
.setMemberEpoch(10)
@@ -2170,15 +2170,15 @@ public class GroupMetadataManagerTest {
.setTopicPartitions(List.of(
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(fooTopicId)
- .setPartitions(Collections.singletonList(3)),
+ .setPartitions(List.of(3)),
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(barTopicId)
- .setPartitions(Collections.singletonList(2))
+ .setPartitions(List.of(2))
))),
result.response()
);
- assertRecordsEquals(Collections.singletonList(
+ assertRecordsEquals(List.of(
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, new ConsumerGroupMember.Builder(memberId2)
.setState(MemberState.UNREVOKED_PARTITIONS)
.setMemberEpoch(10)
@@ -2210,7 +2210,7 @@ public class GroupMetadataManagerTest {
result.response()
);
- assertRecordsEquals(Collections.singletonList(
+ assertRecordsEquals(List.of(
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, new ConsumerGroupMember.Builder(memberId3)
.setState(MemberState.UNRELEASED_PARTITIONS)
.setMemberEpoch(11)
@@ -2235,7 +2235,7 @@ public class GroupMetadataManagerTest {
.setPartitions(List.of(0, 1)),
new ConsumerGroupHeartbeatRequestData.TopicPartitions()
.setTopicId(barTopicId)
- .setPartitions(Collections.singletonList(0))
+ .setPartitions(List.of(0))
)));
assertResponseEquals(
@@ -2246,7 +2246,7 @@ public class GroupMetadataManagerTest {
result.response()
);
- assertRecordsEquals(Collections.singletonList(
+ assertRecordsEquals(List.of(
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, new ConsumerGroupMember.Builder(memberId1)
.setState(MemberState.STABLE)
.setMemberEpoch(11)
@@ -2292,14 +2292,14 @@ public class GroupMetadataManagerTest {
.setMemberEpoch(11)
.setHeartbeatIntervalMs(5000)
.setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
- .setTopicPartitions(Collections.singletonList(
+ .setTopicPartitions(List.of(
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(barTopicId)
- .setPartitions(Collections.singletonList(1))))),
+ .setPartitions(List.of(1))))),
result.response()
);
- assertRecordsEquals(Collections.singletonList(
+ assertRecordsEquals(List.of(
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, new ConsumerGroupMember.Builder(memberId3)
.setState(MemberState.UNRELEASED_PARTITIONS)
.setMemberEpoch(11)
@@ -2342,10 +2342,10 @@ public class GroupMetadataManagerTest {
.setTopicPartitions(List.of(
new ConsumerGroupHeartbeatRequestData.TopicPartitions()
.setTopicId(fooTopicId)
- .setPartitions(Collections.singletonList(3)),
+ .setPartitions(List.of(3)),
new ConsumerGroupHeartbeatRequestData.TopicPartitions()
.setTopicId(barTopicId)
- .setPartitions(Collections.singletonList(2))
+ .setPartitions(List.of(2))
)));
assertResponseEquals(
@@ -2360,12 +2360,12 @@ public class GroupMetadataManagerTest {
.setPartitions(List.of(2, 3)),
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(barTopicId)
- .setPartitions(Collections.singletonList(2))
+ .setPartitions(List.of(2))
))),
result.response()
);
- assertRecordsEquals(Collections.singletonList(
+ assertRecordsEquals(List.of(
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, new ConsumerGroupMember.Builder(memberId2)
.setState(MemberState.STABLE)
.setMemberEpoch(11)
@@ -2386,10 +2386,10 @@ public class GroupMetadataManagerTest {
.setGroupId(groupId)
.setMemberId(memberId3)
.setMemberEpoch(11)
- .setTopicPartitions(Collections.singletonList(
+ .setTopicPartitions(List.of(
new ConsumerGroupHeartbeatRequestData.TopicPartitions()
.setTopicId(barTopicId)
- .setPartitions(Collections.singletonList(1)))));
+ .setPartitions(List.of(1)))));
assertResponseEquals(
new ConsumerGroupHeartbeatResponseData()
@@ -2403,11 +2403,11 @@ public class GroupMetadataManagerTest {
.setPartitions(List.of(4, 5)),
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(barTopicId)
- .setPartitions(Collections.singletonList(1))))),
+ .setPartitions(List.of(1))))),
result.response()
);
- assertRecordsEquals(Collections.singletonList(
+ assertRecordsEquals(List.of(
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, new ConsumerGroupMember.Builder(memberId3)
.setState(MemberState.STABLE)
.setMemberEpoch(11)
@@ -2439,12 +2439,12 @@ public class GroupMetadataManagerTest {
// Create a context with one consumer group containing two members.
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
.build())
- .withConsumerGroupMaxSize(2)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG, 2)
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
.withMember(new ConsumerGroupMember.Builder(memberId1)
.setState(MemberState.STABLE)
@@ -2502,7 +2502,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10))
.build();
@@ -2510,7 +2510,7 @@ public class GroupMetadataManagerTest {
context.replay(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, new ConsumerGroupMember.Builder(memberId1)
.setState(MemberState.STABLE)
- .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .setSubscribedTopicNames(List.of(fooTopicName))
.build()));
context.replay(GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 11));
@@ -2557,7 +2557,7 @@ public class GroupMetadataManagerTest {
when(assignor.assign(any(), any())).thenThrow(new PartitionAssignorException("Assignment failed."));
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -2591,8 +2591,7 @@ public class GroupMetadataManagerTest {
// Create a context with one consumer group containing one member.
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
- .withConsumerGroupMetadataRefreshIntervalMs(5 * 60 * 1000)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addRacks()
@@ -2627,7 +2626,7 @@ public class GroupMetadataManagerTest {
// Prepare the assignment result.
assignor.prepareGroupAssignment(new GroupAssignment(
- Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
+ Map.of(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)))
));
@@ -2646,7 +2645,7 @@ public class GroupMetadataManagerTest {
.setMemberEpoch(11)
.setHeartbeatIntervalMs(5000)
.setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
- .setTopicPartitions(Collections.singletonList(
+ .setTopicPartitions(List.of(
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(fooTopicId)
.setPartitions(List.of(0, 1, 2, 3, 4, 5))
@@ -2682,7 +2681,7 @@ public class GroupMetadataManagerTest {
// Check next refresh time.
assertFalse(consumerGroup.hasMetadataExpired(context.time.milliseconds()));
- assertEquals(context.time.milliseconds() + 5 * 60 * 1000, consumerGroup.metadataRefreshDeadline().deadlineMs);
+ assertEquals(context.time.milliseconds() + Integer.MAX_VALUE, consumerGroup.metadataRefreshDeadline().deadlineMs);
assertEquals(11, consumerGroup.metadataRefreshDeadline().epoch);
}
@@ -2698,8 +2697,7 @@ public class GroupMetadataManagerTest {
// Create a context with one consumer group containing one member.
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
- .withConsumerGroupMetadataRefreshIntervalMs(5 * 60 * 1000)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addRacks()
@@ -2734,7 +2732,7 @@ public class GroupMetadataManagerTest {
// Prepare the assignment result.
assignor.prepareGroupAssignment(new GroupAssignment(
- Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
+ Map.of(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)))
));
@@ -2748,7 +2746,7 @@ public class GroupMetadataManagerTest {
// The metadata refresh flag is set to a future time.
assertFalse(consumerGroup.hasMetadataExpired(context.time.milliseconds()));
- assertEquals(context.time.milliseconds() + 5 * 60 * 1000, consumerGroup.metadataRefreshDeadline().deadlineMs);
+ assertEquals(context.time.milliseconds() + Integer.MAX_VALUE, consumerGroup.metadataRefreshDeadline().deadlineMs);
assertEquals(11, consumerGroup.metadataRefreshDeadline().epoch);
// Rollback the uncommitted changes. This does not rollback the metadata flag
@@ -2771,7 +2769,7 @@ public class GroupMetadataManagerTest {
.setMemberEpoch(11)
.setHeartbeatIntervalMs(5000)
.setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
- .setTopicPartitions(Collections.singletonList(
+ .setTopicPartitions(List.of(
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(fooTopicId)
.setPartitions(List.of(0, 1, 2, 3, 4, 5))
@@ -2807,7 +2805,7 @@ public class GroupMetadataManagerTest {
// Check next refresh time.
assertFalse(consumerGroup.hasMetadataExpired(context.time.milliseconds()));
- assertEquals(context.time.milliseconds() + 5 * 60 * 1000, consumerGroup.metadataRefreshDeadline().deadlineMs);
+ assertEquals(context.time.milliseconds() + Integer.MAX_VALUE, consumerGroup.metadataRefreshDeadline().deadlineMs);
assertEquals(11, consumerGroup.metadataRefreshDeadline().epoch);
}
@@ -2818,7 +2816,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.build();
assertEquals(Collections.emptySet(), context.groupMetadataManager.groupsSubscribedToTopic("foo"));
@@ -2886,7 +2884,7 @@ public class GroupMetadataManagerTest {
// M2 in group 2 subscribes to foo.
context.replay(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId2,
new ConsumerGroupMember.Builder("group2-m2")
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.build()));
assertEquals(Set.of(groupId2), context.groupMetadataManager.groupsSubscribedToTopic("foo"));
@@ -2917,7 +2915,7 @@ public class GroupMetadataManagerTest {
@Test
public void testOnNewMetadataImageWithEmptyDelta() {
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(new MockPartitionAssignor("range")))
.build();
MetadataDelta delta = new MetadataDelta(MetadataImage.EMPTY);
@@ -2930,7 +2928,6 @@ public class GroupMetadataManagerTest {
@Test
public void testOnNewMetadataImage() {
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
.build();
// M1 in group 1 subscribes to a and b.
@@ -2948,19 +2945,19 @@ public class GroupMetadataManagerTest {
// M1 in group 3 subscribes to d.
context.replay(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord("group3",
new ConsumerGroupMember.Builder("group3-m1")
- .setSubscribedTopicNames(Collections.singletonList("d"))
+ .setSubscribedTopicNames(List.of("d"))
.build()));
// M1 in group 4 subscribes to e.
context.replay(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord("group4",
new ConsumerGroupMember.Builder("group4-m1")
- .setSubscribedTopicNames(Collections.singletonList("e"))
+ .setSubscribedTopicNames(List.of("e"))
.build()));
// M1 in group 5 subscribes to f.
context.replay(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord("group5",
new ConsumerGroupMember.Builder("group5-m1")
- .setSubscribedTopicNames(Collections.singletonList("f"))
+ .setSubscribedTopicNames(List.of("f"))
.build()));
// Ensures that all refresh flags are set to the future.
@@ -3006,7 +3003,7 @@ public class GroupMetadataManagerTest {
assertTrue(group.hasMetadataExpired(context.time.milliseconds()));
});
- Collections.singletonList("group5").forEach(groupId -> {
+ List.of("group5").forEach(groupId -> {
ConsumerGroup group = context.groupMetadataManager.consumerGroup(groupId);
assertFalse(group.hasMetadataExpired(context.time.milliseconds()));
});
@@ -3026,7 +3023,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addRacks()
@@ -3034,7 +3031,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
- Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
+ Map.of(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)))
));
@@ -3047,7 +3044,7 @@ public class GroupMetadataManagerTest {
.setMemberId(memberId)
.setMemberEpoch(0)
.setRebalanceTimeoutMs(90000)
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setTopicPartitions(Collections.emptyList()));
assertEquals(1, result.response().memberEpoch());
@@ -3101,7 +3098,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addRacks()
@@ -3109,7 +3106,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
- Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
+ Map.of(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)))
));
@@ -3122,7 +3119,7 @@ public class GroupMetadataManagerTest {
.setMemberId(memberId)
.setMemberEpoch(0)
.setRebalanceTimeoutMs(90000)
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setTopicPartitions(Collections.emptyList()));
assertEquals(1, result.response().memberEpoch());
@@ -3134,7 +3131,7 @@ public class GroupMetadataManagerTest {
// Verify the expired timeout.
assertEquals(
- Collections.singletonList(new ExpiredTimeout<Void, CoordinatorRecord>(
+ List.of(new ExpiredTimeout<Void, CoordinatorRecord>(
groupSessionTimeoutKey(groupId, memberId),
new CoordinatorResult<>(
List.of(
@@ -3165,7 +3162,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addRacks()
@@ -3173,7 +3170,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
- Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
+ Map.of(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)))
));
@@ -3187,7 +3184,7 @@ public class GroupMetadataManagerTest {
.setInstanceId(memberId)
.setMemberEpoch(0)
.setRebalanceTimeoutMs(90000)
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setTopicPartitions(Collections.emptyList()));
assertEquals(1, result.response().memberEpoch());
@@ -3202,7 +3199,7 @@ public class GroupMetadataManagerTest {
.setInstanceId(memberId)
.setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
.setRebalanceTimeoutMs(90000)
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setTopicPartitions(Collections.emptyList()));
assertEquals(-2, result.response().memberEpoch());
@@ -3215,7 +3212,7 @@ public class GroupMetadataManagerTest {
// Verify the expired timeout.
assertEquals(
- Collections.singletonList(new ExpiredTimeout<Void, CoordinatorRecord>(
+ List.of(new ExpiredTimeout<Void, CoordinatorRecord>(
groupSessionTimeoutKey(groupId, memberId),
new CoordinatorResult<>(
List.of(
@@ -3247,7 +3244,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 3)
.addRacks()
@@ -3266,7 +3263,7 @@ public class GroupMetadataManagerTest {
.setMemberId(memberId1)
.setMemberEpoch(0)
.setRebalanceTimeoutMs(180000)
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setTopicPartitions(Collections.emptyList()));
assertResponseEquals(
@@ -3275,7 +3272,7 @@ public class GroupMetadataManagerTest {
.setMemberEpoch(1)
.setHeartbeatIntervalMs(5000)
.setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
- .setTopicPartitions(Collections.singletonList(
+ .setTopicPartitions(List.of(
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(fooTopicId)
.setPartitions(List.of(0, 1, 2))))),
@@ -3304,7 +3301,7 @@ public class GroupMetadataManagerTest {
.setMemberId(memberId2)
.setMemberEpoch(0)
.setRebalanceTimeoutMs(90000)
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setTopicPartitions(Collections.emptyList()));
assertResponseEquals(
@@ -3329,7 +3326,7 @@ public class GroupMetadataManagerTest {
.setMemberId(memberId1)
.setMemberEpoch(1)
.setRebalanceTimeoutMs(12000)
- .setSubscribedTopicNames(Collections.singletonList("foo")));
+ .setSubscribedTopicNames(List.of("foo")));
assertResponseEquals(
new ConsumerGroupHeartbeatResponseData()
@@ -3337,7 +3334,7 @@ public class GroupMetadataManagerTest {
.setMemberEpoch(1)
.setHeartbeatIntervalMs(5000)
.setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
- .setTopicPartitions(Collections.singletonList(
+ .setTopicPartitions(List.of(
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(fooTopicId)
.setPartitions(List.of(0, 1))))),
@@ -3360,7 +3357,7 @@ public class GroupMetadataManagerTest {
.setGroupId(groupId)
.setMemberId(memberId1)
.setMemberEpoch(1)
- .setTopicPartitions(Collections.singletonList(new ConsumerGroupHeartbeatRequestData.TopicPartitions()
+ .setTopicPartitions(List.of(new ConsumerGroupHeartbeatRequestData.TopicPartitions()
.setTopicId(fooTopicId)
.setPartitions(List.of(0, 1)))));
@@ -3392,7 +3389,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 3)
.addRacks()
@@ -3411,7 +3408,7 @@ public class GroupMetadataManagerTest {
.setMemberId(memberId1)
.setMemberEpoch(0)
.setRebalanceTimeoutMs(10000) // Use timeout smaller than session timeout.
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setTopicPartitions(Collections.emptyList()));
assertResponseEquals(
@@ -3420,7 +3417,7 @@ public class GroupMetadataManagerTest {
.setMemberEpoch(1)
.setHeartbeatIntervalMs(5000)
.setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
- .setTopicPartitions(Collections.singletonList(
+ .setTopicPartitions(List.of(
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(fooTopicId)
.setPartitions(List.of(0, 1, 2))))),
@@ -3449,7 +3446,7 @@ public class GroupMetadataManagerTest {
.setMemberId(memberId2)
.setMemberEpoch(0)
.setRebalanceTimeoutMs(10000)
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setTopicPartitions(Collections.emptyList()));
assertResponseEquals(
@@ -3480,7 +3477,7 @@ public class GroupMetadataManagerTest {
.setMemberEpoch(1)
.setHeartbeatIntervalMs(5000)
.setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
- .setTopicPartitions(Collections.singletonList(
+ .setTopicPartitions(List.of(
new ConsumerGroupHeartbeatResponseData.TopicPartitions()
.setTopicId(fooTopicId)
.setPartitions(List.of(0, 1))))),
@@ -3492,7 +3489,7 @@ public class GroupMetadataManagerTest {
// Verify the expired timeout.
assertEquals(
- Collections.singletonList(new ExpiredTimeout<Void, CoordinatorRecord>(
+ List.of(new ExpiredTimeout<Void, CoordinatorRecord>(
consumerGroupRebalanceTimeoutKey(groupId, memberId1),
new CoordinatorResult<>(
List.of(
@@ -3519,7 +3516,6 @@ public class GroupMetadataManagerTest {
String barTopicName = "bar";
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -3531,7 +3527,7 @@ public class GroupMetadataManagerTest {
.setPreviousMemberEpoch(9)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setServerAssignorName("range")
.setAssignedPartitions(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2)))
@@ -3544,7 +3540,7 @@ public class GroupMetadataManagerTest {
.setPreviousMemberEpoch(10)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setServerAssignorName("range")
.build())
.withAssignment("foo-1", mkAssignment(
@@ -3621,7 +3617,7 @@ public class GroupMetadataManagerTest {
ClassicGroup group = context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", false);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newEmptyGroupMetadataRecord(group, MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newEmptyGroupMetadataRecord(group, MetadataVersion.latestTesting())),
joinResult.records
);
}
@@ -3654,7 +3650,7 @@ public class GroupMetadataManagerTest {
.build();
byte[] subscription = ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList("foo"))).array();
+ List.of("foo"))).array();
List<GroupMetadataValue.MemberMetadata> members = new ArrayList<>();
List<ClassicGroupMember> expectedMembers = new ArrayList<>();
JoinGroupRequestProtocolCollection expectedProtocols = new JoinGroupRequestProtocolCollection(0);
@@ -3726,11 +3722,11 @@ public class GroupMetadataManagerTest {
@Test
public void testOnLoadedExceedGroupMaxSizeTriggersRebalance() {
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withClassicGroupMaxSize(1)
+ .withConfig(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG, 1)
.build();
byte[] subscription = ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList("foo"))).array();
+ List.of("foo"))).array();
List<GroupMetadataValue.MemberMetadata> members = new ArrayList<>();
IntStream.range(0, 2).forEach(i -> members.add(
@@ -3769,7 +3765,7 @@ public class GroupMetadataManagerTest {
.build();
byte[] subscription = ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList("foo"))).array();
+ List.of("foo"))).array();
List<GroupMetadataValue.MemberMetadata> members = new ArrayList<>();
IntStream.range(0, 2).forEach(i -> members.add(
@@ -3810,7 +3806,7 @@ public class GroupMetadataManagerTest {
@Test
public void testJoinGroupShouldReceiveErrorIfGroupOverMaxSize() throws Exception {
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withClassicGroupMaxSize(10)
+ .withConfig(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG, 10)
.build();
context.createClassicGroup("group-id");
@@ -3838,8 +3834,8 @@ public class GroupMetadataManagerTest {
boolean requiredKnownMemberId = true;
int groupMaxSize = 10;
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withClassicGroupMaxSize(groupMaxSize)
- .withClassicGroupInitialRebalanceDelayMs(50)
+ .withConfig(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG, groupMaxSize)
+ .withConfig(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, 50)
.build();
ClassicGroup group = context.createClassicGroup("group-id");
@@ -3897,8 +3893,8 @@ public class GroupMetadataManagerTest {
boolean requiredKnownMemberId = false;
int groupMaxSize = 10;
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withClassicGroupMaxSize(groupMaxSize)
- .withClassicGroupInitialRebalanceDelayMs(50)
+ .withConfig(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG, groupMaxSize)
+ .withConfig(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, 50)
.build();
ClassicGroup group = context.createClassicGroup("group-id");
@@ -3950,8 +3946,8 @@ public class GroupMetadataManagerTest {
.collect(Collectors.toList());
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withClassicGroupMaxSize(groupMaxSize)
- .withClassicGroupInitialRebalanceDelayMs(50)
+ .withConfig(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG, groupMaxSize)
+ .withConfig(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, 50)
.build();
ClassicGroup group = context.createClassicGroup("group-id");
@@ -4000,8 +3996,8 @@ public class GroupMetadataManagerTest {
boolean requiredKnownMemberId = true;
int groupMaxSize = 10;
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withClassicGroupMaxSize(groupMaxSize)
- .withClassicGroupInitialRebalanceDelayMs(50)
+ .withConfig(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG, groupMaxSize)
+ .withConfig(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, 50)
.build();
ClassicGroup group = context.createClassicGroup("group-id");
@@ -4063,8 +4059,8 @@ public class GroupMetadataManagerTest {
int groupMaxSize = 10;
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withClassicGroupMaxSize(groupMaxSize)
- .withClassicGroupInitialRebalanceDelayMs(50)
+ .withConfig(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG, groupMaxSize)
+ .withConfig(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, 50)
.build();
// Create a group and add members that exceed the group max size.
@@ -4666,7 +4662,7 @@ public class GroupMetadataManagerTest {
public void testHeartbeatExpirationShouldRemoveMember() throws Exception {
// Set initial rebalance delay to simulate a long running rebalance.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withClassicGroupInitialRebalanceDelayMs(10 * 60 * 1000)
+ .withConfig(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, 10 * 60 * 1000)
.build();
ClassicGroup group = context.createClassicGroup("group-id");
@@ -4691,7 +4687,7 @@ public class GroupMetadataManagerTest {
timeouts.forEach(timeout -> {
assertEquals(classicGroupHeartbeatKey("group-id", memberId), timeout.key);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
timeout.result.records()
);
});
@@ -5171,7 +5167,7 @@ public class GroupMetadataManagerTest {
);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
joinResult.records
);
assertFalse(joinResult.joinFuture.isDone());
@@ -5194,7 +5190,7 @@ public class GroupMetadataManagerTest {
if (supportSkippingAssignment) {
expectedResponse
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new JoinGroupResponseData.JoinGroupResponseMember()
.setMemberId(newMemberId)
.setGroupInstanceId("group-instance-id")
@@ -5286,7 +5282,7 @@ public class GroupMetadataManagerTest {
protocols.add(new JoinGroupRequestProtocol()
.setName("roundrobin")
.setMetadata(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList("bar"))).array()));
+ List.of("bar"))).array()));
GroupMetadataManagerTestContext.JoinResult joinResult = context.sendClassicGroupJoin(
request
@@ -5298,7 +5294,7 @@ public class GroupMetadataManagerTest {
);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
joinResult.records
);
assertFalse(joinResult.joinFuture.isDone());
@@ -5374,7 +5370,7 @@ public class GroupMetadataManagerTest {
supportSkippingAssignment);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
joinResult.records
);
assertFalse(joinResult.joinFuture.isDone());
@@ -5503,7 +5499,7 @@ public class GroupMetadataManagerTest {
// Member should be removed as heartbeat expires. The group is now empty.
List<ExpiredTimeout<Void, CoordinatorRecord>> timeouts = context.sleep(5000);
- List<CoordinatorRecord> expectedRecords = Collections.singletonList(GroupMetadataManagerTestContext.newGroupMetadataRecord(
+ List<CoordinatorRecord> expectedRecords = List.of(GroupMetadataManagerTestContext.newGroupMetadataRecord(
group.groupId(),
new GroupMetadataValue()
.setMembers(Collections.emptyList())
@@ -5942,7 +5938,7 @@ public class GroupMetadataManagerTest {
);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
joinResult.records
);
// Simulate a successful write to the log.
@@ -6046,7 +6042,7 @@ public class GroupMetadataManagerTest {
joinResponse,
group,
COMPLETING_REBALANCE,
- Collections.singleton("leader-instance-id")
+ Set.of("leader-instance-id")
);
}
@@ -6255,7 +6251,7 @@ public class GroupMetadataManagerTest {
);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
followerJoinResult.records
);
// Simulate a failed write to the log.
@@ -6312,7 +6308,7 @@ public class GroupMetadataManagerTest {
leaderSyncResult.appendFuture.complete(null);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
leaderSyncResult.records
);
@@ -6362,7 +6358,7 @@ public class GroupMetadataManagerTest {
);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
followerJoinResult.records
);
@@ -6543,7 +6539,7 @@ public class GroupMetadataManagerTest {
followerJoinResult.joinFuture.get(),
group,
COMPLETING_REBALANCE,
- Collections.singleton("follower-instance-id")
+ Set.of("follower-instance-id")
);
}
@@ -6574,7 +6570,7 @@ public class GroupMetadataManagerTest {
);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
followerJoinResult.records
);
// Simulate a successful write to log.
@@ -6781,7 +6777,7 @@ public class GroupMetadataManagerTest {
);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
leaderJoinResult.records
);
// Simulate a successful write to log.
@@ -7561,7 +7557,7 @@ public class GroupMetadataManagerTest {
));
assertEquals(
- Collections.singletonList(
+ List.of(
GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, updatedAssignment, MetadataVersion.latestTesting())),
syncResult.records
);
@@ -8221,7 +8217,7 @@ public class GroupMetadataManagerTest {
ExpiredTimeout<Void, CoordinatorRecord> timeout = timeouts.get(0);
assertEquals(classicGroupSyncKey("group-id"), timeout.key);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
timeout.result.records()
);
@@ -8377,7 +8373,7 @@ public class GroupMetadataManagerTest {
if (response.memberId().equals(leaderId)) {
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
syncResult.records
);
@@ -8458,7 +8454,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withShareGroupAssignor(assignor)
.withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
.build();
@@ -8479,7 +8475,7 @@ public class GroupMetadataManagerTest {
context.commit();
ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(classicGroupId, false);
context.replay(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(consumerGroupId, new ConsumerGroupMember.Builder(memberId1)
- .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .setSubscribedTopicNames(List.of(fooTopicName))
.build()));
context.replay(GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(consumerGroupId, 11));
@@ -8511,7 +8507,7 @@ public class GroupMetadataManagerTest {
// List group with case-insensitive ‘empty’.
actualAllGroupMap =
- context.sendListGroups(Collections.singletonList("empty"), Collections.emptyList())
+ context.sendListGroups(List.of("empty"), Collections.emptyList())
.stream().collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
assertEquals(expectAllGroupMap, actualAllGroupMap);
@@ -8519,7 +8515,7 @@ public class GroupMetadataManagerTest {
context.commit();
// Test list group response to check assigning state in the consumer group.
- actualAllGroupMap = context.sendListGroups(Collections.singletonList("assigning"), Collections.emptyList()).stream()
+ actualAllGroupMap = context.sendListGroups(List.of("assigning"), Collections.emptyList()).stream()
.collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
expectAllGroupMap =
Stream.of(
@@ -8533,7 +8529,7 @@ public class GroupMetadataManagerTest {
assertEquals(expectAllGroupMap, actualAllGroupMap);
// Test list group response with group state filter and no group type filter.
- actualAllGroupMap = context.sendListGroups(Collections.singletonList("Empty"), Collections.emptyList()).stream()
+ actualAllGroupMap = context.sendListGroups(List.of("Empty"), Collections.emptyList()).stream()
.collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
expectAllGroupMap = Stream.of(
new ListGroupsResponseData.ListedGroup()
@@ -8551,7 +8547,7 @@ public class GroupMetadataManagerTest {
assertEquals(expectAllGroupMap, actualAllGroupMap);
// Test list group response with no group state filter and with group type filter.
- actualAllGroupMap = context.sendListGroups(Collections.emptyList(), Collections.singletonList(Group.GroupType.CLASSIC.toString())).stream()
+ actualAllGroupMap = context.sendListGroups(Collections.emptyList(), List.of(Group.GroupType.CLASSIC.toString())).stream()
.collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
expectAllGroupMap = Stream.of(
new ListGroupsResponseData.ListedGroup()
@@ -8564,7 +8560,7 @@ public class GroupMetadataManagerTest {
assertEquals(expectAllGroupMap, actualAllGroupMap);
// Test list group response with no group state filter and with group type filter in a different case.
- actualAllGroupMap = context.sendListGroups(Collections.emptyList(), Collections.singletonList("Consumer")).stream()
+ actualAllGroupMap = context.sendListGroups(Collections.emptyList(), List.of("Consumer")).stream()
.collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
expectAllGroupMap = Stream.of(
new ListGroupsResponseData.ListedGroup()
@@ -8576,7 +8572,7 @@ public class GroupMetadataManagerTest {
assertEquals(expectAllGroupMap, actualAllGroupMap);
- actualAllGroupMap = context.sendListGroups(Collections.emptyList(), Collections.singletonList("Share")).stream()
+ actualAllGroupMap = context.sendListGroups(Collections.emptyList(), List.of("Share")).stream()
.collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
expectAllGroupMap = Stream.of(
new ListGroupsResponseData.ListedGroup()
@@ -8611,14 +8607,14 @@ public class GroupMetadataManagerTest {
assertEquals(expectAllGroupMap, actualAllGroupMap);
// Test list group response with no group state filter and with invalid group type filter .
- actualAllGroupMap = context.sendListGroups(Collections.emptyList(), Collections.singletonList("Invalid")).stream()
+ actualAllGroupMap = context.sendListGroups(Collections.emptyList(), List.of("Invalid")).stream()
.collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
expectAllGroupMap = Collections.emptyMap();
assertEquals(expectAllGroupMap, actualAllGroupMap);
// Test list group response with invalid group state filter and with no group type filter .
- actualAllGroupMap = context.sendListGroups(Collections.singletonList("Invalid"), Collections.emptyList()).stream()
+ actualAllGroupMap = context.sendListGroups(List.of("Invalid"), Collections.emptyList()).stream()
.collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
expectAllGroupMap = Collections.emptyMap();
@@ -8632,12 +8628,12 @@ public class GroupMetadataManagerTest {
String memberId = "member-id";
String topicName = "topicName";
ConsumerGroupMember.Builder memberBuilder = new ConsumerGroupMember.Builder(memberId)
- .setSubscribedTopicNames(Collections.singletonList(topicName))
+ .setSubscribedTopicNames(List.of(topicName))
.setServerAssignorName("assignorName");
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withConsumerGroup(new ConsumerGroupBuilder(consumerGroupIds.get(0), epoch))
.withConsumerGroup(new ConsumerGroupBuilder(consumerGroupIds.get(1), epoch)
.withMember(memberBuilder.build()))
@@ -8652,7 +8648,7 @@ public class GroupMetadataManagerTest {
new ConsumerGroupDescribeResponseData.DescribedGroup()
.setGroupEpoch(epoch)
.setGroupId(consumerGroupIds.get(1))
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
memberBuilder.build().asConsumerGroupDescribeMember(
new Assignment(Collections.emptyMap()),
new MetadataImageBuilder().build().topics()
@@ -8672,14 +8668,14 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.build();
- List<ConsumerGroupDescribeResponseData.DescribedGroup> actual = context.sendConsumerGroupDescribe(Collections.singletonList(groupId));
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> actual = context.sendConsumerGroupDescribe(List.of(groupId));
ConsumerGroupDescribeResponseData.DescribedGroup describedGroup = new ConsumerGroupDescribeResponseData.DescribedGroup()
.setGroupId(groupId)
.setErrorCode(Errors.GROUP_ID_NOT_FOUND.code());
- List<ConsumerGroupDescribeResponseData.DescribedGroup> expected = Collections.singletonList(
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> expected = List.of(
describedGroup
);
@@ -8700,12 +8696,12 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(metadataImage)
.build();
ConsumerGroupMember.Builder memberBuilder1 = new ConsumerGroupMember.Builder(memberId1)
- .setSubscribedTopicNames(Collections.singletonList(topicName));
+ .setSubscribedTopicNames(List.of(topicName));
context.replay(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(consumerGroupId, memberBuilder1.build()));
context.replay(GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(consumerGroupId, epoch + 1));
@@ -8717,11 +8713,11 @@ public class GroupMetadataManagerTest {
context.replay(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(consumerGroupId, memberBuilder2.build()));
context.replay(GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(consumerGroupId, epoch + 2));
- List<ConsumerGroupDescribeResponseData.DescribedGroup> actual = context.groupMetadataManager.consumerGroupDescribe(Collections.singletonList(consumerGroupId), context.lastCommittedOffset);
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> actual = context.groupMetadataManager.consumerGroupDescribe(List.of(consumerGroupId), context.lastCommittedOffset);
ConsumerGroupDescribeResponseData.DescribedGroup describedGroup = new ConsumerGroupDescribeResponseData.DescribedGroup()
.setGroupId(consumerGroupId)
.setErrorCode(Errors.GROUP_ID_NOT_FOUND.code());
- List<ConsumerGroupDescribeResponseData.DescribedGroup> expected = Collections.singletonList(
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> expected = List.of(
describedGroup
);
assertEquals(expected, actual);
@@ -8729,7 +8725,7 @@ public class GroupMetadataManagerTest {
// Commit the offset and test again
context.commit();
- actual = context.groupMetadataManager.consumerGroupDescribe(Collections.singletonList(consumerGroupId), context.lastCommittedOffset);
+ actual = context.groupMetadataManager.consumerGroupDescribe(List.of(consumerGroupId), context.lastCommittedOffset);
describedGroup = new ConsumerGroupDescribeResponseData.DescribedGroup()
.setGroupId(consumerGroupId)
.setMembers(List.of(
@@ -8739,7 +8735,7 @@ public class GroupMetadataManagerTest {
.setGroupState(ConsumerGroup.ConsumerGroupState.ASSIGNING.toString())
.setAssignorName("range")
.setGroupEpoch(epoch + 2);
- expected = Collections.singletonList(
+ expected = List.of(
describedGroup
);
assertEquals(expected, actual);
@@ -8758,7 +8754,7 @@ public class GroupMetadataManagerTest {
.setAssignment(new byte[]{0})
.setSubscription(new byte[]{0, 1, 2});
GroupMetadataValue groupMetadataValue = new GroupMetadataValue()
- .setMembers(Collections.singletonList(memberMetadata))
+ .setMembers(List.of(memberMetadata))
.setProtocolType("consumer")
.setProtocol("range")
.setCurrentStateTimestamp(context.time.milliseconds());
@@ -8771,13 +8767,13 @@ public class GroupMetadataManagerTest {
context.verifyDescribeGroupsReturnsDeadGroup("group-id");
context.commit();
- List<DescribeGroupsResponseData.DescribedGroup> expectedDescribedGroups = Collections.singletonList(
+ List<DescribeGroupsResponseData.DescribedGroup> expectedDescribedGroups = List.of(
new DescribeGroupsResponseData.DescribedGroup()
.setGroupId("group-id")
.setGroupState(STABLE.toString())
.setProtocolType(groupMetadataValue.protocolType())
.setProtocolData(groupMetadataValue.protocol())
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new DescribeGroupsResponseData.DescribedGroupMember()
.setMemberId(memberMetadata.memberId())
.setGroupInstanceId(memberMetadata.groupInstanceId())
@@ -8789,7 +8785,7 @@ public class GroupMetadataManagerTest {
);
List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
- context.describeGroups(Collections.singletonList("group-id"));
+ context.describeGroups(List.of("group-id"));
assertEquals(expectedDescribedGroups, describedGroups);
}
@@ -8807,7 +8803,7 @@ public class GroupMetadataManagerTest {
.setAssignment(new byte[]{0})
.setSubscription(new byte[]{0, 1, 2});
GroupMetadataValue groupMetadataValue = new GroupMetadataValue()
- .setMembers(Collections.singletonList(memberMetadata))
+ .setMembers(List.of(memberMetadata))
.setProtocolType("consumer")
.setProtocol("range")
.setCurrentStateTimestamp(context.time.milliseconds());
@@ -8823,13 +8819,13 @@ public class GroupMetadataManagerTest {
context.verifyDescribeGroupsReturnsDeadGroup("group-id");
context.commit();
- List<DescribeGroupsResponseData.DescribedGroup> expectedDescribedGroups = Collections.singletonList(
+ List<DescribeGroupsResponseData.DescribedGroup> expectedDescribedGroups = List.of(
new DescribeGroupsResponseData.DescribedGroup()
.setGroupId("group-id")
.setGroupState(PREPARING_REBALANCE.toString())
.setProtocolType(groupMetadataValue.protocolType())
.setProtocolData("")
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new DescribeGroupsResponseData.DescribedGroupMember()
.setMemberId(memberMetadata.memberId())
.setGroupInstanceId(memberMetadata.groupInstanceId())
@@ -8840,7 +8836,7 @@ public class GroupMetadataManagerTest {
);
List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
- context.describeGroups(Collections.singletonList("group-id"));
+ context.describeGroups(List.of("group-id"));
assertEquals(expectedDescribedGroups, describedGroups);
}
@@ -8903,7 +8899,7 @@ public class GroupMetadataManagerTest {
CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> leaveResult = context.sendClassicGroupLeave(
new LeaveGroupRequestData()
.setGroupId("group-id")
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new MemberIdentity()
.setMemberId(rebalanceResult.followerId)
.setGroupInstanceId("follower-instance-id")
@@ -8911,7 +8907,7 @@ public class GroupMetadataManagerTest {
);
LeaveGroupResponseData expectedResponse = new LeaveGroupResponseData()
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new LeaveGroupResponseData.MemberResponse()
.setMemberId(rebalanceResult.followerId)
.setGroupInstanceId("follower-instance-id")));
@@ -8938,14 +8934,14 @@ public class GroupMetadataManagerTest {
CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> leaveResult = context.sendClassicGroupLeave(
new LeaveGroupRequestData()
.setGroupId("group-id")
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new MemberIdentity()
.setMemberId(pendingJoinResponse.memberId())
))
);
LeaveGroupResponseData expectedResponse = new LeaveGroupResponseData()
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new LeaveGroupResponseData.MemberResponse()
.setGroupInstanceId(null)
.setMemberId(pendingJoinResponse.memberId())));
@@ -8979,7 +8975,7 @@ public class GroupMetadataManagerTest {
assertThrows(UnknownMemberIdException.class, () -> context.sendClassicGroupLeave(
new LeaveGroupRequestData()
.setGroupId("unknown-group-id")
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new MemberIdentity()
.setMemberId("member-id")
))
@@ -9003,14 +8999,14 @@ public class GroupMetadataManagerTest {
CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> leaveResult = context.sendClassicGroupLeave(
new LeaveGroupRequestData()
.setGroupId("group-id")
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new MemberIdentity()
.setMemberId("unknown-member-id")
))
);
LeaveGroupResponseData expectedResponse = new LeaveGroupResponseData()
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new LeaveGroupResponseData.MemberResponse()
.setGroupInstanceId(null)
.setMemberId("unknown-member-id")
@@ -9030,7 +9026,7 @@ public class GroupMetadataManagerTest {
CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> leaveResult = context.sendClassicGroupLeave(
new LeaveGroupRequestData()
.setGroupId("group-id")
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new MemberIdentity()
.setMemberId("member-id")
))
@@ -9061,20 +9057,20 @@ public class GroupMetadataManagerTest {
CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> leaveResult = context.sendClassicGroupLeave(
new LeaveGroupRequestData()
.setGroupId("group-id")
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new MemberIdentity()
.setMemberId(joinResponse.memberId())
))
);
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
leaveResult.records()
);
// Simulate a successful write to the log.
leaveResult.appendFuture().complete(null);
LeaveGroupResponseData expectedResponse = new LeaveGroupResponseData()
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new LeaveGroupResponseData.MemberResponse()
.setGroupInstanceId(null)
.setMemberId(joinResponse.memberId())));
@@ -9104,7 +9100,7 @@ public class GroupMetadataManagerTest {
CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> leaveResult = context.sendClassicGroupLeave(
new LeaveGroupRequestData()
.setGroupId("group-id")
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new MemberIdentity()
.setGroupInstanceId("group-instance-id")
.setMemberId("other-member-id") // invalid member id
@@ -9112,7 +9108,7 @@ public class GroupMetadataManagerTest {
);
LeaveGroupResponseData expectedResponse = new LeaveGroupResponseData()
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new LeaveGroupResponseData.MemberResponse()
.setGroupInstanceId("group-instance-id")
.setMemberId("other-member-id")
@@ -9143,7 +9139,7 @@ public class GroupMetadataManagerTest {
CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> leaveResult = context.sendClassicGroupLeave(
new LeaveGroupRequestData()
.setGroupId("group-id")
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new MemberIdentity()
.setGroupInstanceId("group-instance-id")
.setMemberId(UNKNOWN_MEMBER_ID)
@@ -9151,7 +9147,7 @@ public class GroupMetadataManagerTest {
);
LeaveGroupResponseData expectedResponse = new LeaveGroupResponseData()
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new LeaveGroupResponseData.MemberResponse()
.setGroupInstanceId("group-instance-id")));
@@ -9404,7 +9400,7 @@ public class GroupMetadataManagerTest {
.build();
context.createClassicGroup("group-id");
- List<CoordinatorRecord> expectedRecords = Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord("group-id"));
+ List<CoordinatorRecord> expectedRecords = List.of(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord("group-id"));
List<CoordinatorRecord> records = new ArrayList<>();
context.groupMetadataManager.createGroupTombstoneRecords("group-id", records);
assertEquals(expectedRecords, records);
@@ -9416,7 +9412,7 @@ public class GroupMetadataManagerTest {
.build();
ClassicGroup group = context.createClassicGroup("group-id");
- List<CoordinatorRecord> expectedRecords = Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord("group-id"));
+ List<CoordinatorRecord> expectedRecords = List.of(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord("group-id"));
List<CoordinatorRecord> records = new ArrayList<>();
context.groupMetadataManager.maybeDeleteGroup("group-id", records);
assertEquals(expectedRecords, records);
@@ -9494,7 +9490,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -9503,7 +9499,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
- Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
+ Map.of(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5),
mkTopicAssignment(barTopicId, 0, 1, 2)
)))
@@ -9596,7 +9592,7 @@ public class GroupMetadataManagerTest {
String classicGroupId = "classic-group-id";
String memberId = Uuid.randomUuid().toString();
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new NoOpPartitionAssignor()))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(new NoOpPartitionAssignor()))
.build();
ClassicGroup classicGroup = new ClassicGroup(
new LogContext(),
@@ -9624,7 +9620,7 @@ public class GroupMetadataManagerTest {
String classicGroupId = "classic-group-id";
String memberId = Uuid.randomUuid().toString();
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new NoOpPartitionAssignor()))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(new NoOpPartitionAssignor()))
.build();
ClassicGroup classicGroup = new ClassicGroup(
new LogContext(),
@@ -9729,8 +9725,8 @@ public class GroupMetadataManagerTest {
.build();
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.UPGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.UPGRADE.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(metadataImage)
.build();
@@ -9747,7 +9743,7 @@ public class GroupMetadataManagerTest {
))))
);
- Map<String, byte[]> assignments = Collections.singletonMap(
+ Map<String, byte[]> assignments = Map.of(
memberId1,
Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(List.of(
new TopicPartition(fooTopicName, 0),
@@ -9890,8 +9886,8 @@ public class GroupMetadataManagerTest {
.build();
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.UPGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.UPGRADE.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(metadataImage)
.build();
@@ -9914,7 +9910,7 @@ public class GroupMetadataManagerTest {
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
List.of(fooTopicName, barTopicName),
null,
- Collections.singletonList(new TopicPartition(barTopicName, 0))
+ List.of(new TopicPartition(barTopicName, 0))
))))
);
@@ -9923,7 +9919,7 @@ public class GroupMetadataManagerTest {
new TopicPartition(fooTopicName, 0),
new TopicPartition(fooTopicName, 1)
)))),
- memberId2, Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Collections.singletonList(
+ memberId2, Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(List.of(
new TopicPartition(barTopicName, 0)
))))
);
@@ -10095,8 +10091,8 @@ public class GroupMetadataManagerTest {
.build();
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.UPGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(new NoOpPartitionAssignor()))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.UPGRADE.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(new NoOpPartitionAssignor()))
.withMetadataImage(metadataImage)
.build();
@@ -10104,16 +10100,16 @@ public class GroupMetadataManagerTest {
protocols.add(new JoinGroupRequestData.JoinGroupRequestProtocol()
.setName(NoOpPartitionAssignor.NAME)
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList(fooTopicName),
+ List.of(fooTopicName),
null,
- Collections.singletonList(new TopicPartition(fooTopicName, 0))
+ List.of(new TopicPartition(fooTopicName, 0))
))))
);
- Map<String, byte[]> assignments = Collections.singletonMap(
+ Map<String, byte[]> assignments = Map.of(
memberId,
Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(
- Collections.singletonList(new TopicPartition(fooTopicName, 0))
+ List.of(new TopicPartition(fooTopicName, 0))
)))
);
@@ -10150,7 +10146,7 @@ public class GroupMetadataManagerTest {
.setInstanceId(instanceId)
.setRebalanceTimeoutMs(5000)
.setServerAssignor(NoOpPartitionAssignor.NAME)
- .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .setSubscribedTopicNames(List.of(fooTopicName))
.setTopicPartitions(Collections.emptyList()),
ApiKeys.CONSUMER_GROUP_HEARTBEAT.latestVersion()
);
@@ -10161,7 +10157,7 @@ public class GroupMetadataManagerTest {
.setPreviousMemberEpoch(0)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .setSubscribedTopicNames(List.of(fooTopicName))
.setRebalanceTimeoutMs(10000)
.setClassicMemberMetadata(
new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
@@ -10257,8 +10253,8 @@ public class GroupMetadataManagerTest {
.build();
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.UPGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(new NoOpPartitionAssignor()))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.UPGRADE.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(new NoOpPartitionAssignor()))
.withMetadataImage(metadataImage)
.build();
@@ -10374,7 +10370,7 @@ public class GroupMetadataManagerTest {
Uuid barTopicId = Uuid.randomUuid();
String barTopicName = "bar";
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
@@ -10427,8 +10423,8 @@ public class GroupMetadataManagerTest {
// Consumer group with two static members.
// Member 1 uses the classic protocol and member 2 uses the consumer protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.UPGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(new NoOpPartitionAssignor()))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.UPGRADE.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(new NoOpPartitionAssignor()))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -10549,8 +10545,8 @@ public class GroupMetadataManagerTest {
.build();
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.UPGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.UPGRADE.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(metadataImage)
.build();
@@ -10573,7 +10569,7 @@ public class GroupMetadataManagerTest {
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
List.of(fooTopicName, barTopicName),
null,
- Collections.singletonList(new TopicPartition(barTopicName, 0))
+ List.of(new TopicPartition(barTopicName, 0))
))))
);
@@ -10582,7 +10578,7 @@ public class GroupMetadataManagerTest {
new TopicPartition(fooTopicName, 0),
new TopicPartition(fooTopicName, 1)
)))),
- memberId2, Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Collections.singletonList(
+ memberId2, Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(List.of(
new TopicPartition(barTopicName, 0)
))))
);
@@ -10872,7 +10868,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
@@ -10923,8 +10919,8 @@ public class GroupMetadataManagerTest {
// Consumer group with two members.
// Member 1 uses the classic protocol and member 2 uses the consumer protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DOWNGRADE.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -10968,7 +10964,7 @@ public class GroupMetadataManagerTest {
new TopicPartition(barTopicName, 0),
new TopicPartition(barTopicName, 1)
))));
- Map<String, byte[]> assignments = Collections.singletonMap(memberId1, assignment);
+ Map<String, byte[]> assignments = Map.of(memberId1, assignment);
ClassicGroup expectedClassicGroup = new ClassicGroup(
new LogContext(),
@@ -11054,7 +11050,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
@@ -11105,8 +11101,8 @@ public class GroupMetadataManagerTest {
// Consumer group with two members.
// Member 1 uses the classic protocol and member 2 uses the consumer protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DOWNGRADE.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -11155,7 +11151,7 @@ public class GroupMetadataManagerTest {
new TopicPartition(barTopicName, 0),
new TopicPartition(barTopicName, 1)
))));
- Map<String, byte[]> assignments = Collections.singletonMap(memberId1, assignment);
+ Map<String, byte[]> assignments = Map.of(memberId1, assignment);
ClassicGroup expectedClassicGroup = new ClassicGroup(
new LogContext(),
@@ -11236,7 +11232,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
@@ -11287,8 +11283,8 @@ public class GroupMetadataManagerTest {
// Consumer group with two members.
// Member 1 uses the classic protocol and member 2 uses the consumer protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DOWNGRADE.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -11359,7 +11355,7 @@ public class GroupMetadataManagerTest {
new TopicPartition(barTopicName, 0),
new TopicPartition(barTopicName, 1)
))));
- Map<String, byte[]> assignments = Collections.singletonMap(memberId1, assignment);
+ Map<String, byte[]> assignments = Map.of(memberId1, assignment);
ClassicGroup expectedClassicGroup = new ClassicGroup(
new LogContext(),
@@ -11440,7 +11436,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols1 = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols1 = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
@@ -11481,7 +11477,7 @@ public class GroupMetadataManagerTest {
.setPreviousMemberEpoch(9)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setServerAssignorName("range")
.setRebalanceTimeoutMs(45000)
.setAssignedPartitions(mkAssignment(
@@ -11491,8 +11487,8 @@ public class GroupMetadataManagerTest {
// Consumer group with two members.
// Member 1 uses the classic protocol and static member 2 uses the consumer protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DOWNGRADE.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 2)
@@ -11538,10 +11534,10 @@ public class GroupMetadataManagerTest {
.setClassicMemberMetadata(
new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
.setSessionTimeoutMs(joinRequest.sessionTimeoutMs())
- .setSupportedProtocols(Collections.singletonList(new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .setSupportedProtocols(List.of(new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList(fooTopicName)))))))
+ List.of(fooTopicName)))))))
).build();
byte[] assignment1 = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(List.of(
@@ -11672,7 +11668,7 @@ public class GroupMetadataManagerTest {
.setMemberEpoch(10)
.setPreviousMemberEpoch(10)
.build()))
- .withConsumerGroupMaxSize(1)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG, 1)
.build();
JoinGroupRequestData request = new GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
@@ -11734,7 +11730,7 @@ public class GroupMetadataManagerTest {
String memberId = Uuid.randomUuid().toString();
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 2)
.addTopic(barTopicId, barTopicName, 1)
@@ -11872,7 +11868,7 @@ public class GroupMetadataManagerTest {
)));
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 2)
.addRacks()
@@ -11898,7 +11894,7 @@ public class GroupMetadataManagerTest {
.withGroupId(groupId)
.withMemberId(newMemberId)
.withProtocols(GroupMetadataManagerTestContext.toConsumerProtocol(
- Collections.singletonList(fooTopicName),
+ List.of(fooTopicName),
Collections.emptyList()))
.build();
@@ -11925,7 +11921,7 @@ public class GroupMetadataManagerTest {
String instanceId = "instance-id";
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new NoOpPartitionAssignor()))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(new NoOpPartitionAssignor()))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 2)
.addTopic(barTopicId, barTopicName, 1)
@@ -12017,7 +12013,8 @@ public class GroupMetadataManagerTest {
String memberId = Uuid.randomUuid().toString();
String instanceId = "instance-id";
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new NoOpPartitionAssignor()))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DISABLED.toString())
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(new NoOpPartitionAssignor()))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 2)
.addRacks()
@@ -12031,7 +12028,7 @@ public class GroupMetadataManagerTest {
.setState(MemberState.STABLE)
.setMemberEpoch(10)
.setPreviousMemberEpoch(10)
- .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .setSubscribedTopicNames(List.of(fooTopicName))
.setRebalanceTimeoutMs(500)
.setAssignedPartitions(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1)))
@@ -12047,7 +12044,7 @@ public class GroupMetadataManagerTest {
.withMemberId(UNKNOWN_MEMBER_ID)
.withGroupInstanceId(instanceId)
.withProtocols(GroupMetadataManagerTestContext.toConsumerProtocol(
- Collections.singletonList(fooTopicName),
+ List.of(fooTopicName),
Collections.emptyList()))
.build();
@@ -12067,7 +12064,7 @@ public class GroupMetadataManagerTest {
.setPreviousMemberEpoch(0)
.setInstanceId(instanceId)
.setState(MemberState.STABLE)
- .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .setSubscribedTopicNames(List.of(fooTopicName))
.setAssignedPartitions(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1)))
.setRebalanceTimeoutMs(500)
@@ -12080,7 +12077,7 @@ public class GroupMetadataManagerTest {
.setState(MemberState.STABLE)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .setSubscribedTopicNames(List.of(fooTopicName))
.setAssignedPartitions(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1)))
.setRebalanceTimeoutMs(500)
@@ -12135,7 +12132,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 2)
.addTopic(barTopicId, barTopicName, 1)
@@ -12359,7 +12356,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 2)
.addTopic(barTopicId, barTopicName, 1)
@@ -12544,7 +12541,7 @@ public class GroupMetadataManagerTest {
.build();
assertRecordsEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember2)),
+ List.of(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember2)),
joinResult2.records
);
assertEquals(expectedMember2.state(), group.getOrMaybeCreateMember(memberId1, false).state());
@@ -12593,7 +12590,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 2)
.addTopic(barTopicId, barTopicName, 1)
@@ -12740,7 +12737,7 @@ public class GroupMetadataManagerTest {
joinResponse1.generationId(),
joinResponse1.protocolName(),
joinResponse1.protocolType(),
- Collections.singletonList(new TopicPartition(fooTopicName, 0))
+ List.of(new TopicPartition(fooTopicName, 0))
);
// Member 1 heartbeats to be notified to rejoin.
@@ -12762,7 +12759,7 @@ public class GroupMetadataManagerTest {
.withSessionTimeoutMs(5000)
.withProtocols(GroupMetadataManagerTestContext.toConsumerProtocol(
List.of(fooTopicName, barTopicName, zarTopicName),
- Collections.singletonList(new TopicPartition(fooTopicName, 0))))
+ List.of(new TopicPartition(fooTopicName, 0))))
.build();
GroupMetadataManagerTestContext.JoinResult joinResult2 = context.sendClassicGroupJoin(request2);
@@ -12779,7 +12776,7 @@ public class GroupMetadataManagerTest {
.setSupportedProtocols(ConsumerGroupMember.classicProtocolListFromJoinRequestProtocolCollection(
GroupMetadataManagerTestContext.toConsumerProtocol(
List.of(fooTopicName, barTopicName, zarTopicName),
- Collections.singletonList(new TopicPartition(fooTopicName, 0))
+ List.of(new TopicPartition(fooTopicName, 0))
)
))
)
@@ -12929,7 +12926,7 @@ public class GroupMetadataManagerTest {
new TopicPartition(barTopicName, 1)
);
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(
@@ -12969,8 +12966,7 @@ public class GroupMetadataManagerTest {
// Consumer group with two members.
// Member 1 uses the classic protocol and member 2 uses the consumer protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DOWNGRADE.toString())
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -13007,8 +13003,7 @@ public class GroupMetadataManagerTest {
// Consumer group with a member that doesn't use the classic protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DOWNGRADE.toString())
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
.withMember(new ConsumerGroupMember.Builder(memberId)
.build()))
@@ -13057,8 +13052,7 @@ public class GroupMetadataManagerTest {
// Consumer group with a static member.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DOWNGRADE.toString())
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
.withMember(new ConsumerGroupMember.Builder(memberId)
.setInstanceId(instanceId)
@@ -13082,12 +13076,12 @@ public class GroupMetadataManagerTest {
String groupId = "group-id";
String memberId = Uuid.randomUuid().toString();
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(
new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList("foo"),
+ List.of("foo"),
null,
Collections.emptyList()
)
@@ -13096,8 +13090,7 @@ public class GroupMetadataManagerTest {
// Consumer group with a member using the classic protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DOWNGRADE.toString())
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
.withMember(new ConsumerGroupMember.Builder(memberId)
.setClassicMemberMetadata(
@@ -13147,12 +13140,12 @@ public class GroupMetadataManagerTest {
String groupId = "group-id";
String memberId = Uuid.randomUuid().toString();
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(
new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList("foo"),
+ List.of("foo"),
null,
Collections.emptyList()
)
@@ -13161,8 +13154,7 @@ public class GroupMetadataManagerTest {
// Consumer group with a member using the classic protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DOWNGRADE.toString())
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
.withMember(new ConsumerGroupMember.Builder(memberId)
.setClassicMemberMetadata(
@@ -13190,12 +13182,12 @@ public class GroupMetadataManagerTest {
String groupId = "group-id";
String memberId = Uuid.randomUuid().toString();
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(
new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList("foo"),
+ List.of("foo"),
null,
Collections.emptyList()
)
@@ -13205,8 +13197,7 @@ public class GroupMetadataManagerTest {
// Consumer group with a member using the classic protocol.
// The group epoch is greater than the member epoch.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DOWNGRADE.toString())
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 11)
.withMember(new ConsumerGroupMember.Builder(memberId)
.setRebalanceTimeoutMs(10000)
@@ -13237,12 +13228,12 @@ public class GroupMetadataManagerTest {
String memberId = Uuid.randomUuid().toString();
int sessionTimeout = 5000;
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(
new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList("foo"),
+ List.of("foo"),
null,
Collections.emptyList()
)
@@ -13251,16 +13242,15 @@ public class GroupMetadataManagerTest {
// Consumer group with a member using the classic protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
- .withMember(new ConsumerGroupMember.Builder(memberId)
- .setClassicMemberMetadata(
- new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
- .setSessionTimeoutMs(sessionTimeout)
- .setSupportedProtocols(protocols)
- )
- .setMemberEpoch(10)
- .build()))
+ .withMember(new ConsumerGroupMember.Builder(memberId)
+ .setClassicMemberMetadata(
+ new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSessionTimeoutMs(sessionTimeout)
+ .setSupportedProtocols(protocols)
+ )
+ .setMemberEpoch(10)
+ .build()))
.build();
// Heartbeat to schedule the session timeout.
@@ -13297,12 +13287,12 @@ public class GroupMetadataManagerTest {
int sessionTimeout = 5000;
int rebalanceTimeout = 10000;
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(
new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList("foo"),
+ List.of("foo"),
null,
Collections.emptyList()
)
@@ -13347,7 +13337,6 @@ public class GroupMetadataManagerTest {
.build();
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
.withMember(member1)
.withMember(member2)
@@ -13472,12 +13461,12 @@ public class GroupMetadataManagerTest {
String memberId = Uuid.randomUuid().toString();
int sessionTimeout = 5000;
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(
new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList("foo"),
+ List.of("foo"),
null,
Collections.emptyList()
)
@@ -13486,7 +13475,6 @@ public class GroupMetadataManagerTest {
// Consumer group with a member using the classic protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
.withMember(new ConsumerGroupMember.Builder(memberId)
.setClassicMemberMetadata(
@@ -13533,12 +13521,12 @@ public class GroupMetadataManagerTest {
String memberId = Uuid.randomUuid().toString();
int rebalanceTimeout = 500;
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(
new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList("foo"),
+ List.of("foo"),
null,
Collections.emptyList()
)
@@ -13547,7 +13535,6 @@ public class GroupMetadataManagerTest {
// Consumer group with a member using the classic protocol whose member epoch is smaller than the group epoch.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
.withMember(new ConsumerGroupMember.Builder(memberId)
.setRebalanceTimeoutMs(rebalanceTimeout)
@@ -13607,22 +13594,22 @@ public class GroupMetadataManagerTest {
Uuid barTopicId = Uuid.randomUuid();
String barTopicName = "bar";
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocol1 = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocol1 = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
List.of(fooTopicName, barTopicName),
null,
- Collections.singletonList(new TopicPartition(fooTopicName, 0))
+ List.of(new TopicPartition(fooTopicName, 0))
))))
);
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocol2 = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocol2 = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
List.of(fooTopicName, barTopicName),
null,
- Collections.singletonList(new TopicPartition(fooTopicName, 1))
+ List.of(new TopicPartition(fooTopicName, 1))
))))
);
@@ -13677,7 +13664,6 @@ public class GroupMetadataManagerTest {
// Static member 2 uses the classic protocol.
// Static member 3 uses the consumer protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 2)
.addTopic(barTopicId, barTopicName, 1)
@@ -13707,7 +13693,7 @@ public class GroupMetadataManagerTest {
.withSessionTimeoutMs(member1.classicMemberMetadata().get().sessionTimeoutMs())
.withProtocols(GroupMetadataManagerTestContext.toConsumerProtocol(
List.of(fooTopicName, barTopicName),
- Collections.singletonList(new TopicPartition(fooTopicName, 0))))
+ List.of(new TopicPartition(fooTopicName, 0))))
.build()
).appendFuture.complete(null);
context.assertSyncTimeout(groupId, memberId1, member1.rebalanceTimeoutMs());
@@ -13808,13 +13794,13 @@ public class GroupMetadataManagerTest {
Uuid barTopicId = Uuid.randomUuid();
String barTopicName = "bar";
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocol = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocol = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
List.of(fooTopicName, barTopicName),
null,
- Collections.singletonList(new TopicPartition(fooTopicName, 0))
+ List.of(new TopicPartition(fooTopicName, 0))
))))
);
@@ -13840,7 +13826,7 @@ public class GroupMetadataManagerTest {
.setPreviousMemberEpoch(9)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setServerAssignorName("range")
.setRebalanceTimeoutMs(45000)
.setAssignedPartitions(mkAssignment(mkTopicAssignment(barTopicId, 0)))
@@ -13849,7 +13835,6 @@ public class GroupMetadataManagerTest {
// Consumer group with two members.
// Member 1 uses the classic protocol and member 2 uses the consumer protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 2)
.addTopic(barTopicId, barTopicName, 1)
@@ -13872,7 +13857,7 @@ public class GroupMetadataManagerTest {
CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> leaveResult = context.sendClassicGroupLeave(
new LeaveGroupRequestData()
.setGroupId("group-id")
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new MemberIdentity()
.setMemberId(memberId1)
))
@@ -13880,7 +13865,7 @@ public class GroupMetadataManagerTest {
assertEquals(
new LeaveGroupResponseData()
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new LeaveGroupResponseData.MemberResponse()
.setGroupInstanceId(null)
.setMemberId(memberId1)
@@ -13910,7 +13895,6 @@ public class GroupMetadataManagerTest {
// Consumer group without member using the classic protocol.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
.withMember(new ConsumerGroupMember.Builder(memberId)
.build()))
@@ -13951,7 +13935,7 @@ public class GroupMetadataManagerTest {
String groupId = "group-id";
String nonClassicMemberId = "1";
- List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = List.of(
new ConsumerGroupMemberMetadataValue.ClassicProtocol()
.setName("range")
.setMetadata(new byte[0])
@@ -13966,8 +13950,8 @@ public class GroupMetadataManagerTest {
.build();
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withClassicGroupMaxSize(1)
- .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withConfig(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG, 1)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DOWNGRADE.toString())
.withConsumerGroup(
new ConsumerGroupBuilder(groupId, 10)
.withMember(member)
@@ -14051,7 +14035,7 @@ public class GroupMetadataManagerTest {
assertEquals(0, groups.size());
// Group id not found
- groups = context.sendShareGroupDescribe(Collections.singletonList("unknown-group"));
+ groups = context.sendShareGroupDescribe(List.of("unknown-group"));
assertEquals(1, groups.size());
assertEquals(Errors.GROUP_ID_NOT_FOUND.code(), groups.get(0).errorCode());
}
@@ -14076,7 +14060,7 @@ public class GroupMetadataManagerTest {
.setGroupId(groupIds.get(1))
.setMemberId(Uuid.randomUuid().toString())
.setMemberEpoch(0)
- .setSubscribedTopicNames(Collections.singletonList("foo")));
+ .setSubscribedTopicNames(List.of("foo")));
// Verify that a member id was generated for the new member.
String memberId = result.response().memberId();
@@ -14093,12 +14077,12 @@ public class GroupMetadataManagerTest {
.setGroupEpoch(16)
.setAssignmentEpoch(16)
.setGroupId(groupIds.get(1))
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new ShareGroupMember.Builder(memberId)
.setMemberEpoch(16)
.setClientId("client")
.setClientHost("localhost/127.0.0.1")
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.build()
.asShareGroupDescribeMember(
new MetadataImageBuilder().build().topics()
@@ -14219,7 +14203,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
- Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
+ Map.of(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5),
mkTopicAssignment(barTopicId, 0, 1, 2)
)))
@@ -14380,7 +14364,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("share");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
.withShareGroupAssignor(assignor)
- .withShareGroupMaxSize(1)
+ .withConfig(GroupCoordinatorConfig.SHARE_GROUP_MAX_SIZE_CONFIG, 1)
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
@@ -14443,7 +14427,7 @@ public class GroupMetadataManagerTest {
context.replay(GroupCoordinatorRecordHelpers.newShareGroupMemberSubscriptionRecord(groupId, new ShareGroupMember.Builder(memberId1)
.setState(MemberState.STABLE)
- .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .setSubscribedTopicNames(List.of(fooTopicName))
.build()));
context.replay(GroupCoordinatorRecordHelpers.newShareGroupEpochRecord(groupId, 11));
@@ -14485,7 +14469,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addRacks()
@@ -14493,7 +14477,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
- Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
+ Map.of(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)))
));
@@ -14506,7 +14490,7 @@ public class GroupMetadataManagerTest {
.setMemberId(memberId)
.setMemberEpoch(0)
.setRebalanceTimeoutMs(90000)
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.setTopicPartitions(Collections.emptyList()));
assertEquals(1, result.response().memberEpoch());
@@ -14580,7 +14564,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
- Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
+ Map.of(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)))
));
@@ -14592,7 +14576,7 @@ public class GroupMetadataManagerTest {
.setGroupId(groupId)
.setMemberId(memberId)
.setMemberEpoch(0)
- .setSubscribedTopicNames(Collections.singletonList("foo")));
+ .setSubscribedTopicNames(List.of("foo")));
assertEquals(1, result.response().memberEpoch());
// Verify heartbeat interval
@@ -14656,7 +14640,7 @@ public class GroupMetadataManagerTest {
.setClientHost("clienthost")
.setServerAssignorName("range")
.setRackId("rackid")
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.build();
// The group and the member are created if they do not exist.
@@ -14707,7 +14691,7 @@ public class GroupMetadataManagerTest {
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
.build();
- Map<String, TopicMetadata> metadata = Collections.singletonMap(
+ Map<String, TopicMetadata> metadata = Map.of(
"bar",
new TopicMetadata(Uuid.randomUuid(), "bar", 10)
);
@@ -14825,7 +14809,7 @@ public class GroupMetadataManagerTest {
.setPreviousMemberEpoch(0)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.build())
.withAssignment(memberId, mkAssignment())
.withAssignmentEpoch(1))
@@ -14858,7 +14842,7 @@ public class GroupMetadataManagerTest {
.setPreviousMemberEpoch(0)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.build())
.withAssignment(memberId, mkAssignment())
.withAssignmentEpoch(1))
@@ -14892,7 +14876,7 @@ public class GroupMetadataManagerTest {
.setPreviousMemberEpoch(0)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.build())
.withAssignment(memberId, mkAssignment())
.withAssignmentEpoch(1))
@@ -14927,7 +14911,7 @@ public class GroupMetadataManagerTest {
.setPreviousMemberEpoch(0)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.build())
.withAssignment(memberId, mkAssignment())
.withAssignmentEpoch(1))
@@ -14936,7 +14920,7 @@ public class GroupMetadataManagerTest {
assertThrows(UnknownMemberIdException.class, () -> context.sendClassicGroupLeave(
new LeaveGroupRequestData()
.setGroupId(groupId)
- .setMembers(Collections.singletonList(
+ .setMembers(List.of(
new MemberIdentity()
.setMemberId(memberId)))));
}
@@ -14957,19 +14941,19 @@ public class GroupMetadataManagerTest {
.setPreviousMemberEpoch(0)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
- .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicNames(List.of("foo"))
.build())
.withAssignment(memberId, mkAssignment())
.withAssignmentEpoch(1))
.build();
- List<ConsumerGroupDescribeResponseData.DescribedGroup> expected = Collections.singletonList(
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> expected = List.of(
new ConsumerGroupDescribeResponseData.DescribedGroup()
.setGroupId(groupId)
.setErrorCode(Errors.GROUP_ID_NOT_FOUND.code())
);
- List<ConsumerGroupDescribeResponseData.DescribedGroup> actual = context.sendConsumerGroupDescribe(Collections.singletonList(groupId));
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> actual = context.sendConsumerGroupDescribe(List.of(groupId));
assertEquals(expected, actual);
}
@@ -14986,7 +14970,7 @@ public class GroupMetadataManagerTest {
// Consumer group with one static member.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.build())
@@ -15025,23 +15009,23 @@ public class GroupMetadataManagerTest {
int epoch = 10;
String topicName = "topicName";
ConsumerGroupMember.Builder memberBuilder = new ConsumerGroupMember.Builder(memberId)
- .setSubscribedTopicNames(Collections.singletonList(topicName))
+ .setSubscribedTopicNames(List.of(topicName))
.setServerAssignorName("assignorName");
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withConsumerGroup(new ConsumerGroupBuilder(groupId, epoch)
.withMember(memberBuilder.build()))
.build();
- List<ShareGroupDescribeResponseData.DescribedGroup> expected = Collections.singletonList(
+ List<ShareGroupDescribeResponseData.DescribedGroup> expected = List.of(
new ShareGroupDescribeResponseData.DescribedGroup()
.setGroupId(groupId)
.setErrorCode(Errors.GROUP_ID_NOT_FOUND.code())
);
- List<ShareGroupDescribeResponseData.DescribedGroup> actual = context.sendShareGroupDescribe(Collections.singletonList(groupId));
+ List<ShareGroupDescribeResponseData.DescribedGroup> actual = context.sendShareGroupDescribe(List.of(groupId));
assertEquals(expected, actual);
}
@@ -15135,7 +15119,7 @@ public class GroupMetadataManagerTest {
});
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 2)
.build())
@@ -15190,7 +15174,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.build(12345L))
@@ -15288,7 +15272,7 @@ public class GroupMetadataManagerTest {
MockPartitionAssignor assignor = new MockPartitionAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -15397,7 +15381,7 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(Collections.emptyMap()));
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addTopic(barTopicId, barTopicName, 3)
@@ -15581,7 +15565,7 @@ public class GroupMetadataManagerTest {
.build(1L);
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(image)
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
.withMember(new ConsumerGroupMember.Builder(memberId1)
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 4230934a499..13ced4b6dab 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
@@ -19,6 +19,7 @@ package org.apache.kafka.coordinator.group;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor;
import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
@@ -49,6 +50,7 @@ import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.common.runtime.CoordinatorResult;
import org.apache.kafka.coordinator.common.runtime.MockCoordinatorExecutor;
@@ -99,6 +101,7 @@ import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -136,6 +139,45 @@ public class GroupMetadataManagerTestContext {
static final String DEFAULT_CLIENT_ID = "client";
static final InetAddress DEFAULT_CLIENT_ADDRESS = InetAddress.getLoopbackAddress();
+ private static class GroupCoordinatorConfigContext extends GroupCoordinatorConfig {
+ GroupCoordinatorConfigContext(AbstractConfig config) {
+ super(config);
+ }
+
+ public static GroupCoordinatorConfig fromProps(
+ Map<?, ?> props
+ ) {
+ return new GroupCoordinatorConfigContext(
+ new AbstractConfig(
+ Utils.mergeConfigs(List.of(
+ GroupCoordinatorConfig.GROUP_COORDINATOR_CONFIG_DEF,
+ GroupCoordinatorConfig.NEW_GROUP_CONFIG_DEF,
+ GroupCoordinatorConfig.OFFSET_MANAGEMENT_CONFIG_DEF,
+ GroupCoordinatorConfig.CONSUMER_GROUP_CONFIG_DEF,
+ GroupCoordinatorConfig.SHARE_GROUP_CONFIG_DEF
+ )),
+ props
+ )
+ );
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected List<ConsumerGroupPartitionAssignor> consumerGroupAssignors(
+ AbstractConfig config
+ ) {
+ // In unit tests, it is pretty convenient to have the ability to pass instantiated
+ // assignors. Hence, we check if the provided assignors are already instantiated.
+ // Otherwise, we use the regular method.
+ List<?> classes = config.getList(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG);
+ if (classes.stream().allMatch(o -> o instanceof ConsumerGroupPartitionAssignor)) {
+ return Collections.unmodifiableList((List<ConsumerGroupPartitionAssignor>) classes);
+ }
+
+ return super.consumerGroupAssignors(config);
+ }
+ }
+
public static void assertNoOrEmptyResult(List<MockCoordinatorTimer.ExpiredTimeout<Void, CoordinatorRecord>> timeouts) {
assertTrue(timeouts.size() <= 1);
timeouts.forEach(timeout -> assertEquals(EMPTY_RESULT, timeout.result));
@@ -148,7 +190,7 @@ public class GroupMetadataManagerTestContext {
protocols.add(new JoinGroupRequestData.JoinGroupRequestProtocol()
.setName(protocolNames[i])
.setMetadata(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
- Collections.singletonList(topicNames.get(i % topicNames.size())))).array())
+ List.of(topicNames.get(i % topicNames.size())))).array())
);
}
return protocols;
@@ -406,29 +448,19 @@ public class GroupMetadataManagerTestContext {
private final SnapshotRegistry snapshotRegistry = new SnapshotRegistry(logContext);
private MetadataImage metadataImage;
private GroupConfigManager groupConfigManager;
- private List<ConsumerGroupPartitionAssignor> consumerGroupAssignors = Collections.singletonList(new MockPartitionAssignor("range"));
private final List<ConsumerGroupBuilder> consumerGroupBuilders = new ArrayList<>();
- private int consumerGroupMaxSize = Integer.MAX_VALUE;
- private int consumerGroupMetadataRefreshIntervalMs = Integer.MAX_VALUE;
- private int classicGroupMaxSize = Integer.MAX_VALUE;
- private int classicGroupInitialRebalanceDelayMs = 3000;
- private final int classicGroupNewMemberJoinTimeoutMs = 5 * 60 * 1000;
- private int classicGroupMinSessionTimeoutMs = 10;
- private int classicGroupMaxSessionTimeoutMs = 10 * 60 * 1000;
private final GroupCoordinatorMetricsShard metrics = mock(GroupCoordinatorMetricsShard.class);
- private ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy = ConsumerGroupMigrationPolicy.DISABLED;
- // Share group configs
private ShareGroupPartitionAssignor shareGroupAssignor = new MockPartitionAssignor("share");
private final List<ShareGroupBuilder> shareGroupBuilders = new ArrayList<>();
- private int shareGroupMaxSize = Integer.MAX_VALUE;
+ private final Map<String, Object> config = new HashMap<>();
- public Builder withMetadataImage(MetadataImage metadataImage) {
- this.metadataImage = metadataImage;
+ public Builder withConfig(String key, Object value) {
+ config.put(key, value);
return this;
}
- public Builder withConsumerGroupAssignors(List<ConsumerGroupPartitionAssignor> assignors) {
- this.consumerGroupAssignors = assignors;
+ public Builder withMetadataImage(MetadataImage metadataImage) {
+ this.metadataImage = metadataImage;
return this;
}
@@ -437,41 +469,6 @@ public class GroupMetadataManagerTestContext {
return this;
}
- public Builder withConsumerGroupMaxSize(int consumerGroupMaxSize) {
- this.consumerGroupMaxSize = consumerGroupMaxSize;
- return this;
- }
-
- public Builder withConsumerGroupMetadataRefreshIntervalMs(int consumerGroupMetadataRefreshIntervalMs) {
- this.consumerGroupMetadataRefreshIntervalMs = consumerGroupMetadataRefreshIntervalMs;
- return this;
- }
-
- public Builder withClassicGroupMaxSize(int classicGroupMaxSize) {
- this.classicGroupMaxSize = classicGroupMaxSize;
- return this;
- }
-
- public Builder withClassicGroupInitialRebalanceDelayMs(int classicGroupInitialRebalanceDelayMs) {
- this.classicGroupInitialRebalanceDelayMs = classicGroupInitialRebalanceDelayMs;
- return this;
- }
-
- public Builder withClassicGroupMinSessionTimeoutMs(int classicGroupMinSessionTimeoutMs) {
- this.classicGroupMinSessionTimeoutMs = classicGroupMinSessionTimeoutMs;
- return this;
- }
-
- public Builder withClassicGroupMaxSessionTimeoutMs(int classicGroupMaxSessionTimeoutMs) {
- this.classicGroupMaxSessionTimeoutMs = classicGroupMaxSessionTimeoutMs;
- return this;
- }
-
- public Builder withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy) {
- this.consumerGroupMigrationPolicy = consumerGroupMigrationPolicy;
- return this;
- }
-
public Builder withShareGroup(ShareGroupBuilder builder) {
this.shareGroupBuilders.add(builder);
return this;
@@ -482,48 +479,37 @@ public class GroupMetadataManagerTestContext {
return this;
}
- public Builder withShareGroupMaxSize(int shareGroupMaxSize) {
- this.shareGroupMaxSize = shareGroupMaxSize;
- return this;
- }
-
public GroupMetadataManagerTestContext build() {
if (metadataImage == null) metadataImage = MetadataImage.EMPTY;
- if (consumerGroupAssignors == null) consumerGroupAssignors = Collections.emptyList();
if (groupConfigManager == null) groupConfigManager = createConfigManager();
+ config.putIfAbsent(
+ GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG,
+ List.of(new MockPartitionAssignor("range"))
+ );
+
+ GroupCoordinatorConfig groupCoordinatorConfig = GroupCoordinatorConfigContext.fromProps(config);
+
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext(
time,
timer,
executor,
snapshotRegistry,
metrics,
+ groupCoordinatorConfig,
new GroupMetadataManager.Builder()
.withSnapshotRegistry(snapshotRegistry)
.withLogContext(logContext)
.withTime(time)
.withTimer(timer)
.withExecutor(executor)
+ .withConfig(groupCoordinatorConfig)
.withMetadataImage(metadataImage)
- .withConsumerGroupHeartbeatInterval(5000)
- .withConsumerGroupSessionTimeout(45000)
- .withConsumerGroupMaxSize(consumerGroupMaxSize)
- .withConsumerGroupAssignors(consumerGroupAssignors)
- .withConsumerGroupMetadataRefreshIntervalMs(consumerGroupMetadataRefreshIntervalMs)
- .withClassicGroupMaxSize(classicGroupMaxSize)
- .withClassicGroupMinSessionTimeoutMs(classicGroupMinSessionTimeoutMs)
- .withClassicGroupMaxSessionTimeoutMs(classicGroupMaxSessionTimeoutMs)
- .withClassicGroupInitialRebalanceDelayMs(classicGroupInitialRebalanceDelayMs)
- .withClassicGroupNewMemberJoinTimeoutMs(classicGroupNewMemberJoinTimeoutMs)
.withGroupCoordinatorMetricsShard(metrics)
- .withConsumerGroupMigrationPolicy(consumerGroupMigrationPolicy)
.withShareGroupAssignor(shareGroupAssignor)
- .withShareGroupMaxSize(shareGroupMaxSize)
.withGroupConfigManager(groupConfigManager)
.build(),
- groupConfigManager,
- classicGroupInitialRebalanceDelayMs,
- classicGroupNewMemberJoinTimeoutMs
+ groupConfigManager
);
consumerGroupBuilders.forEach(builder -> builder.build(metadataImage.topics()).forEach(context::replay));
@@ -554,10 +540,9 @@ public class GroupMetadataManagerTestContext {
MockCoordinatorExecutor<CoordinatorRecord> executor,
SnapshotRegistry snapshotRegistry,
GroupCoordinatorMetricsShard metrics,
+ GroupCoordinatorConfig config,
GroupMetadataManager groupMetadataManager,
- GroupConfigManager groupConfigManager,
- int classicGroupInitialRebalanceDelayMs,
- int classicGroupNewMemberJoinTimeoutMs
+ GroupConfigManager groupConfigManager
) {
this.time = time;
this.timer = timer;
@@ -566,8 +551,8 @@ public class GroupMetadataManagerTestContext {
this.metrics = metrics;
this.groupMetadataManager = groupMetadataManager;
this.groupConfigManager = groupConfigManager;
- this.classicGroupInitialRebalanceDelayMs = classicGroupInitialRebalanceDelayMs;
- this.classicGroupNewMemberJoinTimeoutMs = classicGroupNewMemberJoinTimeoutMs;
+ this.classicGroupInitialRebalanceDelayMs = config.classicGroupInitialRebalanceDelayMs();
+ this.classicGroupNewMemberJoinTimeoutMs = config.classicGroupNewMemberJoinTimeoutMs();
snapshotRegistry.idempotentCreateSnapshot(lastWrittenOffset);
}
@@ -864,7 +849,7 @@ public class GroupMetadataManagerTestContext {
.build());
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
syncResult.records
);
// Simulate a successful write to the log.
@@ -1071,7 +1056,7 @@ public class GroupMetadataManagerTestContext {
SyncGroupRequestData.SyncGroupRequestAssignment::memberId, SyncGroupRequestData.SyncGroupRequestAssignment::assignment
));
assertEquals(
- Collections.singletonList(
+ List.of(
GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, groupAssignment, MetadataVersion.latestTesting())),
leaderSyncResult.records
);
@@ -1132,7 +1117,7 @@ public class GroupMetadataManagerTestContext {
// Now the group is stable, with the one member that joined above
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
syncResult.records
);
// Simulate a successful write to log.
@@ -1170,7 +1155,7 @@ public class GroupMetadataManagerTestContext {
syncResult = sendClassicGroupSync(syncRequest.setGenerationId(nextGenerationId));
assertEquals(
- Collections.singletonList(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
+ List.of(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, group.groupAssignment(), MetadataVersion.latestTesting())),
syncResult.records
);
// Simulate a successful write to log.
@@ -1225,7 +1210,7 @@ public class GroupMetadataManagerTestContext {
// Member should be removed as session expires.
List<MockCoordinatorTimer.ExpiredTimeout<Void, CoordinatorRecord>> timeouts = sleep(timeoutMs);
- List<CoordinatorRecord> expectedRecords = Collections.singletonList(newGroupMetadataRecord(
+ List<CoordinatorRecord> expectedRecords = List.of(newGroupMetadataRecord(
group.groupId(),
new GroupMetadataValue()
.setMembers(Collections.emptyList())
@@ -1399,10 +1384,10 @@ public class GroupMetadataManagerTestContext {
public void verifyDescribeGroupsReturnsDeadGroup(String groupId) {
List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
- describeGroups(Collections.singletonList(groupId));
+ describeGroups(List.of(groupId));
assertEquals(
- Collections.singletonList(new DescribeGroupsResponseData.DescribedGroup()
+ List.of(new DescribeGroupsResponseData.DescribedGroup()
.setGroupId(groupId)
.setGroupState(DEAD.toString())
),
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java
index a7e314f3f07..00ec5e9739e 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java
@@ -51,7 +51,6 @@ import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.common.runtime.CoordinatorResult;
import org.apache.kafka.coordinator.common.runtime.MockCoordinatorExecutor;
import org.apache.kafka.coordinator.common.runtime.MockCoordinatorTimer;
-import org.apache.kafka.coordinator.group.assignor.RangeAssignor;
import org.apache.kafka.coordinator.group.classic.ClassicGroup;
import org.apache.kafka.coordinator.group.classic.ClassicGroupMember;
import org.apache.kafka.coordinator.group.classic.ClassicGroupState;
@@ -139,9 +138,9 @@ public class OffsetMetadataManagerTest {
.withSnapshotRegistry(snapshotRegistry)
.withLogContext(logContext)
.withMetadataImage(metadataImage)
- .withConsumerGroupAssignors(Collections.singletonList(new RangeAssignor()))
.withGroupCoordinatorMetricsShard(metrics)
.withGroupConfigManager(configManager)
+ .withConfig(GroupCoordinatorConfig.fromProps(Collections.emptyMap()))
.build();
}
diff --git a/server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java b/server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java
index 537b0869d52..c32b89087a2 100644
--- a/server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java
+++ b/server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java
@@ -32,7 +32,7 @@ import org.apache.kafka.server.metrics.MetricConfigs;
import org.apache.kafka.storage.internals.log.CleanerConfig;
import org.apache.kafka.storage.internals.log.LogConfig;
-import java.util.Arrays;
+import java.util.List;
import java.util.Map;
/**
@@ -42,32 +42,32 @@ import java.util.Map;
* For more details check KAFKA-15853
*/
public abstract class AbstractKafkaConfig extends AbstractConfig {
- @SuppressWarnings("deprecation")
- public static final ConfigDef CONFIG_DEF = Utils.mergeConfigs(Arrays.asList(
- RemoteLogManagerConfig.configDef(),
- ZkConfigs.CONFIG_DEF,
- ServerConfigs.CONFIG_DEF,
- KRaftConfigs.CONFIG_DEF,
- SocketServerConfigs.CONFIG_DEF,
- ReplicationConfigs.CONFIG_DEF,
- GroupCoordinatorConfig.GROUP_COORDINATOR_CONFIG_DEF,
- GroupCoordinatorConfig.NEW_GROUP_CONFIG_DEF,
- GroupCoordinatorConfig.OFFSET_MANAGEMENT_CONFIG_DEF,
- GroupCoordinatorConfig.CONSUMER_GROUP_CONFIG_DEF,
- GroupCoordinatorConfig.SHARE_GROUP_CONFIG_DEF,
- CleanerConfig.CONFIG_DEF,
- LogConfig.SERVER_CONFIG_DEF,
- ShareGroupConfig.CONFIG_DEF,
- ShareCoordinatorConfig.CONFIG_DEF,
- TransactionLogConfig.CONFIG_DEF,
- TransactionStateManagerConfig.CONFIG_DEF,
- QuorumConfig.CONFIG_DEF,
- MetricConfigs.CONFIG_DEF,
- QuotaConfig.CONFIG_DEF,
- BrokerSecurityConfigs.CONFIG_DEF,
- DelegationTokenManagerConfigs.CONFIG_DEF,
- PasswordEncoderConfigs.CONFIG_DEF
- ));
+ public static final ConfigDef CONFIG_DEF = Utils.mergeConfigs(List.of(
+ RemoteLogManagerConfig.configDef(),
+ ZkConfigs.CONFIG_DEF,
+ ServerConfigs.CONFIG_DEF,
+ KRaftConfigs.CONFIG_DEF,
+ SocketServerConfigs.CONFIG_DEF,
+ ReplicationConfigs.CONFIG_DEF,
+ GroupCoordinatorConfig.GROUP_COORDINATOR_CONFIG_DEF,
+ GroupCoordinatorConfig.NEW_GROUP_CONFIG_DEF,
+ GroupCoordinatorConfig.OFFSET_MANAGEMENT_CONFIG_DEF,
+ GroupCoordinatorConfig.CONSUMER_GROUP_CONFIG_DEF,
+ GroupCoordinatorConfig.SHARE_GROUP_CONFIG_DEF,
+ CleanerConfig.CONFIG_DEF,
+ LogConfig.SERVER_CONFIG_DEF,
+ ShareGroupConfig.CONFIG_DEF,
+ ShareCoordinatorConfig.CONFIG_DEF,
+ TransactionLogConfig.CONFIG_DEF,
+ TransactionStateManagerConfig.CONFIG_DEF,
+ QuorumConfig.CONFIG_DEF,
+ MetricConfigs.CONFIG_DEF,
+ QuotaConfig.CONFIG_DEF,
+ BrokerSecurityConfigs.CONFIG_DEF,
+ DelegationTokenManagerConfigs.CONFIG_DEF,
+ PasswordEncoderConfigs.CONFIG_DEF
+ ));
+
public AbstractKafkaConfig(ConfigDef definition, Map<?, ?> originals, Map<String, ?> configProviderProps, boolean doLog) {
super(definition, originals, configProviderProps, doLog);
}