This is an automated email from the ASF dual-hosted git repository.
mjsax 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 359763b2865 KAFKA-20790: Add group.streams.assignors broker config +
loader (#22920)
359763b2865 is described below
commit 359763b2865848ddd1645ec644595adc5fab08fb
Author: gabriellafu <[email protected]>
AuthorDate: Wed Jul 29 23:49:20 2026 -0400
KAFKA-20790: Add group.streams.assignors broker config + loader (#22920)
Part of KIP-1357.
Adds the new broker config, allowing to register custom assignors, plus
corresponding runtime logic to load customer assignors. Also add the
corresponding group config, so each group can select the assignor it
want to use.
Reviewers: Matthias J. Sax <[email protected]>, Sean Quah
<[email protected]>
---
.../clients/admin/StreamsGroupDescription.java | 27 ++-
.../internals/DescribeStreamsGroupsHandler.java | 3 +-
.../message/StreamsGroupDescribeResponse.json | 7 +-
.../kafka/clients/admin/MockAdminClientTest.java | 6 +-
.../kafka/clients/admin/MockAdminClient.java | 3 +-
.../src/main/scala/kafka/server/ConfigHelper.scala | 2 +-
.../scala/unit/kafka/server/KafkaApisTest.scala | 3 +-
.../scala/unit/kafka/server/KafkaConfigTest.scala | 1 +
.../server/StreamsGroupHeartbeatRequestTest.scala | 162 +++++++++++++++++
docs/getting-started/upgrade.md | 1 +
.../developer-guide/streams-rebalance-protocol.md | 12 +-
docs/streams/upgrade-guide.md | 2 +
.../kafka/coordinator/group/GroupConfig.java | 31 ++++
.../coordinator/group/GroupCoordinatorConfig.java | 101 +++++++++++
.../coordinator/group/GroupCoordinatorShard.java | 1 +
.../coordinator/group/GroupMetadataManager.java | 39 +++-
.../coordinator/group/streams/StreamsGroup.java | 9 +-
.../kafka/coordinator/group/GroupConfigTest.java | 70 +++++++-
.../group/GroupCoordinatorConfigTest.java | 197 +++++++++++++++++++++
.../group/GroupMetadataManagerTest.java | 112 +++++++++++-
.../group/streams/StreamsGroupTest.java | 10 +-
.../kafka/server/config/AbstractKafkaConfig.java | 10 +-
.../server/config/AbstractKafkaConfigTest.java | 53 +++++-
.../kafka/tools/streams/StreamsGroupCommand.java | 14 +-
.../tools/streams/DescribeStreamsGroupTest.java | 10 +-
.../tools/streams/StreamsGroupCommandTest.java | 34 +++-
26 files changed, 867 insertions(+), 53 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupDescription.java
b/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupDescription.java
index 4955c588936..6674f69b7b5 100644
---
a/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupDescription.java
+++
b/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupDescription.java
@@ -47,7 +47,12 @@ public class StreamsGroupDescription {
private final Set<AclOperation> authorizedOperations;
private final Optional<StreamsGroupTopologyDescription>
topologyDescription;
private final StreamsGroupTopologyDescriptionStatus
topologyDescriptionStatus;
+ private final Optional<String> assignorName;
+ /**
+ * @deprecated Since 4.4. Use {@link #StreamsGroupDescription(String, int,
int, int, Collection, Collection, GroupState, Node, Set, Optional,
StreamsGroupTopologyDescriptionStatus, Optional)} instead.
+ */
+ @Deprecated(since = "4.4", forRemoval = true)
public StreamsGroupDescription(
final String groupId,
final int groupEpoch,
@@ -70,7 +75,8 @@ public class StreamsGroupDescription {
coordinator,
authorizedOperations,
Optional.empty(),
- StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED
+ StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
+ Optional.empty()
);
}
@@ -85,7 +91,8 @@ public class StreamsGroupDescription {
final Node coordinator,
final Set<AclOperation> authorizedOperations,
final Optional<StreamsGroupTopologyDescription>
topologyDescription,
- final StreamsGroupTopologyDescriptionStatus
topologyDescriptionStatus
+ final StreamsGroupTopologyDescriptionStatus
topologyDescriptionStatus,
+ final Optional<String> assignorName
) {
this.groupId = Objects.requireNonNull(groupId, "groupId must be
non-null");
this.groupEpoch = groupEpoch;
@@ -98,6 +105,7 @@ public class StreamsGroupDescription {
this.authorizedOperations = authorizedOperations;
this.topologyDescription = Objects.requireNonNull(topologyDescription,
"topologyDescription must be non-null");
this.topologyDescriptionStatus =
Objects.requireNonNull(topologyDescriptionStatus, "topologyDescriptionStatus
must be non-null");
+ this.assignorName = Objects.requireNonNull(assignorName, "assignorName
must be non-null");
}
/**
@@ -179,6 +187,14 @@ public class StreamsGroupDescription {
return topologyDescriptionStatus;
}
+ /**
+ * The task assignor the coordinator will use for the next assignment
computation. May differ from the assignor
+ * that computed the current assignment. Empty if the broker is too old to
report it.
+ */
+ public Optional<String> assignorName() {
+ return assignorName;
+ }
+
@Override
public boolean equals(final Object o) {
if (this == o) {
@@ -198,7 +214,8 @@ public class StreamsGroupDescription {
&& Objects.equals(coordinator, that.coordinator)
&& Objects.equals(authorizedOperations, that.authorizedOperations)
&& Objects.equals(topologyDescription, that.topologyDescription)
- && topologyDescriptionStatus == that.topologyDescriptionStatus;
+ && topologyDescriptionStatus == that.topologyDescriptionStatus
+ && Objects.equals(assignorName, that.assignorName);
}
@Override
@@ -214,7 +231,8 @@ public class StreamsGroupDescription {
coordinator,
authorizedOperations,
topologyDescription,
- topologyDescriptionStatus
+ topologyDescriptionStatus,
+ assignorName
);
}
@@ -232,6 +250,7 @@ public class StreamsGroupDescription {
", authorizedOperations=" +
authorizedOperations.stream().map(AclOperation::toString).collect(Collectors.joining(","))
+
", topologyDescription=" +
topologyDescription.map(Object::toString).orElse("") +
", topologyDescriptionStatus=" + topologyDescriptionStatus +
+ ", assignorName=" + assignorName.orElse("") +
')';
}
}
diff --git
a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java
b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java
index 11d734e5e39..af664afd88c 100644
---
a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java
+++
b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java
@@ -164,7 +164,8 @@ public class DescribeStreamsGroupsHandler extends
AdminApiHandler.Batched<Coordi
coordinator,
authorizedOperations,
topologyDescription,
- topologyDescriptionStatus
+ topologyDescriptionStatus,
+ Optional.ofNullable(describedGroup.assignorName())
);
completed.put(groupIdKey, streamsGroupDescription);
}
diff --git
a/clients/src/main/resources/common/message/StreamsGroupDescribeResponse.json
b/clients/src/main/resources/common/message/StreamsGroupDescribeResponse.json
index 8b6eb165e18..acc35647619 100644
---
a/clients/src/main/resources/common/message/StreamsGroupDescribeResponse.json
+++
b/clients/src/main/resources/common/message/StreamsGroupDescribeResponse.json
@@ -17,7 +17,8 @@
"apiKey": 89,
"type": "response",
"name": "StreamsGroupDescribeResponse",
- // Version 1 adds TopologyDescription and TopologyDescriptionStatus
(KIP-1331).
+ // Version 1 adds TopologyDescription and TopologyDescriptionStatus
(KIP-1331),
+ // and AssignorName (KIP-1357).
"validVersions": "0-1",
"flexibleVersions": "0+",
// Supported errors:
@@ -112,7 +113,9 @@
"nullableVersions": "1+", "default": "null",
"about": "The full topology description for this group. Non-null if
and only if TopologyDescriptionStatus is AVAILABLE (3); null otherwise." },
{ "name": "TopologyDescriptionStatus", "type": "int8", "versions":
"1+", "default": "0",
- "about": "The status of the topology description for this group,
paired with TopologyDescription: 0=NOT_REQUESTED (client did not set
IncludeTopologyDescription; TopologyDescription is null); 1=NOT_STORED (no
description recorded for this group; TopologyDescription is null); 2=ERROR
(broker failed to fetch the description, see broker logs; TopologyDescription
is null); 3=AVAILABLE (TopologyDescription is non-null and carries the
description)." }
+ "about": "The status of the topology description for this group,
paired with TopologyDescription: 0=NOT_REQUESTED (client did not set
IncludeTopologyDescription; TopologyDescription is null); 1=NOT_STORED (no
description recorded for this group; TopologyDescription is null); 2=ERROR
(broker failed to fetch the description, see broker logs; TopologyDescription
is null); 3=AVAILABLE (TopologyDescription is non-null and carries the
description)." },
+ { "name": "AssignorName", "type": "string", "versions": "1+",
"nullableVersions": "1+", "default": "null", "ignorable": true,
+ "about": "The task assignor that the group coordinator will use for
the next assignment computation. This may differ from the assignor that
computed the current assignment, because changing the assignor configuration
does not trigger a rebalance. Null in case of a describe error." }
]
}
],
diff --git
a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClientTest.java
b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClientTest.java
index a9df351c87f..3409e79a8a5 100644
---
a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClientTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClientTest.java
@@ -111,7 +111,8 @@ public class MockAdminClientTest {
new Node(0, "host", 0),
Set.of(),
Optional.empty(),
- StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED);
+ StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
+ Optional.empty());
}
private StreamsGroupDescription
newStreamsGroupDescriptionWithTopology(String groupId) {
@@ -127,6 +128,7 @@ public class MockAdminClientTest {
new Node(0, "host", 0),
Set.of(),
Optional.of(topology),
- StreamsGroupTopologyDescriptionStatus.AVAILABLE);
+ StreamsGroupTopologyDescriptionStatus.AVAILABLE,
+ Optional.empty());
}
}
diff --git
a/clients/src/testFixtures/java/org/apache/kafka/clients/admin/MockAdminClient.java
b/clients/src/testFixtures/java/org/apache/kafka/clients/admin/MockAdminClient.java
index f2f6e4a769b..cc2a37b7aa9 100644
---
a/clients/src/testFixtures/java/org/apache/kafka/clients/admin/MockAdminClient.java
+++
b/clients/src/testFixtures/java/org/apache/kafka/clients/admin/MockAdminClient.java
@@ -1510,7 +1510,8 @@ public class MockAdminClient extends AdminClient {
description.coordinator(),
description.authorizedOperations(),
Optional.empty(),
- StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED
+ StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
+ description.assignorName()
);
}
diff --git a/core/src/main/scala/kafka/server/ConfigHelper.scala
b/core/src/main/scala/kafka/server/ConfigHelper.scala
index 7e4de8eb6fc..a1223218514 100644
--- a/core/src/main/scala/kafka/server/ConfigHelper.scala
+++ b/core/src/main/scala/kafka/server/ConfigHelper.scala
@@ -133,7 +133,7 @@ class ConfigHelper(metadataCache: MetadataCache, config:
KafkaConfig, configRepo
throw new InvalidRequestException("Group name must not be empty")
} else {
val groupProps = configRepository.groupConfig(group)
- val groupConfig =
GroupConfig.fromProps(config.extractGroupConfigMap, groupProps)
+ val groupConfig =
GroupConfig.fromProps(config.extractGroupConfigMap(config.groupCoordinatorConfig),
groupProps)
createResponseConfig(resource, groupConfig,
createGroupConfigEntry(groupConfig, groupProps, includeSynonyms,
includeDocumentation)(_, _))
}
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index 1bed860dfdd..db7dbd8cab6 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -80,7 +80,7 @@ import org.apache.kafka.common.utils.Utils
import org.apache.kafka.common.utils.internals.ImplicitLinkedHashCollection
import org.apache.kafka.common.utils.internals.ProducerIdAndEpoch
import org.apache.kafka.common.utils.internals.SecurityUtils
-import
org.apache.kafka.coordinator.group.GroupConfig.{CONSUMER_ASSIGNMENT_INTERVAL_MS_CONFIG,
CONSUMER_ASSIGNOR_OFFLOAD_ENABLE_CONFIG,
CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, CONSUMER_SESSION_TIMEOUT_MS_CONFIG,
SHARE_ASSIGNMENT_INTERVAL_MS_CONFIG, SHARE_ASSIGNOR_OFFLOAD_ENABLE_CONFIG,
SHARE_AUTO_OFFSET_RESET_CONFIG, SHARE_DELIVERY_COUNT_LIMIT_CONFIG,
SHARE_HEARTBEAT_INTERVAL_MS_CONFIG, SHARE_ISOLATION_LEVEL_CONFIG,
SHARE_PARTITION_MAX_RECORD_LOCKS_CONFIG, SHARE_RECORD_LOCK_DURATION_MS_CO [...]
+import
org.apache.kafka.coordinator.group.GroupConfig.{CONSUMER_ASSIGNMENT_INTERVAL_MS_CONFIG,
CONSUMER_ASSIGNOR_OFFLOAD_ENABLE_CONFIG,
CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, CONSUMER_SESSION_TIMEOUT_MS_CONFIG,
SHARE_ASSIGNMENT_INTERVAL_MS_CONFIG, SHARE_ASSIGNOR_OFFLOAD_ENABLE_CONFIG,
SHARE_AUTO_OFFSET_RESET_CONFIG, SHARE_DELIVERY_COUNT_LIMIT_CONFIG,
SHARE_HEARTBEAT_INTERVAL_MS_CONFIG, SHARE_ISOLATION_LEVEL_CONFIG,
SHARE_PARTITION_MAX_RECORD_LOCKS_CONFIG, SHARE_RECORD_LOCK_DURATION_MS_CO [...]
import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfig
import org.apache.kafka.coordinator.group.{GroupConfig, GroupConfigManager,
GroupCoordinator, GroupCoordinatorConfig}
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult
@@ -385,6 +385,7 @@ class KafkaApisTest extends Logging {
cgConfigs.put(STREAMS_NUM_WARMUP_REPLICAS_CONFIG,
GroupCoordinatorConfig.STREAMS_GROUP_NUM_WARMUP_REPLICAS_DEFAULT.toString)
cgConfigs.put(STREAMS_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG,
GroupCoordinatorConfig.STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT)
cgConfigs.put(STREAMS_ACCEPTABLE_RECOVERY_LAG_CONFIG,
GroupCoordinatorConfig.STREAMS_GROUP_ACCEPTABLE_RECOVERY_LAG_DEFAULT.toString)
+ cgConfigs.put(STREAMS_ASSIGNOR_NAME_CONFIG, "sticky")
cgConfigs.put(ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, "")
cgConfigs.put(ERRORS_DEADLETTERQUEUE_COPY_RECORD_ENABLE_CONFIG, "false")
when(configRepository.groupConfig(consumerGroupId)).thenReturn(cgConfigs)
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index 5cdb38c79af..34f48635581 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -1112,6 +1112,7 @@ class KafkaConfigTest {
case
GroupCoordinatorConfig.STREAMS_GROUP_MIN_TASK_OFFSET_INTERVAL_MS_CONFIG =>
assertPropertyInvalid(baseProperties, name, "not_a_number", -1)
case
GroupCoordinatorConfig.STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG => //
ignore list
case
GroupCoordinatorConfig.STREAMS_GROUP_ACCEPTABLE_RECOVERY_LAG_CONFIG =>
assertPropertyInvalid(baseProperties, name, "not_a_number", -1)
+ case GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG => //
ignore list
/** Share coordinator configs */
case ShareCoordinatorConfig.APPEND_LINGER_MS_CONFIG =>
assertPropertyInvalid(baseProperties, name, "not_a_number", -2, -0.5)
diff --git
a/core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala
b/core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala
index 281abf59e88..23240726a9d 100644
---
a/core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala
+++
b/core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala
@@ -24,6 +24,7 @@ import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.test.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty,
ClusterFeature, ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.coordinator.group.{GroupConfig, GroupCoordinatorConfig}
+import
org.apache.kafka.coordinator.group.api.streams.assignor.{GroupAssignment,
GroupSpec, MemberAssignment, TaskAssignor, TopologyDescriber}
import org.apache.kafka.common.errors.{InvalidConfigurationException,
UnsupportedVersionException}
import org.apache.kafka.server.common.Feature
import org.junit.jupiter.api.Assertions.{assertEquals, assertNotNull,
assertNull, assertThrows, assertTrue}
@@ -926,6 +927,153 @@ class StreamsGroupHeartbeatRequestTest(cluster:
ClusterInstance) extends GroupCo
}
}
+ @ClusterTest
+ def testAlterStreamsAssignorNameGroupConfig(): Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+
+ try {
+ TestUtils.createOffsetsTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq
+ )
+
+ val groupConfigResource = new ConfigResource(ConfigResource.Type.GROUP,
groupId)
+
+ // The config is unset by default, so it falls back to the broker's
default assignor, which is
+ // the name of the first assignor of group.streams.assignors.
+ assertEquals("sticky",
admin.describeConfigs(List(groupConfigResource).asJava).all().get()
+
.get(groupConfigResource).get(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG).value())
+
+ // A name that is not registered on the broker is rejected with
INVALID_CONFIG.
+ val invalidAlterOp = new AlterConfigOp(
+ new ConfigEntry(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG,
"does-not-exist"),
+ AlterConfigOp.OpType.SET
+ )
+ val executionException = assertThrows(classOf[ExecutionException], () =>
+ admin.incrementalAlterConfigs(
+ Map(groupConfigResource ->
List(invalidAlterOp).asJavaCollection).asJava
+ ).all().get()
+ )
+
assertTrue(executionException.getCause.isInstanceOf[InvalidConfigurationException],
+ s"Expected InvalidConfigurationException but got
${executionException.getCause}")
+
assertTrue(executionException.getCause.getMessage.contains("'does-not-exist' is
not a registered task assignor"),
+ s"Unexpected error message: ${executionException.getCause.getMessage}")
+ } finally {
+ admin.close()
+ }
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ // The class name has to be spelled out because annotation values must
be compile-time constants.
+ new ClusterConfigProperty(
+ key = GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
+ value = "sticky,kafka.server.CustomStreamsTaskAssignor"
+ )
+ )
+ )
+ def testAlterStreamsAssignorNameGroupConfigWithCustomAssignor(): Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+
+ try {
+ TestUtils.createOffsetsTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq
+ )
+
+ val groupConfigResource = new ConfigResource(ConfigResource.Type.GROUP,
groupId)
+
+ // The config is unset, so it falls back to the broker's default
assignor, which is the first
+ // entry of group.streams.assignors.
+ assertEquals("sticky",
admin.describeConfigs(List(groupConfigResource).asJava).all().get()
+
.get(groupConfigResource).get(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG).value())
+
+ // A custom assignor registered on the broker is selected by the name it
reports, moving the group
+ // off the default. Group config propagation is asynchronous, so wait
for it.
+ val validAlterOp = new AlterConfigOp(
+ new ConfigEntry(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG,
CustomStreamsTaskAssignor.NAME),
+ AlterConfigOp.OpType.SET
+ )
+ admin.incrementalAlterConfigs(
+ Map(groupConfigResource -> List(validAlterOp).asJavaCollection).asJava
+ ).all().get()
+
+ TestUtils.waitUntilTrue(() => {
+ val describedConfigs =
admin.describeConfigs(List(groupConfigResource).asJava).all().get()
+ val assignorName =
describedConfigs.get(groupConfigResource).get(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG)
+ assignorName != null && assignorName.value() ==
CustomStreamsTaskAssignor.NAME
+ }, s"${GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG} was not updated to the
expected value within the timeout period.")
+
+ // The custom assignor's class name is not a valid selector; only its
name() is.
+ val classNameAlterOp = new AlterConfigOp(
+ new ConfigEntry(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG,
classOf[CustomStreamsTaskAssignor].getName),
+ AlterConfigOp.OpType.SET
+ )
+ val executionException = assertThrows(classOf[ExecutionException], () =>
+ admin.incrementalAlterConfigs(
+ Map(groupConfigResource ->
List(classNameAlterOp).asJavaCollection).asJava
+ ).all().get()
+ )
+
assertTrue(executionException.getCause.isInstanceOf[InvalidConfigurationException],
+ s"Expected InvalidConfigurationException but got
${executionException.getCause}")
+
+ // The alter is rejected at validation time, so the group keeps the
assignor it had selected.
+ assertEquals(CustomStreamsTaskAssignor.NAME,
admin.describeConfigs(List(groupConfigResource).asJava).all().get()
+
.get(groupConfigResource).get(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG).value())
+
+ // Deleting the config returns the group to the broker's default
assignor.
+ val deleteAlterOp = new AlterConfigOp(
+ new ConfigEntry(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, ""),
+ AlterConfigOp.OpType.DELETE
+ )
+ admin.incrementalAlterConfigs(
+ Map(groupConfigResource -> List(deleteAlterOp).asJavaCollection).asJava
+ ).all().get()
+
+ TestUtils.waitUntilTrue(() => {
+ val describedConfigs =
admin.describeConfigs(List(groupConfigResource).asJava).all().get()
+
describedConfigs.get(groupConfigResource).get(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG).value()
== "sticky"
+ }, s"${GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG} was not unset within
the timeout period.")
+ } finally {
+ admin.close()
+ }
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ // The class name has to be spelled out because annotation values must
be compile-time constants.
+ new ClusterConfigProperty(
+ key = GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
+ value = "kafka.server.CustomStreamsTaskAssignor"
+ )
+ )
+ )
+ def testDescribeStreamsAssignorNameGroupConfigFallsBackToAssignorName():
Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+
+ try {
+ TestUtils.createOffsetsTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq
+ )
+
+ val groupConfigResource = new ConfigResource(ConfigResource.Type.GROUP,
groupId)
+
+ // The default assignor is registered by class name, but the group
config selects assignors by
+ // name, so the fallback is the assignor's name and not the configured
class name.
+ assertEquals(CustomStreamsTaskAssignor.NAME,
admin.describeConfigs(List(groupConfigResource).asJava).all().get()
+
.get(groupConfigResource).get(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG).value())
+ } finally {
+ admin.close()
+ }
+ }
+
@ClusterTest(
types = Array(Type.KRAFT),
serverProperties = Array(
@@ -1258,4 +1406,18 @@ class StreamsGroupHeartbeatRequestTest(cluster:
ClusterInstance) extends GroupCo
).asJava)
).asJava)
}
+}
+
+object CustomStreamsTaskAssignor {
+ val NAME = "custom"
+}
+
+/**
+ * Registered on the broker only so that a group can select it by name; it
never computes an assignment.
+ */
+class CustomStreamsTaskAssignor extends TaskAssignor {
+ override def name(): String = CustomStreamsTaskAssignor.NAME
+
+ override def assign(groupSpec: GroupSpec, topologyDescriber:
TopologyDescriber): GroupAssignment =
+ new GroupAssignment(java.util.Map.of[String, MemberAssignment]())
}
\ No newline at end of file
diff --git a/docs/getting-started/upgrade.md b/docs/getting-started/upgrade.md
index 034d157a68a..857e0397dcb 100644
--- a/docs/getting-started/upgrade.md
+++ b/docs/getting-started/upgrade.md
@@ -50,6 +50,7 @@ type: docs
* The `kafka-producer-perf-test.sh` tool now supports `--record-key-range`,
`--key-distribution`, and `--random-seed` options to control the distribution
of record keys. Use `--key-distribution range` for sequential key assignment
(round-robin over the key range) or `--key-distribution random` for random key
selection. The `--random-seed` option allows reproducible benchmark runs when
using random key distribution. For further details, please refer to
[KIP-1299](https://cwiki.apache.or [...]
* Share groups now support dead-letter queue functionality as outlined in
[KIP-1191](https://cwiki.apache.org/confluence/x/fApJFg). Any records which are
released (beyond max delivery count) or rejected by the share consumer become
eligible for DLQ. Share group DLQ gets enabled when the Kafka feature
`share.version` is upgraded to 2. The user can configure a DLQ topic on a share
group by setting the dynamic config `errors.deadletterqueue.topic.name`
(default `""`) to the name of the DL [...]
* Kafka Connect distributed workers now support the
`internal.topics.automatic.creation.enable` configuration (default: `true`).
When set to `false`, Connect will not automatically create internal topics
(offset, config, status, and connector-specific offset topics) and will instead
fail at startup if any of these topics are missing. A new
`connect-internal-topics.sh` tool is also available for manually creating these
topics. For further details, please refer to [KIP-1209](https://cwik [...]
+ * Streams groups now support broker-side custom task assignors, registered
via the new broker configuration `group.streams.assignors` and selected per
group with the new group configuration `streams.assignor.name`. For further
details, please refer to
[KIP-1357](https://cwiki.apache.org/confluence/x/NoSnGQ).
## Upgrading to 4.3.0
diff --git a/docs/streams/developer-guide/streams-rebalance-protocol.md
b/docs/streams/developer-guide/streams-rebalance-protocol.md
index 5f9cc280d58..fb1b07d2fb1 100644
--- a/docs/streams/developer-guide/streams-rebalance-protocol.md
+++ b/docs/streams/developer-guide/streams-rebalance-protocol.md
@@ -39,7 +39,9 @@ The following features are available in the current release:
* **Core Streams Group Rebalance Protocol**: The `group.protocol=streams`
configuration enables the dedicated streams rebalance protocol. This separates
streams groups from consumer groups and provides a streams-specific group
membership lifecycle and metadata management on the broker.
-* **Sticky Task Assignor**: A basic task assignment strategy that minimizes
task movement during rebalances is included.
+* **Sticky Task Assignor**: A basic task assignment strategy that minimizes
task movement during rebalances is included. It is registered under the name
`sticky` and is the default assignor.
+
+* **Custom Task Assignors**: Brokers can be configured with custom task
assignors via `group.streams.assignors`, which takes a list of built-in
assignor names and fully qualified class names of custom `TaskAssignor`
implementations. The first entry is the default assignor. An individual group
selects one of the registered assignors by name with the group configuration
`streams.assignor.name`; when unset, the group uses the first entry of
`group.streams.assignors`. Custom implementations [...]
* **Interactive Query Support**: IQ operations are compatible with the new
streams protocol.
@@ -59,7 +61,9 @@ The following features are not yet available and should be
avoided when using th
* **Topology Updates**: If a topology is changed significantly (e.g., by
adding new source topics or changing the number of subtopologies), a new
streams group must be created.
-* **High Availability Assignor**: Only the sticky assignor is supported. This
implies that "warmup tasks" and rack aware assignment are not supported yet.
+* **High Availability Assignor**: The sticky assignor is the only built-in
assignor and it does not support rack aware assignment, but a custom assignor
registered via `group.streams.assignors` can implement it.
+
+* **Warmup Tasks**: In contrast to the "classic" rebalance protocol, warmup
tasks are not an assignor feature, but the group coordinator would inject
warmup tasks into an assignment. The benefit is, that warmup tasks can be used
independent of the configured assignor. However, warmup task support is not
implemented yet.
* **Regular Expressions**: Pattern-based topic subscription is not supported.
@@ -120,6 +124,7 @@ The following broker configurations control the behavior of
streams groups. For
*
[`group.streams.max.standby.replicas`](/{version}/configuration/broker-configs#brokerconfigs_group.streams.max.standby.replicas):
Maximum for dynamic configurations of the standby replica configuration.
*
[`group.streams.initial.rebalance.delay.ms`](/{version}/configuration/broker-configs#brokerconfigs_group.streams.initial.rebalance.delay.ms):
The first rebalance of a new (ie, previously empty) group is delayed by this
amount to allow more members to join the group.
*
[`group.streams.topology.description.plugin.class`](/{version}/configuration/broker-configs#brokerconfigs_group.streams.topology.description.plugin.class):
The fully qualified class name of a `StreamsGroupTopologyDescriptionPlugin`
implementation. When not set, the [topology description
feature](/{version}/streams/developer-guide/topology-description-plugin/) is
disabled.
+*
[`group.streams.assignors`](/{version}/configuration/broker-configs#brokerconfigs_group.streams.assignors):
The task assignors available to streams groups, as a list of built-in assignor
names and fully qualified class names of custom `TaskAssignor` implementations.
The first entry is the default assignor for groups that do not select one with
`streams.assignor.name`.
## Group Configuration
@@ -133,6 +138,7 @@ The following group-level configurations are available for
streams groups:
*
[`streams.heartbeat.interval.ms`](/{version}/configuration/group-configs#groupconfigs_streams.heartbeat.interval.ms):
The heartbeat interval given to the members.
*
[`streams.num.standby.replicas`](/{version}/configuration/group-configs#groupconfigs_streams.num.standby.replicas):
The number of standby replicas for each task.
*
[`streams.initial.rebalance.delay.ms`](/{version}/configuration/group-configs#groupconfigs_streams.initial.rebalance.delay.ms):
The first rebalance of a group is delayed by this amount to allow more members
to join the group.
+*
[`streams.assignor.name`](/{version}/configuration/group-configs#groupconfigs_streams.assignor.name):
The name of the task assignor to use for this group, which must be one of the
assignors registered on the broker via `group.streams.assignors`. When unset,
the group uses the first entry of `group.streams.assignors`.
### Example: Setting Group-Level Configuration
```
@@ -162,7 +168,7 @@ The following configurations are ignored when the streams
rebalance protocol is
*
[`rack.aware.assignment.strategy`](/{version}/configuration/kafka-streams-configs#streamsconfigs_rack.aware.assignment.strategy)
*
[`rack.aware.assignment.traffic_cost`](/{version}/configuration/kafka-streams-configs#streamsconfigs_rack.aware.assignment.traffic_cost)
*
[`rack.aware.assignment.non_overlap_cost`](/{version}/configuration/kafka-streams-configs#streamsconfigs_rack.aware.assignment.non_overlap_cost)
-*
[`task.assignor.class`](/{version}/configuration/kafka-streams-configs#streamsconfigs_task.assignor.class)
+*
[`task.assignor.class`](/{version}/configuration/kafka-streams-configs#streamsconfigs_task.assignor.class)
(assignment happens on the broker; use `group.streams.assignors` and
`streams.assignor.name` instead)
*
[`session.timeout.ms`](/{version}/configuration/kafka-streams-configs#streamsconfigs_session.timeout.ms)
(use group-level configuration instead)
*
[`heartbeat.interval.ms`](/{version}/configuration/kafka-streams-configs#streamsconfigs_heartbeat.interval.ms)
(use group-level configuration instead)
diff --git a/docs/streams/upgrade-guide.md b/docs/streams/upgrade-guide.md
index 1e1fd54aa21..fa6222fc55d 100644
--- a/docs/streams/upgrade-guide.md
+++ b/docs/streams/upgrade-guide.md
@@ -75,6 +75,8 @@ Kafka Streams now exposes the mapped join key alongside the
`KStream` record key
For applications using the Streams Rebalance Protocol
(`group.protocol=streams`), brokers can now record a human-readable description
of the group's processing topology via a pluggable backend
([KIP-1331](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1331%3A+Streams+Group+Topology+Description+Plugin)).
When the broker configuration
`group.streams.topology.description.plugin.class` is set, Kafka Streams clients
automatically push a description equivalent to `Topology#describe()` t [...]
+For applications using the Streams Rebalance Protocol
(`group.protocol=streams`), the broker-side task assignor is now pluggable
([KIP-1357](https://cwiki.apache.org/confluence/x/NoSnGQ)). Operators register
assignors with the new broker configuration `group.streams.assignors`, which
accepts a list of built-in assignor names and fully qualified class names of
custom `TaskAssignor` implementations; the first entry is the default for
groups that do not select one. An individual group picks [...]
+
## Streams API changes in 4.3.0
**Note:** Kafka Streams 4.3.0 contains a critical native memory leak in the
RocksDB state store layer
([KAFKA-20616](https://issues.apache.org/jira/browse/KAFKA-20616)). The
`ColumnFamilyOptions` for the offsets column family is not closed, and column
family handles can leak on close-path exceptions, which under cascading task
closes (e.g., rebalances or error-triggered recoveries) leads to unbounded
off-heap memory growth and eventual OOM. Users running Kafka Streams should
consider upg [...]
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java
index 24b547a633b..f9e678f2005 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java
@@ -115,6 +115,11 @@ public final class GroupConfig extends AbstractConfig {
public static final String STREAMS_ACCEPTABLE_RECOVERY_LAG_CONFIG =
"streams.acceptable.recovery.lag";
+ public static final String STREAMS_ASSIGNOR_NAME_CONFIG =
"streams.assignor.name";
+ public static final String STREAMS_ASSIGNOR_NAME_DOC = "The task assignor
to use for this streams group, selected by short name from the assignors
registered via the broker configuration " +
+ GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG + ". When unset,
the group defaults to the first assignor configured in the broker's " +
+ GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG + " setting.
Changing the assignor does not trigger a rebalance for the group; the new
assignor takes effect on the next rebalance.";
+
public static final String ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG =
"errors.deadletterqueue.topic.name";
public static final String ERRORS_DEADLETTERQUEUE_TOPIC_NAME_DEFAULT = "";
public static final String ERRORS_DEADLETTERQUEUE_TOPIC_NAME_DOC = "The
name of the topic to be used as the dead-letter queue (DLQ) topic for this
share group. If blank (the default), the group does not have a DLQ topic.";
@@ -167,6 +172,8 @@ public final class GroupConfig extends AbstractConfig {
private final Optional<Long> streamsAcceptableRecoveryLag;
+ private final Optional<String> streamsAssignorName;
+
private final Optional<IsolationLevel> shareIsolationLevel;
private final Optional<Boolean> shareRenewAcknowledgeEnable;
@@ -316,6 +323,11 @@ public final class GroupConfig extends AbstractConfig {
atLeast(0),
MEDIUM,
GroupCoordinatorConfig.STREAMS_GROUP_ACCEPTABLE_RECOVERY_LAG_DOC)
+ .define(STREAMS_ASSIGNOR_NAME_CONFIG,
+ STRING,
+ null,
+ MEDIUM,
+ STREAMS_ASSIGNOR_NAME_DOC)
// DLQ configurations (KIP-1191)
.define(ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG,
@@ -363,6 +375,7 @@ public final class GroupConfig extends AbstractConfig {
Map.entry(STREAMS_NUM_WARMUP_REPLICAS_CONFIG,
Optional.of(GroupCoordinatorConfig.STREAMS_GROUP_NUM_WARMUP_REPLICAS_CONFIG)),
Map.entry(STREAMS_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG,
Optional.of(GroupCoordinatorConfig.STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG)),
Map.entry(STREAMS_ACCEPTABLE_RECOVERY_LAG_CONFIG,
Optional.of(GroupCoordinatorConfig.STREAMS_GROUP_ACCEPTABLE_RECOVERY_LAG_CONFIG)),
+ Map.entry(STREAMS_ASSIGNOR_NAME_CONFIG,
Optional.of(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG)),
// DLQ configs
Map.entry(ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, Optional.empty()),
@@ -414,6 +427,7 @@ public final class GroupConfig extends AbstractConfig {
this.streamsTaskOffsetIntervalMs =
optionalInt(STREAMS_TASK_OFFSET_INTERVAL_MS_CONFIG);
this.streamsNumWarmupReplicas =
optionalInt(STREAMS_NUM_WARMUP_REPLICAS_CONFIG);
this.streamsAcceptableRecoveryLag =
optionalLong(STREAMS_ACCEPTABLE_RECOVERY_LAG_CONFIG);
+ this.streamsAssignorName =
optionalString(STREAMS_ASSIGNOR_NAME_CONFIG);
this.shareIsolationLevel = optionalString(SHARE_ISOLATION_LEVEL_CONFIG)
.map(s -> IsolationLevel.valueOf(s.toUpperCase(Locale.ROOT)));
this.shareRenewAcknowledgeEnable =
optionalBoolean(SHARE_RENEW_ACKNOWLEDGE_ENABLE_CONFIG);
@@ -614,6 +628,16 @@ public final class GroupConfig extends AbstractConfig {
groupCoordinatorConfig.streamsGroupMaxWarmupReplicas()
);
+ // The selected streams assignor must be one of the assignors
registered on the broker.
+ if (parsed.containsKey(STREAMS_ASSIGNOR_NAME_CONFIG)) {
+ String assignorName = (String)
parsed.get(STREAMS_ASSIGNOR_NAME_CONFIG);
+ List<String> registeredAssignors =
groupCoordinatorConfig.streamsGroupAssignorNames();
+ if (!registeredAssignors.contains(assignorName)) {
+ throw new
InvalidConfigurationException(STREAMS_ASSIGNOR_NAME_CONFIG + " '" +
assignorName +
+ "' is not a registered task assignor. Registered assignors
are: " + registeredAssignors + ".");
+ }
+ }
+
// Cross-field validations: session timeout must be greater than
heartbeat interval.
validateSessionExceedsHeartbeat(
parsed,
@@ -1214,6 +1238,13 @@ public final class GroupConfig extends AbstractConfig {
return streamsAcceptableRecoveryLag;
}
+ /**
+ * The task assignor selected for streams groups.
+ */
+ public Optional<String> streamsAssignorName() {
+ return streamsAssignorName;
+ }
+
/**
* The share group isolation level.
*/
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 012ca319878..f88c63d9bea 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
@@ -27,9 +27,11 @@ import org.apache.kafka.common.utils.Utils;
import
org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
import
org.apache.kafka.coordinator.group.api.assignor.ShareGroupPartitionAssignor;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
+import org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignor;
import org.apache.kafka.coordinator.group.assignor.RangeAssignor;
import org.apache.kafka.coordinator.group.assignor.SimpleAssignor;
import org.apache.kafka.coordinator.group.assignor.UniformAssignor;
+import org.apache.kafka.coordinator.group.streams.assignor.StickyTaskAssignor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -388,6 +390,21 @@ public class GroupCoordinatorConfig {
public static final String STREAMS_GROUP_MAX_ASSIGNMENT_INTERVAL_MS_DOC =
"The maximum interval between assignment updates for a streams group.";
public static final int STREAMS_GROUP_MAX_ASSIGNMENT_INTERVAL_MS_DEFAULT =
15000;
+ // The first entry is the default assignor for groups that do not select
one. New built-in
+ // assignors must be appended so that the default does not change for
existing groups.
+ private static final List<TaskAssignor> STREAMS_GROUP_BUILTIN_ASSIGNORS =
List.of(
+ new StickyTaskAssignor()
+ );
+ public static final String STREAMS_GROUP_ASSIGNORS_CONFIG =
"group.streams.assignors";
+ public static final String STREAMS_GROUP_ASSIGNORS_DOC = "The server side
task assignors for streams groups as a list of either names for built-in
assignors or fully qualified class names for custom assignors. " +
+ "The first one in the list is considered as the default assignor to be
used in the case where the streams group does not specify an assignor. " +
+ "Changing the default assignor does not trigger a rebalance for
existing groups; the new default takes effect on the next rebalance. " +
+ "The supported built-in assignors are: " +
STREAMS_GROUP_BUILTIN_ASSIGNORS.stream().map(TaskAssignor::name).collect(Collectors.joining(",
")) + ".";
+ public static final List<String> STREAMS_GROUP_ASSIGNORS_DEFAULT =
STREAMS_GROUP_BUILTIN_ASSIGNORS
+ .stream()
+ .map(TaskAssignor::name)
+ .toList();
+
public static final String STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG
= "group.streams.rack.aware.assignment.tags";
public static final String
STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT = "";
public static final String STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DOC =
"List of client tag keys used to distribute standby replicas across Kafka
Streams instances. When configured, and the used broker-side assignor supports
it, it will make a best-effort to distribute standby tasks over each client tag
dimension.";
@@ -510,6 +527,7 @@ public class GroupCoordinatorConfig {
.define(STREAMS_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, INT,
STREAMS_GROUP_ASSIGNMENT_INTERVAL_MS_DEFAULT, atLeast(0), MEDIUM,
STREAMS_GROUP_ASSIGNMENT_INTERVAL_MS_DOC)
.define(STREAMS_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, INT,
STREAMS_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_DEFAULT, atLeast(0), MEDIUM,
STREAMS_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_DOC)
.define(STREAMS_GROUP_MAX_ASSIGNMENT_INTERVAL_MS_CONFIG, INT,
STREAMS_GROUP_MAX_ASSIGNMENT_INTERVAL_MS_DEFAULT, atLeast(0), MEDIUM,
STREAMS_GROUP_MAX_ASSIGNMENT_INTERVAL_MS_DOC)
+ .define(STREAMS_GROUP_ASSIGNORS_CONFIG, LIST,
STREAMS_GROUP_ASSIGNORS_DEFAULT,
ConfigDef.ValidList.anyNonDuplicateValues(false, false), MEDIUM,
STREAMS_GROUP_ASSIGNORS_DOC)
.define(STREAMS_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, BOOLEAN,
STREAMS_GROUP_ASSIGNOR_OFFLOAD_ENABLE_DEFAULT, MEDIUM,
STREAMS_GROUP_ASSIGNOR_OFFLOAD_ENABLE_DOC)
.define(STREAMS_GROUP_TASK_OFFSET_INTERVAL_MS_CONFIG, INT,
STREAMS_GROUP_TASK_OFFSET_INTERVAL_MS_DEFAULT, atLeast(1), MEDIUM,
STREAMS_GROUP_TASK_OFFSET_INTERVAL_MS_DOC)
.define(STREAMS_GROUP_MIN_TASK_OFFSET_INTERVAL_MS_CONFIG, INT,
STREAMS_GROUP_MIN_TASK_OFFSET_INTERVAL_MS_DEFAULT, atLeast(1), MEDIUM,
STREAMS_GROUP_MIN_TASK_OFFSET_INTERVAL_MS_DOC)
@@ -584,6 +602,8 @@ public class GroupCoordinatorConfig {
private final int streamsGroupMaxWarmupReplicas;
private final List<String> streamsGroupRackAwareAssignmentTags;
private final long streamsGroupAcceptableRecoveryLag;
+ private final List<TaskAssignor> streamsGroupAssignors;
+ private final List<String> streamsGroupAssignorNames;
private final AbstractConfig config;
@@ -652,6 +672,8 @@ public class GroupCoordinatorConfig {
this.streamsGroupNumWarmupReplicas =
config.getInt(GroupCoordinatorConfig.STREAMS_GROUP_NUM_WARMUP_REPLICAS_CONFIG);
this.streamsGroupMaxWarmupReplicas =
config.getInt(GroupCoordinatorConfig.STREAMS_GROUP_MAX_WARMUP_REPLICAS_CONFIG);
this.streamsGroupAcceptableRecoveryLag =
config.getLong(GroupCoordinatorConfig.STREAMS_GROUP_ACCEPTABLE_RECOVERY_LAG_CONFIG);
+ this.streamsGroupAssignors = streamsGroupAssignors(config);
+ this.streamsGroupAssignorNames =
this.streamsGroupAssignors.stream().map(TaskAssignor::name).toList();
this.config = config;
checkConstraints();
@@ -916,6 +938,71 @@ public class GroupCoordinatorConfig {
return assignors;
}
+ protected List<TaskAssignor> streamsGroupAssignors(
+ AbstractConfig config
+ ) {
+ Map<String, TaskAssignor> builtInAssignors =
STREAMS_GROUP_BUILTIN_ASSIGNORS
+ .stream()
+ .collect(Collectors.toMap(TaskAssignor::name,
Function.identity()));
+ // A built-in may be configured either by its name or by its class
name, so it is recognised
+ // by class rather than by how it was resolved below.
+ Set<Class<? extends TaskAssignor>> builtInAssignorClasses =
STREAMS_GROUP_BUILTIN_ASSIGNORS
+ .stream()
+ .map(TaskAssignor::getClass)
+ .collect(Collectors.toSet());
+
+ List<TaskAssignor> assignors = new ArrayList<>();
+ Set<String> assignorNames = new HashSet<>();
+
+ try {
+ // `configuredAssignor` is either the name of a built-in assignor,
+ // or a fully qualified class name of a custom assignor
+ for (String configuredAssignor :
config.getList(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG)) {
+ TaskAssignor assignor =
builtInAssignors.get(configuredAssignor);
+ if (assignor == null) {
+ try {
+ assignor = Utils.newInstance(configuredAssignor,
TaskAssignor.class);
+ } catch (ClassNotFoundException e) {
+ throw new
ConfigException(STREAMS_GROUP_ASSIGNORS_CONFIG, configuredAssignor,
+ "Class cannot be found");
+ } catch (ClassCastException e) {
+ throw new
ConfigException(STREAMS_GROUP_ASSIGNORS_CONFIG, configuredAssignor,
+ "Class is not an instance of " +
TaskAssignor.class.getName());
+ } catch (KafkaException e) {
+ // Utils#newInstance reports instantiation failures,
for example a missing
+ // public no-argument constructor, without naming the
config that caused them.
+ throw new
ConfigException(STREAMS_GROUP_ASSIGNORS_CONFIG, configuredAssignor,
e.getMessage());
+ }
+ }
+
+ assignors.add(assignor);
+
+ if (!builtInAssignorClasses.contains(assignor.getClass()) &&
builtInAssignors.containsKey(assignor.name())) {
+ throw new ConfigException(STREAMS_GROUP_ASSIGNORS_CONFIG,
configuredAssignor,
+ "Assignor name '" + assignor.name() + "' is reserved
by a built-in assignor. " +
+ "A custom assignor must not reuse the name of a
built-in assignor");
+ }
+
+ if (!assignorNames.add(assignor.name())) {
+ throw new ConfigException(STREAMS_GROUP_ASSIGNORS_CONFIG,
configuredAssignor,
+ "Assignor name '" + assignor.name() + "' is already
registered by another configured assignor. " +
+ "Assignor names, whether built-in or custom, must
be unique");
+ }
+
+ if (assignor instanceof Configurable configurable) {
+ configurable.configure(config.originals());
+ }
+ }
+ } catch (Exception e) {
+ for (TaskAssignor assignor : assignors) {
+ maybeCloseQuietly(assignor, "AutoCloseable object constructed
and configured during failed call to streamsGroupAssignors");
+ }
+ throw e;
+ }
+
+ return assignors;
+ }
+
protected List<ShareGroupPartitionAssignor> shareGroupAssignors(
AbstractConfig config
) {
@@ -1468,4 +1555,18 @@ public class GroupCoordinatorConfig {
public long streamsGroupAcceptableRecoveryLag() {
return streamsGroupAcceptableRecoveryLag;
}
+
+ /**
+ * The streams group task assignors.
+ */
+ public List<TaskAssignor> streamsGroupAssignors() {
+ return streamsGroupAssignors;
+ }
+
+ /**
+ * The names of the registered streams group task assignors, in configured
order.
+ */
+ public List<String> streamsGroupAssignorNames() {
+ return streamsGroupAssignorNames;
+ }
}
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 58639a20b70..42279c644d3 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
@@ -274,6 +274,7 @@ public class GroupCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
.withGroupConfigManager(groupConfigManager)
.withGroupCoordinatorMetricsShard(metricsShard)
.withShareGroupAssignor(config.shareGroupAssignors().get(0))
+ .withStreamsGroupAssignors(config.streamsGroupAssignors())
.withAuthorizerPlugin(authorizerPlugin)
.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 e801efc8722..a437e4d10d4 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
@@ -510,6 +510,11 @@ public class GroupMetadataManager {
*/
private final Map<String, TaskAssignor> streamsGroupAssignors;
+ /**
+ * The default streams group task assignor used.
+ */
+ private final TaskAssignor defaultStreamsGroupAssignor;
+
/**
* The metadata image.
*/
@@ -580,6 +585,7 @@ public class GroupMetadataManager {
this.shareGroupStatePartitionMetadata = new
TimelineHashMap<>(snapshotRegistry, 0);
this.groupConfigManager = groupConfigManager;
this.shareGroupAssignor = shareGroupAssignor;
+ this.defaultStreamsGroupAssignor = streamsGroupAssignors.get(0);
this.streamsGroupAssignors =
streamsGroupAssignors.stream().collect(Collectors.toMap(TaskAssignor::name,
Function.identity()));
this.topicRegexResolver = new TopicRegexResolver(() ->
authorizerPlugin, this.time);
this.topicHashCache = new HashMap<>();
@@ -752,7 +758,10 @@ public class GroupMetadataManager {
groupIds.forEach(groupId -> {
try {
StreamsGroup group = streamsGroup(groupId, committedOffset);
- describedGroups.add(group.asDescribedGroup(committedOffset));
+ describedGroups.add(group.asDescribedGroup(
+ committedOffset,
+ streamsGroupAssignor(groupId, false).name()
+ ));
groupIdToStoredDescriptionTopologyEpochs.put(groupId,
group.storedDescriptionTopologyEpoch(committedOffset));
} catch (GroupIdNotFoundException exception) {
describedGroups.add(new
StreamsGroupDescribeResponseData.DescribedGroup()
@@ -4456,7 +4465,7 @@ public class GroupMetadataManager {
return new UpdateTargetAssignmentResult<>(group.assignmentEpoch(),
updatedMembersAndTargetAssignment.targetAssignment());
}
- TaskAssignor assignor = streamsGroupAssignor(group.groupId());
+ TaskAssignor assignor = streamsGroupAssignor(group.groupId(), true);
try {
org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder
assignmentResultBuilder =
new
org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder(
@@ -9806,9 +9815,31 @@ public class GroupMetadataManager {
/**
* Get the assignor of the provided streams group.
+ *
+ * <p>The assignor is selected by the group-level {@link
GroupConfig#STREAMS_ASSIGNOR_NAME_CONFIG}
+ * configuration. When the group does not select an assignor, the broker's
default assignor
+ * (the first entry of {@code group.streams.assignors}) is used. If the
selected assignor is no
+ * longer registered on the broker, the coordinator falls back to the
default.
+ *
+ * @param maybeLogWarning Whether to warn about the fallback. Set to false
on read-only paths such as
+ * describe, which would otherwise log on every
request.
*/
- private TaskAssignor streamsGroupAssignor(String groupId) {
- return streamsGroupAssignors.get("sticky");
+ // Visible for testing
+ TaskAssignor streamsGroupAssignor(String groupId, boolean maybeLogWarning)
{
+ Optional<String> configuredName =
groupConfigManager.groupConfig(groupId)
+ .flatMap(GroupConfig::streamsAssignorName);
+ if (configuredName.isPresent()) {
+ TaskAssignor assignor =
streamsGroupAssignors.get(configuredName.get());
+ if (assignor != null) {
+ return assignor;
+ }
+ if (maybeLogWarning) {
+ log.warn("[GroupId {}] The configured task assignor '{}' is
not available; " +
+ "falling back to the default assignor '{}'.",
+ groupId, configuredName.get(),
defaultStreamsGroupAssignor.name());
+ }
+ }
+ return defaultStreamsGroupAssignor;
}
/**
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
index 8e9116c0f83..3a481459f1c 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
@@ -1312,14 +1312,21 @@ public class StreamsGroup implements Group {
});
}
+ /**
+ * @param committedOffset The last committed offset of this shard.
+ * @param assignorName The name of the assignor that the coordinator
will use for the next assignment
+ * computation, already resolved against the
assignors registered on this broker.
+ */
public StreamsGroupDescribeResponseData.DescribedGroup asDescribedGroup(
- long committedOffset
+ long committedOffset,
+ String assignorName
) {
StreamsGroupDescribeResponseData.DescribedGroup describedGroup = new
StreamsGroupDescribeResponseData.DescribedGroup()
.setGroupId(groupId)
.setGroupEpoch(groupEpoch.get(committedOffset))
.setGroupState(state.get(committedOffset).toString())
.setAssignmentEpoch(targetAssignmentMetadata.get(committedOffset).assignmentEpoch())
+ .setAssignorName(assignorName)
.setTopology(
configuredTopology.get(committedOffset)
.filter(ConfiguredTopology::isReady)
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java
index 5436c7c4466..db64d0945b6 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java
@@ -137,7 +137,10 @@ public class GroupConfigTest {
// This is a free-form list of tag keys, so values like
"not_a_number" are valid. Only an
// empty tag key (an empty element between commas) is rejected.
assertPropertyInvalid(name, "tag1,,tag2");
- } else if
(!GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG.equals(name)) {
+ } else if
(!GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG.equals(name)
+ && !GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG.equals(name))
{
+ // Free-form string configs (no ConfigDef validator) accept
any value at construction
+ // time; their values are validated separately in
GroupConfig.validate.
assertPropertyInvalid(name, "not_a_number", "-0.1");
}
});
@@ -379,6 +382,55 @@ public class GroupConfigTest {
new GroupConfig(whitespaceProps).streamsRackAwareAssignmentTags());
}
+ @Test
+ public void testStreamsAssignorNameValidation() {
+ // A registered assignor name is accepted.
+ Map<String, String> props = createValidGroupConfig();
+ props.put(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "sticky");
+ doTestValidProps(props);
+
+ // An unknown assignor name is rejected with INVALID_CONFIG.
+ props = createValidGroupConfig();
+ props.put(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "does-not-exist");
+ doTestInvalidProps(props, InvalidConfigurationException.class);
+ }
+
+ @Test
+ public void testStreamsAssignorNameSelectsCustomAssignor() {
+ // A custom assignor registered on the broker can be selected by its
name.
+ GroupCoordinatorConfig groupCoordinatorConfig =
createGroupCoordinatorConfig(Map.of(
+ GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
+ "sticky," +
GroupCoordinatorConfigTest.CustomTaskAssignor.class.getName()
+ ));
+
+ Map<String, String> props = createValidGroupConfig();
+ props.put(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG,
"CustomTaskAssignor");
+ assertDoesNotThrow(() -> GroupConfig.validate(props,
groupCoordinatorConfig, createShareGroupConfig()));
+
+ // The built-in assignor is still selectable alongside it.
+ props.put(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "sticky");
+ assertDoesNotThrow(() -> GroupConfig.validate(props,
groupCoordinatorConfig, createShareGroupConfig()));
+
+ // The custom assignor's class name is not a valid selector; only its
name() is.
+ props.put(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG,
GroupCoordinatorConfigTest.CustomTaskAssignor.class.getName());
+ assertThrows(InvalidConfigurationException.class,
+ () -> GroupConfig.validate(props, groupCoordinatorConfig,
createShareGroupConfig()));
+ }
+
+ @Test
+ public void testStreamsAssignorNameEvaluateIsLenient() {
+ // The Admin path (validate) rejects an unknown assignor name...
+ Map<String, String> props = createValidGroupConfig();
+ props.put(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "does-not-exist");
+ doTestInvalidProps(props, InvalidConfigurationException.class);
+
+ // ...but the metadata-replay path (evaluate) accepts it, so a value
that was valid when set
+ // survives a broker restart even if the assignor was later removed
from the broker config.
+ Properties replayed = new Properties();
+ replayed.setProperty(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG,
"does-not-exist");
+ assertDoesNotThrow(() -> GroupConfig.evaluate(replayed, "group",
createGroupCoordinatorConfig(), createShareGroupConfig()));
+ }
+
private void doTestInvalidProps(Map<String, String> props, Class<? extends
Exception> exceptionClassName) {
assertThrows(exceptionClassName, () -> GroupConfig.validate(props,
createGroupCoordinatorConfig(), createShareGroupConfig()));
}
@@ -894,15 +946,21 @@ public class GroupConfigTest {
}
private GroupCoordinatorConfig createGroupCoordinatorConfig() {
+ return createGroupCoordinatorConfig(Map.of());
+ }
+
+ private GroupCoordinatorConfig createGroupCoordinatorConfig(Map<String,
Object> overrides) {
+ Map<String, Object> configs = new HashMap<>(Map.of(
+
GroupCoordinatorConfig.CONSUMER_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, 1000,
+
GroupCoordinatorConfig.SHARE_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, 1000,
+
GroupCoordinatorConfig.STREAMS_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, 1000
+ ));
+ configs.putAll(overrides);
return GroupCoordinatorConfigTest.createGroupCoordinatorConfig(
OFFSET_METADATA_MAX_SIZE,
OFFSETS_RETENTION_CHECK_INTERVAL_MS,
OFFSETS_RETENTION_MINUTES,
- Map.of(
-
GroupCoordinatorConfig.CONSUMER_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, 1000,
-
GroupCoordinatorConfig.SHARE_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, 1000,
-
GroupCoordinatorConfig.STREAMS_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, 1000
- )
+ configs
);
}
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java
index 8443624f83c..5d6eaefe804 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java
@@ -28,9 +28,12 @@ import
org.apache.kafka.coordinator.group.api.assignor.ShareGroupPartitionAssign
import
org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
+import org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignor;
+import
org.apache.kafka.coordinator.group.api.streams.assignor.TopologyDescriber;
import org.apache.kafka.coordinator.group.assignor.RangeAssignor;
import org.apache.kafka.coordinator.group.assignor.SimpleAssignor;
import org.apache.kafka.coordinator.group.assignor.UniformAssignor;
+import org.apache.kafka.coordinator.group.streams.assignor.StickyTaskAssignor;
import org.junit.jupiter.api.Test;
@@ -288,6 +291,200 @@ public class GroupCoordinatorConfigTest {
assertThrows(ConfigException.class, () ->
createConfig(configs)).getMessage());
}
+ public static class CustomTaskAssignor implements TaskAssignor,
Configurable {
+ public Map<String, ?> configs;
+
+ @Override
+ public void configure(Map<String, ?> configs) {
+ this.configs = configs;
+ }
+
+ @Override
+ public String name() {
+ return "CustomTaskAssignor";
+ }
+
+ @Override
+ public
org.apache.kafka.coordinator.group.api.streams.assignor.GroupAssignment assign(
+ org.apache.kafka.coordinator.group.api.streams.assignor.GroupSpec
groupSpec,
+ TopologyDescriber topologyDescriber
+ ) {
+ return null;
+ }
+ }
+
+ public static class NoDefaultConstructorTaskAssignor implements
TaskAssignor {
+ public NoDefaultConstructorTaskAssignor(String unused) {
+ }
+
+ @Override
+ public String name() {
+ return "NoDefaultConstructorTaskAssignor";
+ }
+
+ @Override
+ public
org.apache.kafka.coordinator.group.api.streams.assignor.GroupAssignment assign(
+ org.apache.kafka.coordinator.group.api.streams.assignor.GroupSpec
groupSpec,
+ TopologyDescriber topologyDescriber
+ ) {
+ return null;
+ }
+ }
+
+ public static class DuplicateNameTaskAssignor implements TaskAssignor {
+ @Override
+ public String name() {
+ // Collides with CustomTaskAssignor.
+ return "CustomTaskAssignor";
+ }
+
+ @Override
+ public
org.apache.kafka.coordinator.group.api.streams.assignor.GroupAssignment assign(
+ org.apache.kafka.coordinator.group.api.streams.assignor.GroupSpec
groupSpec,
+ TopologyDescriber topologyDescriber
+ ) {
+ return null;
+ }
+ }
+
+ public static class StickyNamedTaskAssignor implements TaskAssignor {
+ @Override
+ public String name() {
+ // Collides with the built-in "sticky" assignor.
+ return "sticky";
+ }
+
+ @Override
+ public
org.apache.kafka.coordinator.group.api.streams.assignor.GroupAssignment assign(
+ org.apache.kafka.coordinator.group.api.streams.assignor.GroupSpec
groupSpec,
+ TopologyDescriber topologyDescriber
+ ) {
+ return null;
+ }
+ }
+
+ @Test
+ public void testStreamsGroupAssignorFullClassNames() {
+ // The full class name of the assignors is part of our public api.
Hence,
+ // we should ensure that they are not changed by mistake.
+ assertEquals(
+
"org.apache.kafka.coordinator.group.streams.assignor.StickyTaskAssignor",
+ StickyTaskAssignor.class.getName()
+ );
+ }
+
+ @Test
+ public void testStreamsGroupAssignors() {
+ Map<String, Object> configs = new HashMap<>();
+ GroupCoordinatorConfig config;
+ List<TaskAssignor> assignors;
+
+ // Test default config. The default is every built-in assignor, in
declaration order.
+ assertEquals(List.of("sticky"),
GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_DEFAULT);
+ config = createConfig(configs);
+ assignors = config.streamsGroupAssignors();
+ assertEquals(1, assignors.size());
+ assertInstanceOf(StickyTaskAssignor.class, assignors.get(0));
+ assertEquals(List.of("sticky"), config.streamsGroupAssignorNames());
+
+ // Test custom assignor.
+ configs.put(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
CustomTaskAssignor.class.getName());
+ config = createConfig(configs);
+ assignors = config.streamsGroupAssignors();
+ assertEquals(1, assignors.size());
+ assertInstanceOf(CustomTaskAssignor.class, assignors.get(0));
+ assertNotNull(((CustomTaskAssignor) assignors.get(0)).configs);
+
+ // Test a combination (built-in short name and custom class name)
supplied as a programmatic
+ // list of strings.
+ configs.put(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
List.of("sticky", CustomTaskAssignor.class.getName()));
+ config = createConfig(configs);
+ assignors = config.streamsGroupAssignors();
+ assertEquals(2, assignors.size());
+ assertInstanceOf(StickyTaskAssignor.class, assignors.get(0));
+ assertInstanceOf(CustomTaskAssignor.class, assignors.get(1));
+ // The names are reported in configured order, so the first one is the
default.
+ assertEquals(List.of("sticky", "CustomTaskAssignor"),
config.streamsGroupAssignorNames());
+
+ // Test the same combination supplied as a comma-separated string (the
form the broker
+ // config is always delivered in).
+ configs.put(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
"sticky, " + CustomTaskAssignor.class.getName());
+ config = createConfig(configs);
+ assignors = config.streamsGroupAssignors();
+ assertEquals(2, assignors.size());
+ assertInstanceOf(StickyTaskAssignor.class, assignors.get(0));
+ assertInstanceOf(CustomTaskAssignor.class, assignors.get(1));
+ }
+
+ @Test
+ public void testStreamsGroupAssignorsWithDuplicateNamesFails() {
+ // Two custom assignors resolving to the same name must fail startup.
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
+ List.of(CustomTaskAssignor.class.getName(),
DuplicateNameTaskAssignor.class.getName()));
+ assertEquals("Invalid value " +
DuplicateNameTaskAssignor.class.getName() +
+ " for configuration group.streams.assignors: Assignor name
'CustomTaskAssignor' is already " +
+ "registered by another configured assignor. Assignor names,
whether built-in or custom, must be unique",
+ assertThrows(ConfigException.class, () ->
createConfig(configs)).getMessage());
+
+ // Configuring the same built-in twice, once by name and once by class
name, must fail startup.
+ configs.put(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
+ List.of("sticky", StickyTaskAssignor.class.getName()));
+ assertEquals("Invalid value " + StickyTaskAssignor.class.getName() +
+ " for configuration group.streams.assignors: Assignor name
'sticky' is already " +
+ "registered by another configured assignor. Assignor names,
whether built-in or custom, must be unique",
+ assertThrows(ConfigException.class, () ->
createConfig(configs)).getMessage());
+ }
+
+ @Test
+ public void testStreamsGroupAssignorsWithReservedBuiltinNameFails() {
+ // A custom assignor must not take the name of a built-in, whether or
not the built-in is
+ // itself configured: a group selecting that name would otherwise
silently get the custom one.
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
StickyNamedTaskAssignor.class.getName());
+ assertEquals("Invalid value " +
StickyNamedTaskAssignor.class.getName() +
+ " for configuration group.streams.assignors: Assignor name
'sticky' is reserved by a " +
+ "built-in assignor. A custom assignor must not reuse the name
of a built-in assignor",
+ assertThrows(ConfigException.class, () ->
createConfig(configs)).getMessage());
+ }
+
+ @Test
+ public void testStreamsGroupAssignorsBuiltinByClassName() {
+ // A built-in configured by its class name is recognised as the
built-in, not as a custom assignor
+ // reusing the reserved name. The name it is registered under is the
built-in short name.
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
StickyTaskAssignor.class.getName());
+ GroupCoordinatorConfig config = createConfig(configs);
+ List<TaskAssignor> assignors = config.streamsGroupAssignors();
+ assertEquals(1, assignors.size());
+ assertInstanceOf(StickyTaskAssignor.class, assignors.get(0));
+ assertEquals(List.of("sticky"), config.streamsGroupAssignorNames());
+ }
+
+ @Test
+ public void testStreamsGroupAssignorsWithInvalidClassFails() {
+ Map<String, Object> configs = new HashMap<>();
+
+ // An entry that is neither a built-in short name nor a loadable class
name must fail startup.
+ configs.put(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
"org.apache.kafka.NonExistentAssignor");
+ assertEquals("Invalid value org.apache.kafka.NonExistentAssignor for
configuration " +
+ "group.streams.assignors: Class cannot be found",
+ assertThrows(ConfigException.class, () ->
createConfig(configs)).getMessage());
+
+ // Test class that is not an assignor.
+ configs.put(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
Object.class.getName());
+ assertEquals("Invalid value java.lang.Object for configuration
group.streams.assignors: " +
+ "Class is not an instance of
org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignor",
+ assertThrows(ConfigException.class, () ->
createConfig(configs)).getMessage());
+
+ // Test class that cannot be instantiated.
+ configs.put(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
NoDefaultConstructorTaskAssignor.class.getName());
+ assertEquals("Invalid value " +
NoDefaultConstructorTaskAssignor.class.getName() +
+ " for configuration group.streams.assignors: Could not find a
public no-argument constructor for " +
+ NoDefaultConstructorTaskAssignor.class.getName(),
+ assertThrows(ConfigException.class, () ->
createConfig(configs)).getMessage());
+ }
+
@Test
public void testConfigs() {
Map<String, Object> configs = new HashMap<>();
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 b5dfd129c16..7636eb7a202 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
@@ -80,6 +80,7 @@ import org.apache.kafka.common.protocol.ByteBufferAccessor;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.MessageUtil;
import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse.Status;
+import org.apache.kafka.common.utils.LogCaptureAppender;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.common.utils.internals.LogContext;
@@ -248,6 +249,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -11061,6 +11063,7 @@ public class GroupMetadataManagerTest {
.setGroupId(streamsGroupIds.get(0))
.setGroupState(StreamsGroupState.EMPTY.toString())
.setAssignmentEpoch(1)
+ .setAssignorName("mock")
.setTopology(expectedTopology),
new StreamsGroupDescribeResponseData.DescribedGroup()
.setGroupEpoch(epoch)
@@ -11074,6 +11077,7 @@ public class GroupMetadataManagerTest {
.setTopology(expectedTopology)
.setGroupState(StreamsGroupState.NOT_READY.toString())
.setAssignmentEpoch(1)
+ .setAssignorName("mock")
);
List<StreamsGroupDescribeResponseData.DescribedGroup> actual =
context.sendStreamsGroupDescribe(streamsGroupIds);
@@ -11229,11 +11233,41 @@ public class GroupMetadataManagerTest {
)
.setGroupState(StreamsGroup.StreamsGroupState.ASSIGNING.toString())
.setGroupEpoch(epoch + 2)
- .setAssignmentEpoch(epoch + 1);
+ .setAssignmentEpoch(epoch + 1)
+ .setAssignorName("mock");
assertEquals(1, actual.size());
assertEquals(describedGroup, actual.get(0));
}
+ @Test
+ public void testStreamsGroupDescribeReportsAssignorSelectedByGroupConfig()
{
+ String groupId = "group-id";
+ String subtopology1 = "subtopology1";
+ StreamsTopology topology = new StreamsTopology(
+ 0,
+ Map.of(subtopology1,
+ new StreamsGroupTopologyValue.Subtopology()
+ .setSubtopologyId(subtopology1)
+ .setSourceTopics(List.of("foo"))
+ )
+ );
+
+ // The broker registers two assignors; the first ("sticky") is the
default.
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new
MockTaskAssignor("sticky"), new MockTaskAssignor("custom")))
+ .withStreamsGroup(new StreamsGroupBuilder(groupId,
10).withTopology(topology))
+ .build();
+
+ Properties groupConfig = new Properties();
+ groupConfig.setProperty(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG,
"custom");
+ context.updateGroupConfig(groupId, groupConfig);
+
+ List<StreamsGroupDescribeResponseData.DescribedGroup> described =
context.sendStreamsGroupDescribe(List.of(groupId));
+
+ assertEquals(1, described.size());
+ assertEquals("custom", described.get(0).assignorName());
+ }
+
@Test
public void testFinalizeAfterDeleteClearsUncertainToNone() {
// The plugin.deleteTopology completed and stored is still UNCERTAIN:
no push raced, the
@@ -18738,6 +18772,79 @@ public class GroupMetadataManagerTest {
assertEquals(String.format("Member %s is not a member of group %s.",
memberId2, groupId), e.getMessage());
}
+ @Test
+ public void testStreamsGroupUsesBrokerDefaultAssignorWhenGroupUnset() {
+ // The broker registers two assignors; the first ("sticky") is the
default.
+ MockTaskAssignor defaultAssignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(defaultAssignor, new
MockTaskAssignor("custom")))
+ .build();
+
+ // The group does not select an assignor (streams.assignor.name is
unset).
+ assertSame(defaultAssignor,
context.groupMetadataManager.streamsGroupAssignor("fooup", true));
+ }
+
+ @Test
+ public void
testStreamsGroupDefaultsToFirstAssignorWhenNoStickyConfigured() {
+ // The broker registers only a custom assignor (no built-in "sticky");
it is the default.
+ MockTaskAssignor customAssignor = new MockTaskAssignor("custom");
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(customAssignor))
+ .build();
+
+ // Resolution must not assume that a "sticky" assignor is registered.
+ assertSame(customAssignor,
context.groupMetadataManager.streamsGroupAssignor("fooup", true));
+ }
+
+ @Test
+ public void testStreamsGroupAssignorSelectedByGroupConfig() {
+ String groupId = "fooup";
+
+ MockTaskAssignor customAssignor = new MockTaskAssignor("custom");
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new
MockTaskAssignor("sticky"), customAssignor))
+ .build();
+
+ // The group selects the custom assignor by name.
+ Properties groupConfig = new Properties();
+ groupConfig.setProperty(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG,
"custom");
+ context.updateGroupConfig(groupId, groupConfig);
+
+ assertSame(customAssignor,
context.groupMetadataManager.streamsGroupAssignor(groupId, true));
+ }
+
+ @Test
+ public void testStreamsGroupAssignorFallsBackToDefaultWhenUnavailable() {
+ String groupId = "fooup";
+
+ MockTaskAssignor defaultAssignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(defaultAssignor, new
MockTaskAssignor("custom")))
+ .build();
+
+ // The group selects an assignor that is not registered on the broker.
+ Properties groupConfig = new Properties();
+ groupConfig.setProperty(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG,
"does-not-exist");
+ context.updateGroupConfig(groupId, groupConfig);
+
+ try (LogCaptureAppender appender =
LogCaptureAppender.createAndRegister(GroupMetadataManager.class)) {
+ // The coordinator falls back to the default (first) assignor.
+ assertSame(defaultAssignor,
context.groupMetadataManager.streamsGroupAssignor(groupId, true));
+
+ // A warning names the unavailable assignor and the fallback.
+ assertEquals(1, appender.getMessages("WARN").stream()
+ .filter(msg -> msg.contains("The configured task assignor
'does-not-exist' is not available"))
+ .count());
+ }
+
+ try (LogCaptureAppender appender =
LogCaptureAppender.createAndRegister(GroupMetadataManager.class)) {
+ // Read-only paths such as describe resolve the same fallback
without warning.
+ assertSame(defaultAssignor,
context.groupMetadataManager.streamsGroupAssignor(groupId, false));
+
+ assertEquals(List.of(), appender.getMessages("WARN"));
+ }
+ }
+
@Test
public void testStreamsGroupMemberEpochValidation() {
String groupId = "fooup";
@@ -19725,7 +19832,8 @@ public class GroupMetadataManagerTest {
TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2, 3,
4, 5)), MemberTaskOffsets.EMPTY)
))
.setGroupState(StreamsGroupState.STABLE.toString())
- .setGroupEpoch(2);
+ .setGroupEpoch(2)
+ .setAssignorName("sticky");
assertEquals(1, actualDescribedGroups.size());
assertEquals(expectedDescribedGroup, actualDescribedGroups.get(0));
}
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
index 82a5e8b7765..a4bc5bdad8f 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
@@ -1029,6 +1029,7 @@ public class StreamsGroupTest {
.setGroupEpoch(1)
.setTopology(new
StreamsGroupDescribeResponseData.Topology().setEpoch(1).setSubtopologies(List.of()))
.setAssignmentEpoch(1)
+ .setAssignorName("sticky")
.setMembers(Arrays.asList(
new StreamsGroupDescribeResponseData.Member()
.setMemberId("member1")
@@ -1057,7 +1058,7 @@ public class StreamsGroupTest {
.setAssignment(new
StreamsGroupDescribeResponseData.Assignment())
.setTargetAssignment(new
StreamsGroupDescribeResponseData.Assignment())
));
- StreamsGroupDescribeResponseData.DescribedGroup actual =
group.asDescribedGroup(1);
+ StreamsGroupDescribeResponseData.DescribedGroup actual =
group.asDescribedGroup(1, "sticky");
assertEquals(expected, actual);
}
@@ -1237,12 +1238,13 @@ public class StreamsGroupTest {
));
snapshotRegistry.idempotentCreateSnapshot(1);
- StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
group.asDescribedGroup(1);
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
group.asDescribedGroup(1, "sticky");
assertEquals("group-id-with-topology", describedGroup.groupId());
assertEquals(StreamsGroup.StreamsGroupState.NOT_READY.toString(),
describedGroup.groupState());
assertEquals(2, describedGroup.groupEpoch());
assertEquals(2, describedGroup.assignmentEpoch());
+ assertEquals("sticky", describedGroup.assignorName());
// Verify topology is correctly described
assertNotNull(describedGroup.topology());
@@ -1288,7 +1290,7 @@ public class StreamsGroupTest {
group.setTargetAssignmentMetadata(3, 12345L);
snapshotRegistry.idempotentCreateSnapshot(1);
- StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
group.asDescribedGroup(1);
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
group.asDescribedGroup(1, "sticky");
// Should prefer ConfiguredTopology over StreamsTopology
assertNotNull(describedGroup.topology());
@@ -1315,7 +1317,7 @@ public class StreamsGroupTest {
group.setTargetAssignmentMetadata(4, 12345L);
snapshotRegistry.idempotentCreateSnapshot(1);
- StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
group.asDescribedGroup(1);
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
group.asDescribedGroup(1, "sticky");
// Should use StreamsTopology when ConfiguredTopology is not available
assertNotNull(describedGroup.topology());
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 d60543b2edd..6220651ed03 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
@@ -654,9 +654,11 @@ public abstract class AbstractKafkaConfig extends
AbstractConfig {
* defaults when building a {@link GroupConfig} for {@code
DescribeConfigs}.
* Internal group configs are excluded unless their broker synonym was
explicitly configured.
*
+ * @param groupCoordinatorConfig The group coordinator config, used to
resolve defaults that are
+ * not the plain value of the broker synonym.
* @return a map of group config names to their corresponding broker-level
values
*/
- public Map<String, Object> extractGroupConfigMap() {
+ public Map<String, Object> extractGroupConfigMap(GroupCoordinatorConfig
groupCoordinatorConfig) {
Map<String, Object> defaults = new HashMap<>();
Map<String, Object> brokerOriginals = originals();
GroupConfig.configNames().forEach(groupConfigName ->
@@ -667,6 +669,12 @@ public abstract class AbstractKafkaConfig extends
AbstractConfig {
}
})
);
+ // The group config holds a single assignor name, whereas the broker
config is a list that may also
+ // use class names, so the default is the name of the first registered
assignor.
+ defaults.computeIfPresent(
+ GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG,
+ (groupConfigName, brokerValue) ->
groupCoordinatorConfig.streamsGroupAssignorNames().get(0)
+ );
return defaults;
}
}
diff --git
a/server/src/test/java/org/apache/kafka/server/config/AbstractKafkaConfigTest.java
b/server/src/test/java/org/apache/kafka/server/config/AbstractKafkaConfigTest.java
index 821ab40e2cc..fae45548fa8 100644
---
a/server/src/test/java/org/apache/kafka/server/config/AbstractKafkaConfigTest.java
+++
b/server/src/test/java/org/apache/kafka/server/config/AbstractKafkaConfigTest.java
@@ -16,8 +16,14 @@
*/
package org.apache.kafka.server.config;
+import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.group.api.streams.assignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.api.streams.assignor.GroupSpec;
+import org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignor;
+import
org.apache.kafka.coordinator.group.api.streams.assignor.TopologyDescriber;
import org.apache.kafka.raft.KRaftConfigs;
import org.junit.jupiter.api.Test;
@@ -33,6 +39,7 @@ import java.util.Set;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
public class AbstractKafkaConfigTest {
@@ -88,6 +95,49 @@ public class AbstractKafkaConfigTest {
assertEquals("default-value", config.get(TEST_INTERNAL_GROUP_CONFIG));
}
+ @Test
+ public void testExtractGroupConfigMapReturnsStreamsAssignor() {
+ try (MockedStatic<GroupConfig> mocked = mockStatic(GroupConfig.class,
Mockito.CALLS_REAL_METHODS)) {
+
mocked.when(GroupConfig::configNames).thenReturn(Set.of(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG));
+
+ // The broker synonym of the group config is read from the broker
config, so it must be defined.
+ AbstractKafkaConfig kafkaConfig =
+ new AbstractKafkaConfig(GroupCoordinatorConfig.CONFIG_DEF,
Map.of(), Map.of(), false) { };
+
+ // The group config holds a single assignor name, so the default
is the first entry of the
+ // broker's list rather than the whole list.
+ assertEquals("sticky",
+
kafkaConfig.extractGroupConfigMap(groupCoordinatorConfigWithStreamsAssignors("sticky,"
+ CustomTaskAssignor.class.getName()))
+ .get(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG));
+
+ // A custom assignor is configured by class name, but a group
selects it by name(), so the
+ // default must be the name and not the class name.
+ assertEquals("CustomTaskAssignor",
+
kafkaConfig.extractGroupConfigMap(groupCoordinatorConfigWithStreamsAssignors(CustomTaskAssignor.class.getName()
+ ",sticky"))
+ .get(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG));
+ }
+ }
+
+ private static GroupCoordinatorConfig
groupCoordinatorConfigWithStreamsAssignors(String assignors) {
+ return new GroupCoordinatorConfig(new AbstractConfig(
+ GroupCoordinatorConfig.CONFIG_DEF,
+ Map.of(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG,
assignors),
+ false
+ ));
+ }
+
+ public static class CustomTaskAssignor implements TaskAssignor {
+ @Override
+ public String name() {
+ return "CustomTaskAssignor";
+ }
+
+ @Override
+ public GroupAssignment assign(GroupSpec groupSpec, TopologyDescriber
topologyDescriber) {
+ return null;
+ }
+ }
+
private static Map<String, Object> extractGroupConfigMap(Map<String,
Object> brokerProps, boolean isInternal) {
try (MockedStatic<GroupConfig> mocked = mockStatic(GroupConfig.class,
Mockito.CALLS_REAL_METHODS)) {
@@ -102,7 +152,8 @@ public class AbstractKafkaConfigTest {
AbstractKafkaConfig kafkaConfig = new
AbstractKafkaConfig(configDef, new HashMap<>(brokerProps), Map.of(), false) { };
- return kafkaConfig.extractGroupConfigMap();
+ // The test config is not the streams assignors config, so the
coordinator config is never consulted.
+ return
kafkaConfig.extractGroupConfigMap(mock(GroupCoordinatorConfig.class));
}
}
}
diff --git
a/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java
b/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java
index 083e71e3019..3bd2b50ede0 100644
---
a/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java
+++
b/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java
@@ -414,15 +414,17 @@ public class StreamsGroupCommand {
final int coordinatorLen = Math.max(25, coordinator.length());
final int stateLen = 25;
+ String assignor = description.assignorName().orElse("");
+ final int assignorLen = Math.max(15, assignor.length());
if (!verbose) {
- String fmt = "%" + -groupLen + "s %" + -coordinatorLen + "s %"
+ -stateLen + "s %s\n";
- System.out.printf(fmt, "GROUP", "COORDINATOR (ID)", "STATE",
"#MEMBERS");
- System.out.printf(fmt, description.groupId(), coordinator,
description.groupState().toString(), description.members().size());
+ String fmt = "%" + -groupLen + "s %" + -coordinatorLen + "s %"
+ -assignorLen + "s %" + -stateLen + "s %s\n";
+ System.out.printf(fmt, "GROUP", "COORDINATOR (ID)",
"ASSIGNOR", "STATE", "#MEMBERS");
+ System.out.printf(fmt, description.groupId(), coordinator,
assignor, description.groupState().toString(), description.members().size());
} else {
final int groupEpochLen = 15, targetAssignmentEpochLen = 25;
- String fmt = "%" + -groupLen + "s %" + -coordinatorLen + "s %"
+ -stateLen + "s %" + -groupEpochLen + "s %" + -targetAssignmentEpochLen + "s
%s\n";
- System.out.printf(fmt, "GROUP", "COORDINATOR (ID)", "STATE",
"GROUP-EPOCH", "TARGET-ASSIGNMENT-EPOCH", "#MEMBERS");
- System.out.printf(fmt, description.groupId(), coordinator,
description.groupState().toString(), description.groupEpoch(),
description.targetAssignmentEpoch(), description.members().size());
+ String fmt = "%" + -groupLen + "s %" + -coordinatorLen + "s %"
+ -assignorLen + "s %" + -stateLen + "s %" + -groupEpochLen + "s %" +
-targetAssignmentEpochLen + "s %s\n";
+ System.out.printf(fmt, "GROUP", "COORDINATOR (ID)",
"ASSIGNOR", "STATE", "GROUP-EPOCH", "TARGET-ASSIGNMENT-EPOCH", "#MEMBERS");
+ System.out.printf(fmt, description.groupId(), coordinator,
assignor, description.groupState().toString(), description.groupEpoch(),
description.targetAssignmentEpoch(), description.members().size());
}
}
diff --git
a/tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java
b/tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java
index 393c468cb62..c2f3878cc7b 100644
---
a/tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java
+++
b/tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java
@@ -160,8 +160,8 @@ public class DescribeStreamsGroupTest {
}
private static void assertDescribeStreamsGroupWithStateOption(String
clusterBootstrapServers) throws Exception {
- final List<String> expectedHeader = List.of("GROUP", "COORDINATOR",
"(ID)", "STATE", "#MEMBERS");
- final Set<List<String>> expectedRows = Set.of(List.of(APP_ID, "", "",
"Stable", "2"));
+ final List<String> expectedHeader = List.of("GROUP", "COORDINATOR",
"(ID)", "ASSIGNOR", "STATE", "#MEMBERS");
+ final Set<List<String>> expectedRows = Set.of(List.of(APP_ID, "", "",
"sticky", "Stable", "2"));
// The coordinator is not deterministic, so we don't care about it.
final List<Integer> dontCares = List.of(1, 2);
@@ -170,11 +170,11 @@ public class DescribeStreamsGroupTest {
}
private static void
assertDescribeStreamsGroupWithStateAndVerboseOptions(String
clusterBootstrapServers) throws Exception {
- final List<String> expectedHeader = List.of("GROUP", "COORDINATOR",
"(ID)", "STATE", "GROUP-EPOCH", "TARGET-ASSIGNMENT-EPOCH", "#MEMBERS");
- final Set<List<String>> expectedRows = Set.of(List.of(APP_ID, "", "",
"Stable", "", "", "2"));
+ final List<String> expectedHeader = List.of("GROUP", "COORDINATOR",
"(ID)", "ASSIGNOR", "STATE", "GROUP-EPOCH", "TARGET-ASSIGNMENT-EPOCH",
"#MEMBERS");
+ final Set<List<String>> expectedRows = Set.of(List.of(APP_ID, "", "",
"sticky", "Stable", "", "", "2"));
// The coordinator is not deterministic, so we don't care about it.
// The GROUP-EPOCH and TARGET-ASSIGNMENT-EPOCH can vary due to
rebalance timing, so we don't care about them either.
- final List<Integer> dontCares = List.of(1, 2, 4, 5);
+ final List<Integer> dontCares = List.of(1, 2, 5, 6);
validateDescribeOutput(
List.of("--bootstrap-server", clusterBootstrapServers,
"--describe", "--state", "--verbose", "--group", APP_ID), expectedHeader,
expectedRows, dontCares);
diff --git
a/tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java
b/tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java
index 55e2dde048e..9393e841fad 100644
---
a/tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java
+++
b/tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java
@@ -180,7 +180,10 @@ public class StreamsGroupCommandTest {
List.of(),
GroupState.STABLE,
new Node(0, "bar", 0),
- null);
+ null,
+ Optional.empty(),
+ StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
+ Optional.empty());
resultMap.put(firstGroup, exp);
when(result.all()).thenReturn(KafkaFuture.completedFuture(resultMap));
when(ADMIN_CLIENT.describeStreamsGroups(anyCollection(),
any(DescribeStreamsGroupsOptions.class))).thenReturn(result);
@@ -204,7 +207,7 @@ public class StreamsGroupCommandTest {
List.of());
StreamsGroupDescription exp = new StreamsGroupDescription(
group, 0, 0, 0, List.of(), List.of(), GroupState.STABLE, new
Node(0, "bar", 0), null,
- Optional.of(topology),
StreamsGroupTopologyDescriptionStatus.AVAILABLE);
+ Optional.of(topology),
StreamsGroupTopologyDescriptionStatus.AVAILABLE, Optional.empty());
Admin admin = mock(KafkaAdminClient.class);
DescribeStreamsGroupsResult result =
mock(DescribeStreamsGroupsResult.class);
@@ -236,7 +239,7 @@ public class StreamsGroupCommandTest {
String group = "foo-group";
StreamsGroupDescription exp = new StreamsGroupDescription(
group, 0, 0, 0, List.of(), List.of(), GroupState.STABLE, new
Node(0, "bar", 0), null,
- Optional.empty(),
StreamsGroupTopologyDescriptionStatus.NOT_STORED);
+ Optional.empty(),
StreamsGroupTopologyDescriptionStatus.NOT_STORED, Optional.empty());
Admin admin = mock(KafkaAdminClient.class);
DescribeStreamsGroupsResult result =
mock(DescribeStreamsGroupsResult.class);
@@ -298,7 +301,10 @@ public class StreamsGroupCommandTest {
List.of(description),
GroupState.STABLE,
new Node(0, "host", 0),
- null);
+ null,
+ Optional.empty(),
+ StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
+ Optional.empty());
StreamsGroupCommandOptions streamsGroupCommandOptions = new
StreamsGroupCommandOptions(
new String[]{"--bootstrap-server", BOOTSTRAP_SERVERS, "--group",
groupId, "--describe"});
@@ -428,7 +434,10 @@ public class StreamsGroupCommandTest {
List.of(),
GroupState.DEAD,
new Node(0, "localhost", 9092),
- null));
+ null,
+ Optional.empty(),
+ StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
+ Optional.empty()));
DescribeStreamsGroupsResult result =
mock(DescribeStreamsGroupsResult.class);
when(result.all()).thenReturn(KafkaFuture.completedFuture(resultMap));
when(ADMIN_CLIENT.describeStreamsGroups(anyCollection(),
any(DescribeStreamsGroupsOptions.class))).thenReturn(result);
@@ -602,7 +611,10 @@ public class StreamsGroupCommandTest {
List.of(),
GroupState.STABLE,
new Node(0, "localhost", 9092),
- null
+ null,
+ Optional.empty(),
+ StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
+ Optional.empty()
);
when(describeResult.all()).thenReturn(KafkaFuture.completedFuture(Map.of(groupId,
groupDescription)));
when(ADMIN_CLIENT.describeStreamsGroups(eq(List.of(groupId)),
any(DescribeStreamsGroupsOptions.class)))
@@ -673,7 +685,10 @@ public class StreamsGroupCommandTest {
List.of(),
GroupState.STABLE,
new Node(0, "localhost", 9092),
- null
+ null,
+ Optional.empty(),
+ StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
+ Optional.empty()
);
when(describeResult.all()).thenReturn(KafkaFuture.completedFuture(Map.of(groupId,
groupDescription)));
when(ADMIN_CLIENT.describeStreamsGroups(eq(List.of(groupId)),
any(DescribeStreamsGroupsOptions.class)))
@@ -753,7 +768,10 @@ public class StreamsGroupCommandTest {
List.of(memberDescription),
groupState,
new Node(1, "localhost", 9092),
- Set.of());
+ Set.of(),
+ Optional.empty(),
+ StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
+ Optional.empty());
KafkaFutureImpl<StreamsGroupDescription> future = new
KafkaFutureImpl<>();
future.complete(description);
return new DescribeStreamsGroupsResult(Map.of(groupId, future));
