You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ab...@apache.org on 2024/06/09 02:56:42 UTC
(kafka) branch 3.8 updated: KAFKA-15045: (KIP-924 pt. 20) Custom task assignment configuration fix (#16245)
This is an automated email from the ASF dual-hosted git repository.
ableegoldman pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.8 by this push:
new b0333f2ad5c KAFKA-15045: (KIP-924 pt. 20) Custom task assignment configuration fix (#16245)
b0333f2ad5c is described below
commit b0333f2ad5c20223100e7dd2944890bb1fc3aa8b
Author: Antoine Pourchet <an...@gmail.com>
AuthorDate: Sat Jun 8 20:55:13 2024 -0600
KAFKA-15045: (KIP-924 pt. 20) Custom task assignment configuration fix (#16245)
The StreamsConfig class was not parsing the new task assignment
configuration flag properly, which made it impossible to properly
configure a custom task assignor.
This PR fixes this and adds a bit of INFO logging to help users diagnose
assignor misconfiguration issues.
Reviewers: Anna Sophie Blee-Goldman <ab...@apache.org>
---
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java | 1 +
.../streams/processor/internals/assignment/AssignorConfiguration.java | 1 +
2 files changed, 2 insertions(+)
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
index 025a8a5b393..ffeb4105cf1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -1649,6 +1649,7 @@ public class StreamsConfig extends AbstractConfig {
consumerProps.put(RACK_AWARE_ASSIGNMENT_STRATEGY_CONFIG, getString(RACK_AWARE_ASSIGNMENT_STRATEGY_CONFIG));
consumerProps.put(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG, getList(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG));
consumerProps.put(RACK_AWARE_ASSIGNMENT_TRAFFIC_COST_CONFIG, getInt(RACK_AWARE_ASSIGNMENT_TRAFFIC_COST_CONFIG));
+ consumerProps.put(TASK_ASSIGNOR_CLASS_CONFIG, getString(TASK_ASSIGNOR_CLASS_CONFIG));
// disable auto topic creation
consumerProps.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, "false");
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignorConfiguration.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignorConfiguration.java
index 7004ede98b3..1971642942a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignorConfiguration.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignorConfiguration.java
@@ -256,6 +256,7 @@ public final class AssignorConfiguration {
public Optional<org.apache.kafka.streams.processor.assignment.TaskAssignor> customTaskAssignor() {
final String userTaskAssignorClassname = streamsConfig.getString(StreamsConfig.TASK_ASSIGNOR_CLASS_CONFIG);
if (userTaskAssignorClassname == null) {
+ log.info("No custom task assignors found, defaulting to internal task assignment with {}", INTERNAL_TASK_ASSIGNOR_CLASS);
return Optional.empty();
}
try {