Hi Greg, Thanks for the response.
This is detailed in the KIP two sentences earlier: "If the > connect.protocol is set to static, each worker will send it's > static.connectors and static.tasks to the coordinator during > rebalances." Yes I saw that. My point was if this is going to need a change in the Connect Protocol format, then those changes should be outlined in the KIP. Yes. Arbitrary here just means that the assignment is not > influenced by the static assignment. Okay, the usage of the word arbitrary is a bit confusing but I would leave it to that. Also, when I read Mickael's suggestion above about using a taint/selector mechanism, it seems cleaner. The main reason IMO is that we are letting connectors and tasks define where and how they want to be placed and we let the runtime decide it(aided with a ResourceManager if needed) . With the approach proposed in the KIP, anytime any changes are needed to the placements, we need to modify worker configs and add/edit/remove connectors/tasks on it. This hasn't generally been the case and in my experience worker configs are not really modified that often. Moreover, administrators will need to also choose where to place the connectors/tasks based on the load on the cluster which again is an added overhead. However you brought up a good point in the KIP However, because tasks within a single connector can be heterogeneous, it > is necessary to attach a different selector/affinity declaration to each > task. But because tasks are dynamically created by the connector, either > every task's selector must be added manually after defaulting to a wildcard > affinity, or the Connect REST API would need a way to template multiple > task affinities (parameterized by task-id). as well as here above. If we consider the idea proposed by Mickael above, and consider just the isolated config value for selector, which means that any task being spun up for this connector would be run on a dedicated worker. If we use this, then we might see a proliferation of workers happening on the cluster so if we consider adding another worker level config called max.tasks or something similar and if we couple the 2, in effect we can achieve the same thing as the KIP propoeses (obviously I might be over-simplifying it a bit here, but I hope the idea is clear). The config values suggested above should cover all cases, but I still do agree to the fact that setting the selector config value at a task level is a challenge. Thanks! Sagar. On Fri, Oct 20, 2023 at 10:43 PM Greg Harris <greg.har...@aiven.io.invalid> wrote: > Hey Hector, > > That's a cool idea for the ConnectAssignor plugin. > > While this proposal could be viewed as an "assignor" problem that a > custom assignor could solve, it's really about providing additional > context to the assignor which isn't present currently. This lack of > context would prevent a custom assignor from solving the resource > utilization problem adequately. > > Thanks! > Greg > > On Fri, Oct 20, 2023 at 9:58 AM Greg Harris <greg.har...@aiven.io> wrote: > > > > Mickael, > > > > Thank you for discussing that rejected alternative, I was almost going > > to propose it. > > > > > I still find the proposed mechanism limited and I'm not sure it really > addressed the pain points I've experienced with Connect. > > > > I think that this KIP on its own is insufficient to solve the > > operational difficulties of Connect, and an external management layer > > is necessary. In this KIP i'm trying to find the minimum viable > > abstraction to allow a management layer to make decisions about > > placement, knowing that the abstraction may be non-ergonomic for > > "direct users" without a management layer mediating. > > > > > Connectors may also change the assignment of tasks at runtime so for > example it task-2 is really busy (because it's assigned a partition with > high throughput), this may not be true in 10 minutes as this partition is > now assigned to task-1 > > > > I think this is similar to a concern (#5) that Tom raised, and a > > limitation of the "task index" abstraction. I don't know if there is a > > way for us to manage this sort of fine-grained dynamic utilization of > > tasks. Once we start a task, it has some static resources assigned to > > it (via the JVM). If you notice the resource requirements expand, it > > will need to stop in order to move JVMs and change its resource > > allocation, and stopping the task may cause assignments to change and > > the workload to be distributed elsewhere. > > > > > I think the "hints" where to place a connector/tasks should come from > the connector configuration as it's the engineers building a pipeline that > knows best the requirements (in terms of isolation, resources) of their > workload. This is basically point 3) in my initial email. The mechanism you > propose puts this burden on the cluster administrators who may well not > know the workloads and also have to guess/know in advance to properly > configure workers. > > > > In the Strimzi proposal I imagined that the resource limits would be > > chosen by the users creating the connector configurations, and the > > fact that they were beside the connector configuration rather than > > inside of it didn't change the ownership. I completely agree that the > > users defining configurations should also be responsible for setting > > resource limits. > > > > > I've not looked into the feasibility but I wonder if a simplified > taint/selector approach could give us enough flexibility to make Connect > behave better in containerized environments. > > > > Yes, a taint/selector system would be more helpful for Connect > > clusters without an external management layer, but significantly more > > complex. Do you have a strategy for specifying unique selectors for an > > arbitrary size set of tasks? > > A connector config wishes to give a different selector to each task, > > like `task-id=connector-id-N` where N is the index of the task. My > > best solution was a special $TASK_ID placeholder that would be filled > > in with the task number at runtime, but that felt like an inelegant > > carve-out. > > > > Thanks for your thoughts. > > Greg > > > > On Fri, Oct 20, 2023 at 6:25 AM Hector Geraldino (BLOOMBERG/ 919 3RD > > A) <hgerald...@bloomberg.net> wrote: > > > > > > Hi, > > > > > > I think different algorithms might work for different > workload/scenarios. I have some thoughts that are somewhat tangential to > this KIP: it might be a good idea to elevate the ConnectAssignor to the > category of plugin, so users can provide their own implementation. > > > > > > The fact that there's a public o.a.k.c.r.distributed.ConnectAssignor > interface is brilliant (I actually wanted the same thing on the Kafka > client side, alas > https://cwiki.apache.org/confluence/display/KAFKA/KIP-795%3A+Add+public+APIs+for+AbstractCoordinator). > I think it should play well with the future Connect's counterpart of > KIP-848 (new consumer rebalance protocol). > > > > > > I don't want to hijack this thread, but will definitely raise a KIP > and start a discussion around this idea. > > > > > > From: dev@kafka.apache.org At: 10/20/23 07:21:11 UTC-4:00To: > dev@kafka.apache.org > > > Subject: Re: [DISCUSS] KIP-987: Connect Static Assignments > > > > > > Hi Greg, > > > > > > Thanks for the reply. > > > > > > I still find the proposed mechanism limited and I'm not sure it really > > > addressed the pain points I've experienced with Connect. > > > As you said different tasks from a connector may have different > > > workload. Connectors may also change the assignment of tasks at > > > runtime so for example it task-2 is really busy (because it's assigned > > > a partition with high throughput), this may not be true in 10 minutes > > > as this partition is now assigned to task-1. So having to put which > > > tasks can run on each worker does not really help in this case. > > > > > > I think the "hints" where to place a connector/tasks should come from > > > the connector configuration as it's the engineers building a pipeline > > > that knows best the requirements (in terms of isolation, resources) of > > > their workload. This is basically point 3) in my initial email. The > > > mechanism you propose puts this burden on the cluster administrators > > > who may well not know the workloads and also have to guess/know in > > > advance to properly configure workers. > > > > > > I've not looked into the feasibility but I wonder if a simplified > > > taint/selector approach could give us enough flexibility to make > > > Connect behave better in containerized environments. I understand it's > > > an alternative you rejected but I think could have some benefits. Here > > > is my green field thinking: > > > Add 2 new fields in the connector config: placement and tags. > > > Placement defines the degree of isolation a task requires, it accept 3 > > > values: any (can be placed anywhere like today, the default), > > > colocated (can run on a worker with other tasks from this connector), > > > isolated (requires a dedicated worker). I think these degrees of > > > isolation should cover most use cases. Tags accepts a collections of > > > key=value pair. These can have arbitrary values and are meant to mean > > > something to the management system (for example Strimzi). The accepted > > > values could be configured on the workers by the administrators as > > > they also operate the management system. > > > > > > When a connector is created, the runtime tries to place tasks on the > > > available brokers by matching the placement and tags. If no suitable > > > workers are found, the tasks stay in unassigned state and the runtime > > > waits for the management system to create the necessary workers. > > > > > > We could even envisage to start with only the placement field as in my > > > opinion this is what brings the most value to users. > > > > > > Thanks, > > > Mickael > > > > > > On Wed, Oct 18, 2023 at 8:12 PM Greg Harris > > > <greg.har...@aiven.io.invalid> wrote: > > > > > > > > Hey Sagar, > > > > > > > > Thanks for the questions. I hope you find the answers satisfying: > > > > > > > > 1. This is detailed in the KIP two sentences earlier: "If the > > > > connect.protocol is set to static, each worker will send it's > > > > static.connectors and static.tasks to the coordinator during > > > > rebalances." > > > > > > > > 2. If you have a static worker and a wildcard worker, the static > > > > worker will be assigned the work preferentially. If the static worker > > > > goes offline, the wildcard worker will be used as a backup. > > > > > > > > 3. I don't think that new Connect users will make use of this > feature, > > > > but I've added that clarification. > > > > > > > > 4. Users can implement the strategy you're describing by leaving the > > > > static.connectors field unset. I think that Connect should include > > > > static.connectors for users that do want to control the placement of > > > > connectors. > > > > > > > > 5. Yes. Arbitrary here just means that the assignment is not > > > > influenced by the static assignment. > > > > > > > > 6. Yes. There are no guardrails that ensure that the balance of the > > > > static assignments is better than the builtin algorithm because we > > > > have no method to compare them. > > > > > > > > 7. If the whole cluster uses static assignments with each job only > > > > specified on one worker, the assignments are completely sticky. If a > > > > worker goes offline, those tasks will be offline until that worker > > > > comes back. > > > > If there are multiple workers for a single job, that is specified as > > > > "arbitrary". We could choose to wait for the delay to elapse or > > > > immediately reassign it, the KIP as written could be implemented by > > > > either. > > > > If the assignment would land on a wildcard worker, that should use > > > > cooperative rules, so we would need to respect the rebalance delay. > > > > > > > > Thanks! > > > > Greg > > > > > > > > On Wed, Oct 18, 2023 at 6:21 AM Sagar <sagarmeansoc...@gmail.com> > wrote: > > > > > > > > > > Hi Greg, > > > > > > > > > > Thanks for the KIP. I have a few questions/comments: > > > > > > > > > > 1) You mentioned that during a rebalance if all the members of the > cluster > > > > > support the static protocol, then it would use the steps outlined > in the > > > > > Proposed Section to do the assignments. In those steps, the leader > > > > > identifies the static/wildcard jobs. It is not clear to me how the > leader > > > > > makes that distinction? Are we going to enhance the embedded > protocol to > > > > > also write the static jobs that the worker owns as part of it's > assignment? > > > > > As of today, the workers just write only the owned/revoked > connectors/tasks > > > > > in case of incremental and above and only owned connectors/tasks > in case of > > > > > eager. > > > > > > > > > > 2) Could you also elaborate this statement a bit: > > > > > > > > > > > A cluster with both static and wildcard workers can use wildcard > workers > > > > > > as backups for disaster recovery. > > > > > > > > > > > > > > > I see in the Strimzi proposal you have explained this scenario > (the case > > > > > where shared workers are empty as pods aren't created yet etc > IIUC), but > > > > > reading the KIP doesn't make it too obvious. > > > > > > > > > > 3) A nit comment but we should probably call out that there is no > need to > > > > > assign the connector and task to the same worker when configuring > them. I > > > > > guess for new users of Connect this could be a source of > confusion. WDYT? > > > > > > > > > > 4) Staying on the above, I also wanted to know why do we need to > set > > > > > static.connectors? As such, connectors generally aren't resource > intensive. > > > > > Can we not let connectors be assigned using the same algorithm as > today and > > > > > let only tasks be pinned to workers? > > > > > > > > > > 5) In the proposed section you also mentioned that > > > > > > > > > > If static assignments are not specified, or at least one worker in > the > > > > > > cluster is not using the static protocol, they are ignored and > the > > > > > > worker may receive an arbitrary assignment. > > > > > > > > > > > > > > > The arbitrary assignment is driven by the minimum supported > protocol right > > > > > (sessioned/eager)? > > > > > > > > > > 6) Just for my understanding, because now the onus is on the > connect admin > > > > > to specify which connector/task should be running where, can there > be a > > > > > situation where there could be an imbalance in terms of number of > > > > > connectors/tasks running on workers (due to some oversight on part > of admin > > > > > or some bug in the tooling used to generate worker configs if any) > ? There > > > > > could be an imbalance even today as well but the protocol tries to > balance > > > > > as much as it can in an automated fashion. > > > > > > > > > > 7) Lastly, in a cluster where all workers support the static > protocol, > > > > > there is still no guarantee that all connectors/tasks would be > static. In > > > > > such a case, what happens when one of the workers is restarted or > is being > > > > > rolled? Would the assignment eventually be sticky when the worker > comes > > > > > back? Does this new protocol also abide by the scheduled rebalance > delays? > > > > > Maybe this has been explained in the KIP but it wasn't clear to me > when I > > > > > read it. > > > > > > > > > > Thanks! > > > > > Sagar. > > > > > > > > > > On Tue, Oct 10, 2023 at 11:22 PM Greg Harris > <greg.har...@aiven.io.invalid> > > > > > wrote: > > > > > > > > > > > Hi Mickael, > > > > > > > > > > > > I'm not Chris but I hope I can still respond to your questions :) > > > > > > > > > > > > 1a. This system behaves best when connectors and tasks are known > in > > > > > > advance, but I don't think that is a requirement. I clarified > that the > > > > > > worker configurations are permissive of non-existent jobs, which > > > > > > allows for bootstrapping an empty cluster and concurrent worker > & job > > > > > > provisioning. > > > > > > 1b. I think the wildcard suggestion is similar to the rejected > > > > > > alternative "implement a more complex worker selector...". While > we > > > > > > could add that sort of functionality, I think it is more > difficult to > > > > > > express and reason about. For example, with the current > configuration, > > > > > > users can ensure that at most one job is assigned to a JVM. If a > user > > > > > > wanted to use wildcards with the same constraint, those wildcards > > > > > > would need to express "at most one of <matching job>". > > > > > > 1c. The javadoc doesn't say what happens to the additional > tasks, but > > > > > > I think we could start enforcing it if it makes the > > > > > > fixed-static-assignment use-case more reliable. I have only ever > seen > > > > > > additional tasks emitted due to rather serious bugs in connector > > > > > > implementations. > > > > > > > > > > > > 2. Yes, I've added that information to the proposal. I used the > `GET > > > > > > /connectors/{connector}` API as it doesn't list the task > configs, but > > > > > > it performs the same function. > > > > > > > > > > > > 3. Do you mean the resource limits and requests? This is the > rejected > > > > > > alternative "Model per-job resource constraints and measure > resource > > > > > > utilization within a single JVM", let me know if the reasoning > there > > > > > > is not convincing. > > > > > > > > > > > > 4. I have not explored rack awareness for Connect in detail. I > expect > > > > > > that rack awareness would require some compromise on the "least > > > > > > disruption" property of incremental cooperative rebalancing for > shared > > > > > > JVMs. We could discuss that in a future KIP. > > > > > > As for this proposal, because the management layer is > responsible for > > > > > > placing workers and placing tasks on those workers, it would > also be > > > > > > responsible for implementing rack-awareness in isolated JVMs. > > > > > > > > > > > > Thanks a lot! > > > > > > Greg > > > > > > > > > > > > On Tue, Oct 10, 2023 at 10:08 AM Greg Harris < > greg.har...@aiven.io> wrote: > > > > > > > > > > > > > > Hey Tom, > > > > > > > > > > > > > > Thanks for your comments! I appreciate your insight here and > on the > > > > > > > Strimzi proposal. > > > > > > > > > > > > > > 1. Fixed, thanks. > > > > > > > > > > > > > > 2. We haven't removed a protocol yet, so workers still > announce that > > > > > > > they can use the old protocol versions, and the group > coordinator will > > > > > > > choose the highest mutually supported protocol version. This > would be > > > > > > > true even if static was not a superset of sessioned. > > > > > > > For the static feature, having workers in the cluster that do > not > > > > > > > support static assignments will cause the common version to not > > > > > > > support static assignments, and will mean that workers which > announce > > > > > > > static.connectors and static.tasks may be assigned jobs > outside of > > > > > > > their static assignments, which is called out in the KIP. > > > > > > > If the version gap is wider, then the other protocol-controlled > > > > > > > features will also change behavior (i.e. no longer using > > > > > > > incremental-cooperative-assignments, or not including REST > > > > > > > authentication headers). There would be no difference to the > behavior > > > > > > > of this feature. > > > > > > > > > > > > > > 3. Arbitrary here is specifically meant to say that one of the > workers > > > > > > > will be chosen, but outside implementations should not rely on > the > > > > > > > method of choosing. This allows us to change/improve the > assignment > > > > > > > algorithm in the future without a KIP. The reason I've > specified it > > > > > > > this way is because I'd like to choose the "least loaded" > worker, but > > > > > > > I don't know if that is well-defined or tractable to compute. > I expect > > > > > > > that in practice we will use some heuristics to tie-break, but > won't > > > > > > > be able to ensure that the algorithm will be deterministic > and/or > > > > > > > optimal. If we find an improvement to the heuristics, I think > that we > > > > > > > should be able to implement them without a KIP. > > > > > > > > > > > > > > 4. I added a section for "Choosing static assignments" that > details > > > > > > > both the "dynamic" and "fixed" use-cases. With regards to > tasks.max: > > > > > > > The javadoc for the `taskConfigs` method does state that it > > > > > > > "produc[es] at most {@code maxTasks} configurations," so static > > > > > > > inference of the task IDs is valid. I think that statically > inferring > > > > > > > the number of tasks will always come with the trade-off that > you can > > > > > > > over-provision task workers, it seems fundamental. I think > it's a > > > > > > > trade-off that users of this feature will make for themselves, > and > > > > > > > both methods will be supported. > > > > > > > > > > > > > > 5. No I don't think that it is a safe assumption, as there are > > > > > > > certainly connectors out there which violate it. For example, > sink > > > > > > > connectors with a heterogeneous set of topics, and source > connectors > > > > > > > which distribute a dynamic heterogeneous workload internally. I > > > > > > > believe there are a few connectors with a distinct "task-0" > which > > > > > > > consumes more resources than the other tasks which would have > some > > > > > > > index-stability. Index is chosen because it is the only > distinguishing > > > > > > > feature we have without inspecting task configurations, and > have > > > > > > > control over via coordination and assignments. > > > > > > > > > > > > > > Thanks again! > > > > > > > Greg > > > > > > > > > > > > > > > > > > > > > On Tue, Oct 10, 2023 at 8:05 AM Mickael Maison > > > <mickael.mai...@gmail.com> > > > > > > wrote: > > > > > > > > > > > > > > > > Hi Chris, > > > > > > > > > > > > > > > > Thanks for the KIP! > > > > > > > > > > > > > > > > 1) With this mechanism it seems Connect administrators have > to know in > > > > > > > > advanced the names of connectors that will be running. As > far as I can > > > > > > > > tell, static.connectors and static.tasks will require > restarting the > > > > > > > > worker to change. Also you don't specify if these > configurations can > > > > > > > > accept wildcards or not. This could be useful for tasks as > it's hard > > > > > > > > to know in advance how many tasks will exist, especially as > connectors > > > > > > > > can in theory ignore tasks.max (I wonder if we should make > the runtime > > > > > > > > enforce that config). > > > > > > > > > > > > > > > > 2) The KIP does not explicitly mention a mechanism for > management > > > > > > > > systems to figure out what to do. As far as I can tell (and > looking at > > > > > > > > the proposal in Strimzi), these systems will have to poll > the REST API > > > > > > > > to do so. If I understand correctly, management systems > would first > > > > > > > > have to poll the GET /connectors and then GET > /connectors/{name}/tasks > > > > > > > > for each connector, is that right? > > > > > > > > > > > > > > > > 3) In the Strimzi proposal you mention adding custom fields > to the > > > > > > > > Connector resource to instruct it what to do. I wonder if > these should > > > > > > > > be added directly in the connector configuration, rather > than just > > > > > > > > exist in the management system. That way all the data > necessary to > > > > > > > > make decisions would be available in the same place, the > REST API. > > > > > > > > > > > > > > > > 4) This is not something you mention in the KIP and I don't > think it's > > > > > > > > a blocker for this KIP but I wonder if you thought about rack > > > > > > > > awareness too. I'm bringing it as I think it's a similar > concern when > > > > > > > > it comes to deciding where to assign tasks. > > > > > > > > > > > > > > > > Thanks, > > > > > > > > Mickael > > > > > > > > > > > > > > > > > > > > > > > > On Tue, Oct 10, 2023 at 8:39 AM Tom Bentley < > tbent...@redhat.com> > > > > > > wrote: > > > > > > > > > > > > > > > > > > Hi Greg, > > > > > > > > > > > > > > > > > > Many thanks for the KIP. Here are a few initial questions > > > > > > > > > > > > > > > > > > 1. Incomplete sentence: "But Connect is not situated to be > able to > > > > > > manage > > > > > > > > > resources directly, as workers are given a fixed " > > > > > > > > > 2. You explain how sessioned is now a subset of static, > but what > > > > > > happens in > > > > > > > > > a cluster where some workers are using static and some are > using > > > > > > either > > > > > > > > > eager or compatible? > > > > > > > > > 3. "Assign each unassigned static job to a static worker > which > > > > > > specifies > > > > > > > > > that job, choosing arbitrarily if there are multiple valid > workers." > > > > > > I > > > > > > > > > think there might be less ambiguous words than "arbitrary" > to > > > > > > specify this > > > > > > > > > behaviour. Hashing the task name would _appear_ pretty > arbitrary to > > > > > > the > > > > > > > > > user, but would be deterministic. Picking at random would > be > > > > > > > > > non-deterministic. Even better if you have a rationale. > > > > > > > > > 4. You don't describe how a user, or automated system, > starting with > > > > > > a set > > > > > > > > > of connectors, should find out the tasks that they want to > run. This > > > > > > > > > relates to the contract of > > > > > > > > > > org.apache.kafka.connect.connector.Connector#taskConfigs(int > > > > > > maxTasks). > > > > > > > > > AFAIK (please correct me if I'm wrong, because it's a long > time > > > > > > since I > > > > > > > > > looked at this code) there's nothing that validates that > the > > > > > > returned list > > > > > > > > > has at most the `maxTasks` and connectors can, of course, > return > > > > > > fewer than > > > > > > > > > that many tasks. So without deeper knowledge of a > particular > > > > > > connector it's > > > > > > > > > not clear to the user/operator how to configure their > static workers > > > > > > and > > > > > > > > > static assignments. > > > > > > > > > 5. Is there a lurking assumption that task indices are > stable? E.g. > > > > > > that > > > > > > > > > the task with index 3 will always be the > resource-intensive one. I > > > > > > can see > > > > > > > > > that that would often be a reliable assumption, but it's > not clear > > > > > > to me > > > > > > > > > that it is always the case. > > > > > > > > > > > > > > > > > > Thanks again, > > > > > > > > > > > > > > > > > > Tom > > > > > > > > > > > > > > > > > > On Fri, 6 Oct 2023 at 12:36, Greg Harris > > > > > > <greg.har...@aiven.io.invalid> > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > Hey everyone! > > > > > > > > > > > > > > > > > > > > I'd like to propose an improvement to Kafka Connect's > scheduling > > > > > > > > > > algorithm, with the goal of improving the operability of > connect > > > > > > > > > > clusters through resource isolation. > > > > > > > > > > > > > > > > > > > > Find the KIP here: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-987%3A+Connect+Static+Assi > > > gnments > > > > > > > > > > > > > > > > > > > > This feature is primarily intended to be consumed by > cluster > > > > > > > > > > management systems, so I've opened a sister proposal in > the > > > Strimzi > > > > > > > > > > project that uses this feature to provide user-facing > resource > > > > > > > > > > isolation. The base feature is generic, so it is usable > manually > > > > > > and > > > > > > > > > > with cluster management systems other than Strimzi. > > > > > > > > > > > > > > > > > > > > Find the proposal here: > > > > > > https://github.com/strimzi/proposals/pull/96 > > > > > > > > > > > > > > > > > > > > Thanks! > > > > > > > > > > Greg Harris > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
