C0urante commented on code in PR #15180:
URL: https://github.com/apache/kafka/pull/15180#discussion_r1450799667
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -391,7 +391,15 @@ public List<Map<String, String>>
connectorTaskConfigs(String connName, Connector
Connector connector = workerConnector.connector();
try (LoaderSwap loaderSwap =
plugins.withClassLoader(workerConnector.loader())) {
String taskClassName = connector.taskClass().getName();
- for (Map<String, String> taskProps :
connector.taskConfigs(maxTasks)) {
+ List<Map<String, String>> taskConfigs =
connector.taskConfigs(maxTasks);
+ try {
+ checkTasksMax(connName, taskConfigs.size(), maxTasks,
connConfig.enforceTasksMax());
+ } catch (TooManyTasksException e) {
+ // TODO: This control flow is awkward. Push task config
generation into WorkerConnector class?
Review Comment:
I think it makes sense as a follow-up, because the awkwardness of the
control flow mostly stems from having to trigger a call to
`WorkerConnector::onFailure` from a different thread than the one that `run` is
being executed on. It'd probably make sense to tackle that and making the
herder thread resilient against blocks in `Connector::taskConfigs` in the same
PR.
Does that sound alright?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
