Hi Chris, Sorry for the late reply.
> I don't believe logging an error message is sufficient for > handling failures to reset-after-delete. IMO it's highly > likely that users will either shoot themselves in the foot > by not reading the fine print and realizing that the offset > request may have failed, or will ask for better visibility > into the success or failure of the reset request than > scanning log files. Your reasoning for deferring the reset offsets after delete functionality to a separate KIP makes sense, thanks for the explanation. > I've updated the KIP with the > developer-facing API changes for this logic This is great, I hadn't considered the other two (very valid) use-cases for such an API, thanks for adding these with elaborate documentation! However, the significance / use of the boolean value returned by the two methods is not fully clear, could you please clarify? Thanks, Yash On Fri, Nov 18, 2022 at 1:06 AM Chris Egerton <chr...@aiven.io.invalid> wrote: > Hi Yash, > > I've updated the KIP with the correct "kafka_topic", "kafka_partition", and > "kafka_offset" keys in the JSON examples (settled on those instead of > prefixing with "Kafka " for better interactions with tooling like JQ). I've > also added a note about sink offset requests failing if there are still > active members in the consumer group. > > I don't believe logging an error message is sufficient for handling > failures to reset-after-delete. IMO it's highly likely that users will > either shoot themselves in the foot by not reading the fine print and > realizing that the offset request may have failed, or will ask for better > visibility into the success or failure of the reset request than scanning > log files. I don't doubt that there are ways to address this, but I would > prefer to leave them to a separate KIP since the required design work is > non-trivial and I do not feel that the added burden is worth tying to this > KIP as a blocker. > > I was really hoping to avoid introducing a change to the developer-facing > APIs with this KIP, but after giving it some thought I think this may be > unavoidable. It's debatable whether validation of altered offsets is a good > enough use case on its own for this kind of API, but since there are also > connectors out there that manage offsets externally, we should probably add > a hook to allow those external offsets to be managed, which can then serve > double- or even-triple duty as a hook to validate custom offsets and to > notify users whether offset resets/alterations are supported at all (which > they may not be if, for example, offsets are coupled tightly with the data > written by a sink connector). I've updated the KIP with the > developer-facing API changes for this logic; let me know what you think. > > Cheers, > > Chris > > On Mon, Nov 14, 2022 at 10:16 AM Mickael Maison <mickael.mai...@gmail.com> > wrote: > > > Hi Chris, > > > > Thanks for the update! > > > > It's relatively common to only want to reset offsets for a specific > > resource (for example with MirrorMaker for one or a group of topics). > > Could it be possible to add a way to do so? Either by providing a > > payload to DELETE or by setting the offset field to an empty object in > > the PATCH payload? > > > > Thanks, > > Mickael > > > > On Sat, Nov 12, 2022 at 3:33 PM Yash Mayya <yash.ma...@gmail.com> wrote: > > > > > > Hi Chris, > > > > > > Thanks for pointing out that the consumer group deletion step itself > will > > > fail in case of zombie sink tasks. Since we can't get any stronger > > > guarantees from consumers (unlike with transactional producers), I > think > > it > > > makes perfect sense to fail the offset reset attempt in such scenarios > > with > > > a relevant error message to the user. I was more concerned about > silently > > > failing but it looks like that won't be an issue. It's probably worth > > > calling out this difference between source / sink connectors explicitly > > in > > > the KIP, what do you think? > > > > > > > changing the field names for sink offsets > > > > from "topic", "partition", and "offset" to "Kafka > > > > topic", "Kafka partition", and "Kafka offset" respectively, to > > > > reduce the stuttering effect of having a "partition" field inside > > > > a "partition" field and the same with an "offset" field > > > > > > The KIP is still using the nested partition / offset fields by the way > - > > > has it not been updated because we're waiting for consensus on the > field > > > names? > > > > > > > The reset-after-delete feature, on the other > > > > hand, is actually pretty tricky to design; I've updated the > > > > rationale in the KIP for delaying it and clarified that it's not > > > > just a matter of implementation but also design work. > > > > > > I like the idea of writing an offset reset request to the config topic > > > which will be processed by the herder's config update listener - I'm > not > > > sure I fully follow the concerns with regard to handling failures? Why > > > can't we simply log an error saying that the offset reset for the > deleted > > > connector "xyz" failed due to reason "abc"? As long as it's documented > > that > > > connector deletion and offset resets are asynchronous and a success > > > response only means that the request was initiated successfully (which > is > > > the case even today with normal connector deletion), we should be fine > > > right? > > > > > > Thanks for adding the new PATCH endpoint to the KIP, I think it's a lot > > > more useful for this use case than a PUT endpoint would be! One thing > > > that I was thinking about with the new PATCH endpoint is that while we > > can > > > easily validate the request body format for sink connectors (since it's > > the > > > same across all connectors), we can't do the same for source connectors > > as > > > things stand today since each source connector implementation can > define > > > its own source partition and offset structures. Without any validation, > > > writing a bad offset for a source connector via the PATCH endpoint > could > > > cause it to fail with hard to discern errors. I'm wondering if we could > > add > > > a new method to the `SourceConnector` class (which should be overridden > > by > > > source connector implementations) that would validate whether or not > the > > > provided source partitions and source offsets are valid for the > connector > > > (it could have a default implementation returning true unconditionally > > for > > > backward compatibility). > > > > > > > I've also added an implementation plan to the KIP, which calls > > > > out the different parts that can be worked on independently so that > > > > others (hi Yash 🙂) can also tackle parts of this if they'd like. > > > > > > I'd be more than happy to pick up one or more of the implementation > > parts, > > > thanks for breaking it up into granular pieces! > > > > > > Thanks, > > > Yash > > > > > > On Fri, Nov 11, 2022 at 11:25 PM Chris Egerton <chr...@aiven.io.invalid > > > > > wrote: > > > > > > > Hi Mickael, > > > > > > > > Thanks for your feedback. This has been on my TODO list as well :) > > > > > > > > 1. That's fair! Support for altering offsets is easy enough to > design, > > so > > > > I've added it to the KIP. The reset-after-delete feature, on the > other > > > > hand, is actually pretty tricky to design; I've updated the rationale > > in > > > > the KIP for delaying it and clarified that it's not just a matter of > > > > implementation but also design work. If you or anyone else can think > > of a > > > > clean, simple way to implement it, I'm happy to add it to this KIP, > but > > > > otherwise I'd prefer not to tie it to the approval and release of the > > > > features already proposed in the KIP. > > > > > > > > 2. Yeah, it's a little awkward. In my head I've justified the > ugliness > > of > > > > the implementation with the smooth user-facing experience; falling > back > > > > seamlessly on the PAUSED state without even logging an error message > > is a > > > > lot better than I'd initially hoped for when I was designing this > > feature. > > > > > > > > I've also added an implementation plan to the KIP, which calls out > the > > > > different parts that can be worked on independently so that others > (hi > > Yash > > > > 🙂) can also tackle parts of this if they'd like. > > > > > > > > Finally, I've removed the "type" field from the response body format > > for > > > > offset read requests. This way, users can copy+paste the response > from > > that > > > > endpoint into a request to alter a connector's offsets without having > > to > > > > remove the "type" field first. An alternative was to keep the "type" > > field > > > > and add it to the request body format for altering offsets, but this > > didn't > > > > seem to make enough sense for cases not involving the aforementioned > > > > copy+paste process. > > > > > > > > Cheers, > > > > > > > > Chris > > > > > > > > On Wed, Nov 9, 2022 at 9:57 AM Mickael Maison < > > mickael.mai...@gmail.com> > > > > wrote: > > > > > > > > > Hi Chris, > > > > > > > > > > Thanks for the KIP, you're picking something that has been in my > todo > > > > > list for a while ;) > > > > > > > > > > It looks good overall, I just have a couple of questions: > > > > > 1) I consider both features listed in Future Work pretty important. > > In > > > > > both cases you mention the reason for not addressing them now is > > > > > because of the implementation. If the design is simple and if we > have > > > > > volunteers to implement them, I wonder if we could include them in > > > > > this KIP. So you would not have to implement everything but we > would > > > > > have a single KIP and vote. > > > > > > > > > > 2) Regarding the backward compatibility for the stopped state. The > > > > > "state.v2" field is a bit unfortunate but I can't think of a better > > > > > solution. The other alternative would be to not do anything but I > > > > > think the graceful degradation you propose is a bit better. > > > > > > > > > > Thanks, > > > > > Mickael > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Tue, Nov 8, 2022 at 5:58 PM Chris Egerton > <chr...@aiven.io.invalid > > > > > > > > wrote: > > > > > > > > > > > > Hi Yash, > > > > > > > > > > > > Good question! This is actually a subtle source of asymmetry in > the > > > > > current > > > > > > proposal. Requests to delete a consumer group with active members > > will > > > > > > fail, so if there are zombie sink tasks that are still > > communicating > > > > with > > > > > > Kafka, offset reset requests for that connector will also fail. > It > > is > > > > > > possible to use an admin client to remove all active members from > > the > > > > > group > > > > > > and then delete the group. However, this solution isn't as > > complete as > > > > > the > > > > > > zombie fencing that we can perform for exactly-once source tasks, > > since > > > > > > removing consumers from a group doesn't prevent them from > > immediately > > > > > > rejoining the group, which would either cause the group deletion > > > > request > > > > > to > > > > > > fail (if they rejoin before the group is deleted), or recreate > the > > > > group > > > > > > (if they rejoin after the group is deleted). > > > > > > > > > > > > For ease of implementation, I'd prefer to leave the asymmetry in > > the > > > > API > > > > > > for now and fail fast and clearly if there are still consumers > > active > > > > in > > > > > > the sink connector's group. We can try to detect this case and > > provide > > > > a > > > > > > helpful error message to the user explaining why the offset reset > > > > request > > > > > > has failed and some steps they can take to try to resolve things > > (wait > > > > > for > > > > > > slow task shutdown to complete, restart zombie workers and/or > > workers > > > > > with > > > > > > blocked tasks on them). In the future we can possibly even > revisit > > > > > KIP-611 > > > > > > [1] or something like it to provide better insight into zombie > > tasks > > > > on a > > > > > > worker so that it's easier to find which tasks have been > abandoned > > but > > > > > are > > > > > > still running. > > > > > > > > > > > > Let me know what you think; this is an important point to call > out > > and > > > > if > > > > > > we can reach some consensus on how to handle sink connector > offset > > > > resets > > > > > > w/r/t zombie tasks, I'll update the KIP with the details. > > > > > > > > > > > > [1] - > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-611%3A+Improved+Handling+of+Abandoned+Connectors+and+Tasks > > > > > > > > > > > > Cheers, > > > > > > > > > > > > Chris > > > > > > > > > > > > On Tue, Nov 8, 2022 at 8:00 AM Yash Mayya <yash.ma...@gmail.com> > > > > wrote: > > > > > > > > > > > > > Hi Chris, > > > > > > > > > > > > > > Thanks for the response and the explanations, I think you've > > answered > > > > > > > pretty much all the questions I had meticulously! > > > > > > > > > > > > > > > > > > > > > > if something goes wrong while resetting offsets, there's no > > > > > > > > immediate impact--the connector will still be in the STOPPED > > > > > > > > state. The REST response for requests to reset the offsets > > > > > > > > will clearly call out that the operation has failed, and if > > > > > necessary, > > > > > > > > we can probably also add a scary-looking warning message > > > > > > > > stating that we can't guarantee which offsets have been > > > > successfully > > > > > > > > wiped and which haven't. Users can query the exact offsets > of > > > > > > > > the connector at this point to determine what will happen > > if/what > > > > > they > > > > > > > > resume it. And they can repeat attempts to reset the offsets > as > > > > many > > > > > > > > times as they'd like until they get back a 2XX response, > > > > indicating > > > > > > > > that it's finally safe to resume the connector. Thoughts? > > > > > > > > > > > > > > Yeah, I agree, the case that I mentioned earlier where a user > > would > > > > > try to > > > > > > > resume a stopped connector after a failed offset reset attempt > > > > without > > > > > > > knowing that the offset reset attempt didn't fail cleanly is > > probably > > > > > just > > > > > > > an extreme edge case. I think as long as the response is > verbose > > > > > enough and > > > > > > > self explanatory, we should be fine. > > > > > > > > > > > > > > Another question that I had was behavior w.r.t sink connector > > offset > > > > > resets > > > > > > > when there are zombie tasks/workers in the Connect cluster - > the > > KIP > > > > > > > mentions that for sink connectors offset resets will be done by > > > > > deleting > > > > > > > the consumer group. However, if there are zombie tasks which > are > > > > still > > > > > able > > > > > > > to communicate with the Kafka cluster that the sink connector > is > > > > > consuming > > > > > > > from, I think the consumer group will automatically get > > re-created > > > > and > > > > > the > > > > > > > zombie task may be able to commit offsets for the partitions > > that it > > > > is > > > > > > > consuming from? > > > > > > > > > > > > > > Thanks, > > > > > > > Yash > > > > > > > > > > > > > > > > > > > > > On Fri, Nov 4, 2022 at 10:27 PM Chris Egerton > > > > <chr...@aiven.io.invalid > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > Hi Yash, > > > > > > > > > > > > > > > > Thanks again for your thoughts! Responses to ongoing > > discussions > > > > > inline > > > > > > > > (easier to track context than referencing comment numbers): > > > > > > > > > > > > > > > > > However, this then leads me to wonder if we can make that > > > > explicit > > > > > by > > > > > > > > including "connect" or "connector" in the higher level field > > names? > > > > > Or do > > > > > > > > you think this isn't required given that we're talking about > a > > > > > Connect > > > > > > > > specific REST API in the first place? > > > > > > > > > > > > > > > > I think "partition" and "offset" are fine as field names but > > I'm > > > > not > > > > > > > hugely > > > > > > > > opposed to adding "connector " as a prefix to them; would be > > > > > interested > > > > > > > in > > > > > > > > others' thoughts. > > > > > > > > > > > > > > > > > I'm not sure I followed why the unresolved writes to the > > config > > > > > topic > > > > > > > > would be an issue - wouldn't the delete offsets request be > > added to > > > > > the > > > > > > > > herder's request queue and whenever it is processed, we'd > > anyway > > > > > need to > > > > > > > > check if all the prerequisites for the request are satisfied? > > > > > > > > > > > > > > > > Some requests are handled in multiple steps. For example, > > deleting > > > > a > > > > > > > > connector (1) adds a request to the herder queue to write a > > > > > tombstone to > > > > > > > > the config topic (or, if the worker isn't the leader, forward > > the > > > > > request > > > > > > > > to the leader). (2) Once that tombstone is picked up, (3) a > > > > rebalance > > > > > > > > ensues, and then after it's finally complete, (4) the > > connector and > > > > > its > > > > > > > > tasks are shut down. I probably could have used better > > terminology, > > > > > but > > > > > > > > what I meant by "unresolved writes to the config topic" was a > > case > > > > in > > > > > > > > between steps (2) and (3)--where the worker has already read > > that > > > > > > > tombstone > > > > > > > > from the config topic and knows that a rebalance is pending, > > but > > > > > hasn't > > > > > > > > begun participating in that rebalance yet. In the > > DistributedHerder > > > > > > > class, > > > > > > > > this is done via the `checkRebalanceNeeded` method. > > > > > > > > > > > > > > > > > We can probably revisit this potential deprecation [of the > > PAUSED > > > > > > > state] > > > > > > > > in the future based on user feedback and how the adoption of > > the > > > > new > > > > > > > > proposed stop endpoint looks like, what do you think? > > > > > > > > > > > > > > > > Yeah, revisiting in the future seems reasonable. 👍 > > > > > > > > > > > > > > > > And responses to new comments here: > > > > > > > > > > > > > > > > 8. Yep, we'll start tracking offsets by connector. I don't > > believe > > > > > this > > > > > > > > should be too difficult, and suspect that the only reason we > > track > > > > > raw > > > > > > > byte > > > > > > > > arrays instead of pre-deserializing offset topic information > > into > > > > > > > something > > > > > > > > more useful is because Connect originally had pluggable > > internal > > > > > > > > converters. Now that we're hardcoded to use the JSON > converter > > it > > > > > should > > > > > > > be > > > > > > > > fine to track offsets on a per-connector basis as they're > read > > from > > > > > the > > > > > > > > offsets topic. > > > > > > > > > > > > > > > > 9. I'm hesitant to introduce this type of feature right now > > because > > > > > of > > > > > > > all > > > > > > > > of the gotchas that would come with it. In security-conscious > > > > > > > environments, > > > > > > > > it's possible that a sink connector's principal may have > > access to > > > > > the > > > > > > > > consumer group used by the connector, but the worker's > > principal > > > > may > > > > > not. > > > > > > > > There's also the case where source connectors have separate > > offsets > > > > > > > topics, > > > > > > > > or sink connectors have overridden consumer group IDs, or > sink > > or > > > > > source > > > > > > > > connectors work against a different Kafka cluster than the > one > > that > > > > > their > > > > > > > > worker uses. Overall, I'd rather provide a single API that > > works in > > > > > all > > > > > > > > cases rather than risk confusing and alienating users by > > trying to > > > > > make > > > > > > > > their lives easier in a subset of cases. > > > > > > > > > > > > > > > > 10. Hmm... I don't think the order of the writes matters too > > much > > > > > here, > > > > > > > but > > > > > > > > we probably could start by deleting from the global topic > > first, > > > > > that's > > > > > > > > true. The reason I'm not hugely concerned about this case is > > that > > > > if > > > > > > > > something goes wrong while resetting offsets, there's no > > immediate > > > > > > > > impact--the connector will still be in the STOPPED state. The > > REST > > > > > > > response > > > > > > > > for requests to reset the offsets will clearly call out that > > the > > > > > > > operation > > > > > > > > has failed, and if necessary, we can probably also add a > > > > > scary-looking > > > > > > > > warning message stating that we can't guarantee which offsets > > have > > > > > been > > > > > > > > successfully wiped and which haven't. Users can query the > exact > > > > > offsets > > > > > > > of > > > > > > > > the connector at this point to determine what will happen > > if/what > > > > > they > > > > > > > > resume it. And they can repeat attempts to reset the offsets > as > > > > many > > > > > > > times > > > > > > > > as they'd like until they get back a 2XX response, indicating > > that > > > > > it's > > > > > > > > finally safe to resume the connector. Thoughts? > > > > > > > > > > > > > > > > 11. I haven't thought too much about it. I think something > > like the > > > > > > > > Monitorable* connectors would probably serve our needs here; > > we can > > > > > > > > instantiate them on a running Connect cluster and then use > > various > > > > > > > handles > > > > > > > > to know how many times they've been polled, committed > records, > > etc. > > > > > If > > > > > > > > necessary we can tweak those classes or even write our own. > But > > > > > anyways, > > > > > > > > once that's all done, the test will be something like > "create a > > > > > > > connector, > > > > > > > > wait for it to produce N records (each of which contains some > > kind > > > > of > > > > > > > > predictable offset), and ensure that the offsets for it in > the > > REST > > > > > API > > > > > > > > match up with the ones we'd expect from N records". Does that > > > > answer > > > > > your > > > > > > > > question? > > > > > > > > > > > > > > > > Cheers, > > > > > > > > > > > > > > > > Chris > > > > > > > > > > > > > > > > On Tue, Oct 18, 2022 at 3:28 AM Yash Mayya < > > yash.ma...@gmail.com> > > > > > wrote: > > > > > > > > > > > > > > > > > Hi Chris, > > > > > > > > > > > > > > > > > > 1. Thanks a lot for elaborating on this, I'm now convinced > > about > > > > > the > > > > > > > > > usefulness of the new offset reset endpoint. Regarding the > > > > > follow-up > > > > > > > KIP > > > > > > > > > for a fine-grained offset write API, I'd be happy to take > > that on > > > > > once > > > > > > > > this > > > > > > > > > KIP is finalized and I will definitely look forward to your > > > > > feedback on > > > > > > > > > that one! > > > > > > > > > > > > > > > > > > 2. Gotcha, the motivation makes more sense to me now. So > the > > > > higher > > > > > > > level > > > > > > > > > partition field represents a Connect specific "logical > > partition" > > > > > of > > > > > > > > sorts > > > > > > > > > - i.e. the source partition as defined by a connector for > > source > > > > > > > > connectors > > > > > > > > > and a Kafka topic + partition for sink connectors. I like > the > > > > idea > > > > > of > > > > > > > > > adding a Kafka prefix to the lower level partition/offset > > (and > > > > > topic) > > > > > > > > > fields which basically makes it more clear (although > > implicitly) > > > > > that > > > > > > > the > > > > > > > > > higher level partition/offset field is Connect specific and > > not > > > > the > > > > > > > same > > > > > > > > as > > > > > > > > > what those terms represent in Kafka itself. However, this > > then > > > > > leads me > > > > > > > > to > > > > > > > > > wonder if we can make that explicit by including "connect" > or > > > > > > > "connector" > > > > > > > > > in the higher level field names? Or do you think this isn't > > > > > required > > > > > > > > given > > > > > > > > > that we're talking about a Connect specific REST API in the > > first > > > > > > > place? > > > > > > > > > > > > > > > > > > 3. Thanks, I think the response structure definitely looks > > better > > > > > now! > > > > > > > > > > > > > > > > > > 4. Interesting, I'd be curious to learn why we might want > to > > > > change > > > > > > > this > > > > > > > > in > > > > > > > > > the future but that's probably out of scope for this > > discussion. > > > > > I'm > > > > > > > not > > > > > > > > > sure I followed why the unresolved writes to the config > topic > > > > > would be > > > > > > > an > > > > > > > > > issue - wouldn't the delete offsets request be added to the > > > > > herder's > > > > > > > > > request queue and whenever it is processed, we'd anyway > need > > to > > > > > check > > > > > > > if > > > > > > > > > all the prerequisites for the request are satisfied? > > > > > > > > > > > > > > > > > > 5. Thanks for elaborating that just fencing out the > producer > > > > still > > > > > > > leaves > > > > > > > > > many cases where source tasks remain hanging around and > also > > that > > > > > we > > > > > > > > anyway > > > > > > > > > can't have similar data production guarantees for sink > > connectors > > > > > right > > > > > > > > > now. I agree that it might be better to go with ease of > > > > > implementation > > > > > > > > and > > > > > > > > > consistency for now. > > > > > > > > > > > > > > > > > > 6. Right, that does make sense but I still feel like the > two > > > > states > > > > > > > will > > > > > > > > > end up being confusing to end users who might not be able > to > > > > > discern > > > > > > > the > > > > > > > > > (fairly low-level) differences between them (also the > > nuances of > > > > > state > > > > > > > > > transitions like STOPPED -> PAUSED or PAUSED -> STOPPED > with > > the > > > > > > > > > rebalancing implications as well). We can probably revisit > > this > > > > > > > potential > > > > > > > > > deprecation in the future based on user feedback and how > the > > > > > adoption > > > > > > > of > > > > > > > > > the new proposed stop endpoint looks like, what do you > think? > > > > > > > > > > > > > > > > > > 7. Aha, that is completely my bad, I missed that the v1/v2 > > state > > > > is > > > > > > > only > > > > > > > > > applicable to the connector's target state and that we > don't > > need > > > > > to > > > > > > > > worry > > > > > > > > > about the tasks since we will have an empty set of tasks. I > > > > think I > > > > > > > was a > > > > > > > > > little confused by "pause the parts of the connector that > > they > > > > are > > > > > > > > > assigned" from the KIP. Thanks for clarifying that! > > > > > > > > > > > > > > > > > > > > > > > > > > > Some more thoughts and questions that I had - > > > > > > > > > > > > > > > > > > 8. Could you elaborate on what the implementation for > offset > > > > reset > > > > > for > > > > > > > > > source connectors would look like? Currently, it doesn't > look > > > > like > > > > > we > > > > > > > > track > > > > > > > > > all the partitions for a source connector anywhere. Will we > > need > > > > to > > > > > > > > > book-keep this somewhere in order to be able to emit a > > tombstone > > > > > record > > > > > > > > for > > > > > > > > > each source partition? > > > > > > > > > > > > > > > > > > 9. The KIP describes the offset reset endpoint as only > being > > > > > usable on > > > > > > > > > existing connectors that are in a `STOPPED` state. Why > > wouldn't > > > > we > > > > > want > > > > > > > > to > > > > > > > > > allow resetting offsets for a deleted connector which seems > > to > > > > be a > > > > > > > valid > > > > > > > > > use case? Or do we plan to handle this use case only via > the > > item > > > > > > > > outlined > > > > > > > > > in the future work section - "Automatically delete offsets > > with > > > > > > > > > connectors"? > > > > > > > > > > > > > > > > > > 10. The KIP mentions that source offsets will be reset > > > > > transactionally > > > > > > > > for > > > > > > > > > each topic (worker global offset topic and connector > specific > > > > > offset > > > > > > > > topic > > > > > > > > > if it exists). While it obviously isn't possible to > > atomically do > > > > > the > > > > > > > > > writes to two topics which may be on different Kafka > > clusters, > > > > I'm > > > > > > > > > wondering about what would happen if the first transaction > > > > > succeeds but > > > > > > > > the > > > > > > > > > second one fails. I think the order of the two transactions > > > > matters > > > > > > > here > > > > > > > > - > > > > > > > > > if we successfully emit tombstones to the connector > specific > > > > offset > > > > > > > topic > > > > > > > > > and fail to do so for the worker global offset topic, we'll > > > > > presumably > > > > > > > > fail > > > > > > > > > the offset delete request because the KIP mentions that "A > > > > request > > > > > to > > > > > > > > reset > > > > > > > > > offsets for a source connector will only be considered > > successful > > > > > if > > > > > > > the > > > > > > > > > worker is able to delete all known offsets for that > > connector, on > > > > > both > > > > > > > > the > > > > > > > > > worker's global offsets topic and (if one is used) the > > > > connector's > > > > > > > > > dedicated offsets topic.". However, this will lead to the > > > > connector > > > > > > > only > > > > > > > > > being able to read potentially older offsets from the > worker > > > > global > > > > > > > > offset > > > > > > > > > topic on resumption (based on the combined offset view > > presented > > > > as > > > > > > > > > described in KIP-618 [1]). So, I think we should make sure > > that > > > > the > > > > > > > > worker > > > > > > > > > global offset topic tombstoning is attempted first, right? > > Note > > > > > that in > > > > > > > > the > > > > > > > > > current implementation of > > `ConnectorOffsetBackingStore::set`, the > > > > > > > > primary / > > > > > > > > > connector specific offset store is written to first. > > > > > > > > > > > > > > > > > > 11. This probably isn't necessary to elaborate on in the > KIP > > > > > itself, > > > > > > > but > > > > > > > > I > > > > > > > > > was wondering what the second offset test - "verify that > that > > > > those > > > > > > > > offsets > > > > > > > > > reflect an expected level of progress for each connector > > (i.e., > > > > > they > > > > > > > are > > > > > > > > > greater than or equal to a certain value depending on how > the > > > > > > > connectors > > > > > > > > > are configured and how long they have been running)" - > would > > look > > > > > like? > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > Yash > > > > > > > > > > > > > > > > > > [1] - > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=153817402#KIP618:ExactlyOnceSupportforSourceConnectors-Smoothmigration > > > > > > > > > > > > > > > > > > On Tue, Oct 18, 2022 at 12:42 AM Chris Egerton > > > > > <chr...@aiven.io.invalid > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > Hi Yash, > > > > > > > > > > > > > > > > > > > > Thanks for your detailed thoughts. > > > > > > > > > > > > > > > > > > > > 1. In KAFKA-4107 [1], the primary request is exactly > what's > > > > > proposed > > > > > > > in > > > > > > > > > the > > > > > > > > > > KIP right now: a way to reset offsets for connectors. > Sure, > > > > > there's > > > > > > > an > > > > > > > > > > extra step of stopping the connector, but renaming a > > connector > > > > > isn't > > > > > > > as > > > > > > > > > > convenient of an alternative as it may seem since in many > > cases > > > > > you'd > > > > > > > > > also > > > > > > > > > > want to delete the older one, so the complete sequence of > > steps > > > > > would > > > > > > > > be > > > > > > > > > > something like delete the old connector, rename it > > (possibly > > > > > > > requiring > > > > > > > > > > modifications to its config file, depending on which API > is > > > > > used), > > > > > > > then > > > > > > > > > > create the renamed variant. It's also just not a great > user > > > > > > > > > > experience--even if the practical impacts are limited > > (which, > > > > > IMO, > > > > > > > they > > > > > > > > > are > > > > > > > > > > not), people have been asking for years about why they > > have to > > > > > employ > > > > > > > > > this > > > > > > > > > > kind of a workaround for a fairly common use case, and we > > don't > > > > > > > really > > > > > > > > > have > > > > > > > > > > a good answer beyond "we haven't implemented something > > better > > > > > yet". > > > > > > > On > > > > > > > > > top > > > > > > > > > > of that, you may have external tooling that needs to be > > tweaked > > > > > to > > > > > > > > > handle a > > > > > > > > > > new connector name, you may have strict authorization > > policies > > > > > around > > > > > > > > who > > > > > > > > > > can access what connectors, you may have other ACLs > > attached to > > > > > the > > > > > > > > name > > > > > > > > > of > > > > > > > > > > the connector (which can be especially common in the case > > of > > > > sink > > > > > > > > > > connectors, whose consumer group IDs are tied to their > > names by > > > > > > > > default), > > > > > > > > > > and leaving around state in the offsets topic that can > > never be > > > > > > > cleaned > > > > > > > > > up > > > > > > > > > > presents a bit of a footgun for users. It may not be a > > silver > > > > > bullet, > > > > > > > > but > > > > > > > > > > providing some mechanism to reset that state is a step in > > the > > > > > right > > > > > > > > > > direction and allows responsible users to more carefully > > > > > administer > > > > > > > > their > > > > > > > > > > cluster without resorting to non-public APIs. That said, > I > > do > > > > > agree > > > > > > > > that > > > > > > > > > a > > > > > > > > > > fine-grained reset/overwrite API would be useful, and I'd > > be > > > > > happy to > > > > > > > > > > review a KIP to add that feature if anyone wants to > tackle > > it! > > > > > > > > > > > > > > > > > > > > 2. Keeping the two formats symmetrical is motivated > mostly > > by > > > > > > > > aesthetics > > > > > > > > > > and quality-of-life for programmatic interaction with the > > API; > > > > > it's > > > > > > > not > > > > > > > > > > really a goal to hide the use of consumer groups from > > users. I > > > > do > > > > > > > agree > > > > > > > > > > that the format is a little strange-looking for sink > > > > connectors, > > > > > but > > > > > > > it > > > > > > > > > > seemed like it would be easier to work with for UIs, > > casual jq > > > > > > > queries, > > > > > > > > > and > > > > > > > > > > CLIs than a more Kafka-specific alternative such as > > > > > > > > > {"<topic>-<partition>": > > > > > > > > > > "<offset>"}, and although it is a little strange, I don't > > think > > > > > it's > > > > > > > > any > > > > > > > > > > less readable or intuitive. That said, I've made some > > tweaks to > > > > > the > > > > > > > > > format > > > > > > > > > > that should make programmatic access even easier; > > specifically, > > > > > I've > > > > > > > > > > removed the "source" and "sink" wrapper fields and > instead > > > > moved > > > > > them > > > > > > > > > into > > > > > > > > > > a top-level object with a "type" and "offsets" field, > just > > like > > > > > you > > > > > > > > > > suggested in point 3 (thanks!). We might also consider > > changing > > > > > the > > > > > > > > field > > > > > > > > > > names for sink offsets from "topic", "partition", and > > "offset" > > > > to > > > > > > > > "Kafka > > > > > > > > > > topic", "Kafka partition", and "Kafka offset" > > respectively, to > > > > > reduce > > > > > > > > the > > > > > > > > > > stuttering effect of having a "partition" field inside a > > > > > "partition" > > > > > > > > > field > > > > > > > > > > and the same with an "offset" field; thoughts? One final > > > > > point--by > > > > > > > > > equating > > > > > > > > > > source and sink offsets, we probably make it easier for > > users > > > > to > > > > > > > > > understand > > > > > > > > > > exactly what a source offset is; anyone who's familiar > with > > > > > consumer > > > > > > > > > > offsets can see from the response format that we > identify a > > > > > logical > > > > > > > > > > partition as a combination of two entities (a topic and a > > > > > partition > > > > > > > > > > number); it should make it easier to grok what a source > > offset > > > > > is by > > > > > > > > > seeing > > > > > > > > > > what the two formats have in common. > > > > > > > > > > > > > > > > > > > > 3. Great idea! Done. > > > > > > > > > > > > > > > > > > > > 4. Yes, I'm thinking right now that a 409 will be the > > response > > > > > status > > > > > > > > if > > > > > > > > > a > > > > > > > > > > rebalance is pending. I'd rather not add this to the KIP > > as we > > > > > may > > > > > > > want > > > > > > > > > to > > > > > > > > > > change it at some point and it doesn't seem vital to > > establish > > > > > it as > > > > > > > > part > > > > > > > > > > of the public contract for the new endpoint right now. > > Also, > > > > > small > > > > > > > > > > point--yes, a 409 is useful to avoid forwarding requests > > to an > > > > > > > > incorrect > > > > > > > > > > leader, but it's also useful to ensure that there aren't > > any > > > > > > > unresolved > > > > > > > > > > writes to the config topic that might cause issues with > the > > > > > request > > > > > > > > (such > > > > > > > > > > as deleting the connector). > > > > > > > > > > > > > > > > > > > > 5. That's a good point--it may be misleading to call a > > > > connector > > > > > > > > STOPPED > > > > > > > > > > when it has zombie tasks lying around on the cluster. I > > don't > > > > > think > > > > > > > > it'd > > > > > > > > > be > > > > > > > > > > appropriate to do this synchronously while handling > > requests to > > > > > the > > > > > > > PUT > > > > > > > > > > /connectors/{connector}/stop since we'd want to give all > > > > > > > > > currently-running > > > > > > > > > > tasks a chance to gracefully shut down, though. I'm also > > not > > > > sure > > > > > > > that > > > > > > > > > this > > > > > > > > > > is a significant problem, either. If the connector is > > resumed, > > > > > then > > > > > > > all > > > > > > > > > > zombie tasks will be automatically fenced out by their > > > > > successors on > > > > > > > > > > startup; if it's deleted, then we'll have wasted effort > by > > > > > performing > > > > > > > > an > > > > > > > > > > unnecessary round of fencing. It may be nice to guarantee > > that > > > > > source > > > > > > > > > task > > > > > > > > > > resources will be deallocated after the connector > > transitions > > > > to > > > > > > > > STOPPED, > > > > > > > > > > but realistically, it doesn't do much to just fence out > > their > > > > > > > > producers, > > > > > > > > > > since tasks can be blocked on a number of other > operations > > such > > > > > as > > > > > > > > > > key/value/header conversion, transformation, and task > > polling. > > > > > It may > > > > > > > > be > > > > > > > > > a > > > > > > > > > > little strange if data is produced to Kafka after the > > connector > > > > > has > > > > > > > > > > transitioned to STOPPED, but we can't provide the same > > > > > guarantees for > > > > > > > > > sink > > > > > > > > > > connectors, since their tasks may be stuck on a > > long-running > > > > > > > > > SinkTask::put > > > > > > > > > > that emits data even after the Connect framework has > > abandoned > > > > > them > > > > > > > > after > > > > > > > > > > exhausting their graceful shutdown timeout. Ultimately, > I'd > > > > > prefer to > > > > > > > > err > > > > > > > > > > on the side of consistency and ease of implementation for > > now, > > > > > but I > > > > > > > > may > > > > > > > > > be > > > > > > > > > > missing a case where a few extra records from a task > that's > > > > slow > > > > > to > > > > > > > > shut > > > > > > > > > > down may cause serious issues--let me know. > > > > > > > > > > > > > > > > > > > > 6. I'm hesitant to propose deprecation of the PAUSED > state > > > > right > > > > > now > > > > > > > as > > > > > > > > > it > > > > > > > > > > does serve a few purposes. Leaving tasks idling-but-ready > > makes > > > > > > > > resuming > > > > > > > > > > them less disruptive across the cluster, since a > rebalance > > > > isn't > > > > > > > > > necessary. > > > > > > > > > > It also reduces latency to resume the connector, > > especially for > > > > > ones > > > > > > > > that > > > > > > > > > > have to do a lot of state gathering on initialization to, > > e.g., > > > > > read > > > > > > > > > > offsets from an external system. > > > > > > > > > > > > > > > > > > > > 7. There should be no risk of mixed tasks after a > > downgrade, > > > > > thanks > > > > > > > to > > > > > > > > > the > > > > > > > > > > empty set of task configs that gets published to the > config > > > > > topic. > > > > > > > Both > > > > > > > > > > upgraded and downgraded workers will render an empty set > of > > > > > tasks for > > > > > > > > the > > > > > > > > > > connector, and keep that set of empty tasks until the > > connector > > > > > is > > > > > > > > > resumed. > > > > > > > > > > Does that address your concerns? > > > > > > > > > > > > > > > > > > > > You're also correct that the linked Jira ticket was > wrong; > > > > > thanks for > > > > > > > > > > pointing that out! Yes, KAFKA-4107 is the intended > ticket, > > and > > > > > I've > > > > > > > > > updated > > > > > > > > > > the link in the KIP accordingly. > > > > > > > > > > > > > > > > > > > > Cheers, > > > > > > > > > > > > > > > > > > > > Chris > > > > > > > > > > > > > > > > > > > > [1] - https://issues.apache.org/jira/browse/KAFKA-4107 > > > > > > > > > > > > > > > > > > > > On Sun, Oct 16, 2022 at 10:42 AM Yash Mayya < > > > > > yash.ma...@gmail.com> > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > Hi Chris, > > > > > > > > > > > > > > > > > > > > > > Thanks a lot for this KIP, I think something like this > > has > > > > been > > > > > > > long > > > > > > > > > > > overdue for Kafka Connect :) > > > > > > > > > > > > > > > > > > > > > > Some thoughts and questions that I had - > > > > > > > > > > > > > > > > > > > > > > 1. I'm wondering if you could elaborate a little more > on > > the > > > > > use > > > > > > > case > > > > > > > > > for > > > > > > > > > > > the `DELETE /connectors/{connector}/offsets` API. I > > think we > > > > > can > > > > > > > all > > > > > > > > > > agree > > > > > > > > > > > that a fine grained reset API that allows setting > > arbitrary > > > > > offsets > > > > > > > > for > > > > > > > > > > > partitions would be quite useful (which you talk about > > in the > > > > > > > Future > > > > > > > > > work > > > > > > > > > > > section). But for the `DELETE > > > > /connectors/{connector}/offsets` > > > > > API > > > > > > > in > > > > > > > > > its > > > > > > > > > > > described form, it looks like it would only serve a > > seemingly > > > > > niche > > > > > > > > use > > > > > > > > > > > case where users want to avoid renaming connectors - > > because > > > > > this > > > > > > > new > > > > > > > > > way > > > > > > > > > > > of resetting offsets actually has more steps (i.e. stop > > the > > > > > > > > connector, > > > > > > > > > > > reset offsets via the API, resume the connector) than > > simply > > > > > > > deleting > > > > > > > > > and > > > > > > > > > > > re-creating the connector with a different name? > > > > > > > > > > > > > > > > > > > > > > 2. The KIP talks about taking care that the response > > formats > > > > > > > > > (presumably > > > > > > > > > > > only talking about the new GET API here) are > symmetrical > > for > > > > > both > > > > > > > > > source > > > > > > > > > > > and sink connectors - is the end goal to have users of > > Kafka > > > > > > > Connect > > > > > > > > > not > > > > > > > > > > > even be aware that sink connectors use Kafka consumers > > under > > > > > the > > > > > > > hood > > > > > > > > > > (i.e. > > > > > > > > > > > have that as purely an implementation detail abstracted > > away > > > > > from > > > > > > > > > users)? > > > > > > > > > > > While I understand the value of uniformity here, the > > response > > > > > > > format > > > > > > > > > for > > > > > > > > > > > sink connectors currently looks a little odd with the > > > > > "partition" > > > > > > > > field > > > > > > > > > > > having "topic" and "partition" as sub-fields, > especially > > to > > > > > users > > > > > > > > > > familiar > > > > > > > > > > > with Kafka semantics. Thoughts? > > > > > > > > > > > > > > > > > > > > > > 3. Another little nitpick on the response format - why > > do we > > > > > need > > > > > > > > > > "source" > > > > > > > > > > > / "sink" as a field under "offsets"? Users can query > the > > > > > connector > > > > > > > > type > > > > > > > > > > via > > > > > > > > > > > the existing `GET /connectors` API. If it's deemed > > important > > > > > to let > > > > > > > > > users > > > > > > > > > > > know that the offsets they're seeing correspond to a > > source / > > > > > sink > > > > > > > > > > > connector, maybe we could have a top level field "type" > > in > > > > the > > > > > > > > response > > > > > > > > > > for > > > > > > > > > > > the `GET /connectors/{connector}/offsets` API similar > to > > the > > > > > `GET > > > > > > > > > > > /connectors` API? > > > > > > > > > > > > > > > > > > > > > > 4. For the `DELETE /connectors/{connector}/offsets` > API, > > the > > > > > KIP > > > > > > > > > mentions > > > > > > > > > > > that requests will be rejected if a rebalance is > pending > > - > > > > > > > presumably > > > > > > > > > > this > > > > > > > > > > > is to avoid forwarding requests to a leader which may > no > > > > > longer be > > > > > > > > the > > > > > > > > > > > leader after the pending rebalance? In this case, the > API > > > > will > > > > > > > > return a > > > > > > > > > > > `409 Conflict` response similar to some of the existing > > APIs, > > > > > > > right? > > > > > > > > > > > > > > > > > > > > > > 5. Regarding fencing out previously running tasks for a > > > > > connector, > > > > > > > do > > > > > > > > > you > > > > > > > > > > > think it would make more sense semantically to have > this > > > > > > > implemented > > > > > > > > in > > > > > > > > > > the > > > > > > > > > > > stop endpoint where an empty set of tasks is generated, > > > > rather > > > > > than > > > > > > > > the > > > > > > > > > > > delete offsets endpoint? This would also give the new > > > > `STOPPED` > > > > > > > > state a > > > > > > > > > > > higher confidence of sorts, with any zombie tasks being > > > > fenced > > > > > off > > > > > > > > from > > > > > > > > > > > continuing to produce data. > > > > > > > > > > > > > > > > > > > > > > 6. Thanks for outlining the issues with the current > > state of > > > > > the > > > > > > > > > `PAUSED` > > > > > > > > > > > state - I think a lot of users expect it to behave like > > the > > > > > > > `STOPPED` > > > > > > > > > > state > > > > > > > > > > > you outline in the KIP and are (unpleasantly) surprised > > when > > > > it > > > > > > > > > doesn't. > > > > > > > > > > > However, this does beg the question of what the > > usefulness of > > > > > > > having > > > > > > > > > two > > > > > > > > > > > separate `PAUSED` and `STOPPED` states is? Do we want > to > > > > > continue > > > > > > > > > > > supporting both these states in the future, or do you > > see the > > > > > > > > `STOPPED` > > > > > > > > > > > state eventually causing the existing `PAUSED` state to > > be > > > > > > > > deprecated? > > > > > > > > > > > > > > > > > > > > > > 7. I think the idea outlined in the KIP for handling a > > new > > > > > state > > > > > > > > during > > > > > > > > > > > cluster downgrades / rolling upgrades is quite clever, > > but do > > > > > you > > > > > > > > think > > > > > > > > > > > there could be any issues with having a mix of "paused" > > and > > > > > > > "stopped" > > > > > > > > > > tasks > > > > > > > > > > > for the same connector across workers in a cluster? At > > the > > > > very > > > > > > > > least, > > > > > > > > > I > > > > > > > > > > > think it would be fairly confusing to most users. I'm > > > > > wondering if > > > > > > > > this > > > > > > > > > > can > > > > > > > > > > > be avoided by stating clearly in the KIP that the new > > `PUT > > > > > > > > > > > /connectors/{connector}/stop` > > > > > > > > > > > can only be used on a cluster that is fully upgraded to > > an AK > > > > > > > version > > > > > > > > > > newer > > > > > > > > > > > than the one which ends up containing changes from this > > KIP > > > > and > > > > > > > that > > > > > > > > > if a > > > > > > > > > > > cluster needs to be downgraded to an older version, the > > user > > > > > should > > > > > > > > > > ensure > > > > > > > > > > > that none of the connectors on the cluster are in a > > stopped > > > > > state? > > > > > > > > With > > > > > > > > > > the > > > > > > > > > > > existing implementation, it looks like an > unknown/invalid > > > > > target > > > > > > > > state > > > > > > > > > > > record is basically just discarded (with an error > message > > > > > logged), > > > > > > > so > > > > > > > > > it > > > > > > > > > > > doesn't seem to be a disastrous failure scenario that > can > > > > bring > > > > > > > down > > > > > > > > a > > > > > > > > > > > worker. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > Yash > > > > > > > > > > > > > > > > > > > > > > On Fri, Oct 14, 2022 at 8:35 PM Chris Egerton > > > > > > > > <chr...@aiven.io.invalid > > > > > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > Hi Ashwin, > > > > > > > > > > > > > > > > > > > > > > > > Thanks for your thoughts. Regarding your questions: > > > > > > > > > > > > > > > > > > > > > > > > 1. The response would show the offsets that are > > visible to > > > > > the > > > > > > > > source > > > > > > > > > > > > connector, so it would combine the contents of the > two > > > > > topics, > > > > > > > > giving > > > > > > > > > > > > priority to offsets present in the connector-specific > > > > topic. > > > > > I'm > > > > > > > > > > > imagining > > > > > > > > > > > > a follow-up question that some people may have in > > response > > > > to > > > > > > > that > > > > > > > > is > > > > > > > > > > > > whether we'd want to provide insight into the > contents > > of a > > > > > > > single > > > > > > > > > > topic > > > > > > > > > > > at > > > > > > > > > > > > a time. It may be useful to be able to see this > > information > > > > > in > > > > > > > > order > > > > > > > > > to > > > > > > > > > > > > debug connector issues or verify that it's safe to > stop > > > > > using a > > > > > > > > > > > > connector-specific offsets topic (either explicitly, > or > > > > > > > implicitly > > > > > > > > > via > > > > > > > > > > > > cluster downgrade). What do you think about adding a > > URL > > > > > query > > > > > > > > > > parameter > > > > > > > > > > > > that allows users to dictate which view of the > > connector's > > > > > > > offsets > > > > > > > > > they > > > > > > > > > > > are > > > > > > > > > > > > given in the REST response, with options for the > > worker's > > > > > global > > > > > > > > > topic, > > > > > > > > > > > the > > > > > > > > > > > > connector-specific topic, and the combined view of > them > > > > that > > > > > the > > > > > > > > > > > connector > > > > > > > > > > > > and its tasks see (which would be the default)? This > > may be > > > > > too > > > > > > > > much > > > > > > > > > > for > > > > > > > > > > > V1 > > > > > > > > > > > > but it feels like it's at least worth exploring a > bit. > > > > > > > > > > > > > > > > > > > > > > > > 2. There is no option for this at the moment. Reset > > > > > semantics are > > > > > > > > > > > extremely > > > > > > > > > > > > coarse-grained; for source connectors, we delete all > > source > > > > > > > > offsets, > > > > > > > > > > and > > > > > > > > > > > > for sink connectors, we delete the entire consumer > > group. > > > > I'm > > > > > > > > hoping > > > > > > > > > > this > > > > > > > > > > > > will be enough for V1 and that, if there's sufficient > > > > demand > > > > > for > > > > > > > > it, > > > > > > > > > we > > > > > > > > > > > can > > > > > > > > > > > > introduce a richer API for resetting or even > modifying > > > > > connector > > > > > > > > > > offsets > > > > > > > > > > > in > > > > > > > > > > > > a follow-up KIP. > > > > > > > > > > > > > > > > > > > > > > > > 3. Good eye :) I think it's fine to keep the existing > > > > > behavior > > > > > > > for > > > > > > > > > the > > > > > > > > > > > > PAUSED state with the Connector instance, since the > > primary > > > > > > > purpose > > > > > > > > > of > > > > > > > > > > > the > > > > > > > > > > > > Connector is to generate task configs and monitor the > > > > > external > > > > > > > > system > > > > > > > > > > for > > > > > > > > > > > > changes. If there's no chance for tasks to be running > > > > > anyways, I > > > > > > > > > don't > > > > > > > > > > > see > > > > > > > > > > > > much value in allowing paused connectors to generate > > new > > > > task > > > > > > > > > configs, > > > > > > > > > > > > especially since each time that happens a rebalance > is > > > > > triggered > > > > > > > > and > > > > > > > > > > > > there's a non-zero cost to that. What do you think? > > > > > > > > > > > > > > > > > > > > > > > > Cheers, > > > > > > > > > > > > > > > > > > > > > > > > Chris > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Oct 14, 2022 at 12:59 AM Ashwin > > > > > > > > <apan...@confluent.io.invalid > > > > > > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for KIP Chris - I think this is a useful > > feature. > > > > > > > > > > > > > > > > > > > > > > > > > > Can you please elaborate on the following in the > KIP > > - > > > > > > > > > > > > > > > > > > > > > > > > > > 1. How would the response of GET > > > > > > > /connectors/{connector}/offsets > > > > > > > > > look > > > > > > > > > > > > like > > > > > > > > > > > > > if the worker has both global and connector > specific > > > > > offsets > > > > > > > > topic > > > > > > > > > ? > > > > > > > > > > > > > > > > > > > > > > > > > > 2. How can we pass the reset options like shift-by > , > > > > > > > to-date-time > > > > > > > > > > etc. > > > > > > > > > > > > > using a REST API like DELETE > > > > > /connectors/{connector}/offsets ? > > > > > > > > > > > > > > > > > > > > > > > > > > 3. Today PAUSE operation on a connector invokes its > > stop > > > > > > > method - > > > > > > > > > > will > > > > > > > > > > > > > there be a change here to reduce confusion with the > > new > > > > > > > proposed > > > > > > > > > > > STOPPED > > > > > > > > > > > > > state ? > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > > > Ashwin > > > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Oct 14, 2022 at 2:22 AM Chris Egerton > > > > > > > > > > <chr...@aiven.io.invalid > > > > > > > > > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi all, > > > > > > > > > > > > > > > > > > > > > > > > > > > > I noticed a fairly large gap in the first version > > of > > > > > this KIP > > > > > > > > > that > > > > > > > > > > I > > > > > > > > > > > > > > published last Friday, which has to do with > > > > accommodating > > > > > > > > > > connectors > > > > > > > > > > > > > > that target different Kafka clusters than the one > > that > > > > > the > > > > > > > > Kafka > > > > > > > > > > > > Connect > > > > > > > > > > > > > > cluster uses for its internal topics and source > > > > > connectors > > > > > > > with > > > > > > > > > > > > dedicated > > > > > > > > > > > > > > offsets topics. I've since updated the KIP to > > address > > > > > this > > > > > > > gap, > > > > > > > > > > which > > > > > > > > > > > > has > > > > > > > > > > > > > > substantially altered the design. Wanted to give > a > > > > > heads-up > > > > > > > to > > > > > > > > > > anyone > > > > > > > > > > > > > > that's already started reviewing. > > > > > > > > > > > > > > > > > > > > > > > > > > > > Cheers, > > > > > > > > > > > > > > > > > > > > > > > > > > > > Chris > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Oct 7, 2022 at 1:29 PM Chris Egerton < > > > > > > > chr...@aiven.io> > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi all, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > I'd like to begin discussion on a KIP to add > > offsets > > > > > > > support > > > > > > > > to > > > > > > > > > > the > > > > > > > > > > > > > Kafka > > > > > > > > > > > > > > > Connect REST API: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-875%3A+First-class+offsets+support+in+Kafka+Connect > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Cheers, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Chris > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Nov 4, 2022 at 10:27 PM Chris Egerton > > > > <chr...@aiven.io.invalid > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > Hi Yash, > > > > > > > > > > > > > > > > Thanks again for your thoughts! Responses to ongoing > > discussions > > > > > inline > > > > > > > > (easier to track context than referencing comment numbers): > > > > > > > > > > > > > > > > > However, this then leads me to wonder if we can make that > > > > explicit > > > > > by > > > > > > > > including "connect" or "connector" in the higher level field > > names? > > > > > Or do > > > > > > > > you think this isn't required given that we're talking about > a > > > > > Connect > > > > > > > > specific REST API in the first place? > > > > > > > > > > > > > > > > I think "partition" and "offset" are fine as field names but > > I'm > > > > not > > > > > > > hugely > > > > > > > > opposed to adding "connector " as a prefix to them; would be > > > > > interested > > > > > > > in > > > > > > > > others' thoughts. > > > > > > > > > > > > > > > > > I'm not sure I followed why the unresolved writes to the > > config > > > > > topic > > > > > > > > would be an issue - wouldn't the delete offsets request be > > added to > > > > > the > > > > > > > > herder's request queue and whenever it is processed, we'd > > anyway > > > > > need to > > > > > > > > check if all the prerequisites for the request are satisfied? > > > > > > > > > > > > > > > > Some requests are handled in multiple steps. For example, > > deleting > > > > a > > > > > > > > connector (1) adds a request to the herder queue to write a > > > > > tombstone to > > > > > > > > the config topic (or, if the worker isn't the leader, forward > > the > > > > > request > > > > > > > > to the leader). (2) Once that tombstone is picked up, (3) a > > > > rebalance > > > > > > > > ensues, and then after it's finally complete, (4) the > > connector and > > > > > its > > > > > > > > tasks are shut down. I probably could have used better > > terminology, > > > > > but > > > > > > > > what I meant by "unresolved writes to the config topic" was a > > case > > > > in > > > > > > > > between steps (2) and (3)--where the worker has already read > > that > > > > > > > tombstone > > > > > > > > from the config topic and knows that a rebalance is pending, > > but > > > > > hasn't > > > > > > > > begun participating in that rebalance yet. In the > > DistributedHerder > > > > > > > class, > > > > > > > > this is done via the `checkRebalanceNeeded` method. > > > > > > > > > > > > > > > > > We can probably revisit this potential deprecation [of the > > PAUSED > > > > > > > state] > > > > > > > > in the future based on user feedback and how the adoption of > > the > > > > new > > > > > > > > proposed stop endpoint looks like, what do you think? > > > > > > > > > > > > > > > > Yeah, revisiting in the future seems reasonable. 👍 > > > > > > > > > > > > > > > > And responses to new comments here: > > > > > > > > > > > > > > > > 8. Yep, we'll start tracking offsets by connector. I don't > > believe > > > > > this > > > > > > > > should be too difficult, and suspect that the only reason we > > track > > > > > raw > > > > > > > byte > > > > > > > > arrays instead of pre-deserializing offset topic information > > into > > > > > > > something > > > > > > > > more useful is because Connect originally had pluggable > > internal > > > > > > > > converters. Now that we're hardcoded to use the JSON > converter > > it > > > > > should > > > > > > > be > > > > > > > > fine to track offsets on a per-connector basis as they're > read > > from > > > > > the > > > > > > > > offsets topic. > > > > > > > > > > > > > > > > 9. I'm hesitant to introduce this type of feature right now > > because > > > > > of > > > > > > > all > > > > > > > > of the gotchas that would come with it. In security-conscious > > > > > > > environments, > > > > > > > > it's possible that a sink connector's principal may have > > access to > > > > > the > > > > > > > > consumer group used by the connector, but the worker's > > principal > > > > may > > > > > not. > > > > > > > > There's also the case where source connectors have separate > > offsets > > > > > > > topics, > > > > > > > > or sink connectors have overridden consumer group IDs, or > sink > > or > > > > > source > > > > > > > > connectors work against a different Kafka cluster than the > one > > that > > > > > their > > > > > > > > worker uses. Overall, I'd rather provide a single API that > > works in > > > > > all > > > > > > > > cases rather than risk confusing and alienating users by > > trying to > > > > > make > > > > > > > > their lives easier in a subset of cases. > > > > > > > > > > > > > > > > 10. Hmm... I don't think the order of the writes matters too > > much > > > > > here, > > > > > > > but > > > > > > > > we probably could start by deleting from the global topic > > first, > > > > > that's > > > > > > > > true. The reason I'm not hugely concerned about this case is > > that > > > > if > > > > > > > > something goes wrong while resetting offsets, there's no > > immediate > > > > > > > > impact--the connector will still be in the STOPPED state. The > > REST > > > > > > > response > > > > > > > > for requests to reset the offsets will clearly call out that > > the > > > > > > > operation > > > > > > > > has failed, and if necessary, we can probably also add a > > > > > scary-looking > > > > > > > > warning message stating that we can't guarantee which offsets > > have > > > > > been > > > > > > > > successfully wiped and which haven't. Users can query the > exact > > > > > offsets > > > > > > > of > > > > > > > > the connector at this point to determine what will happen > > if/what > > > > > they > > > > > > > > resume it. And they can repeat attempts to reset the offsets > as > > > > many > > > > > > > times > > > > > > > > as they'd like until they get back a 2XX response, indicating > > that > > > > > it's > > > > > > > > finally safe to resume the connector. Thoughts? > > > > > > > > > > > > > > > > 11. I haven't thought too much about it. I think something > > like the > > > > > > > > Monitorable* connectors would probably serve our needs here; > > we can > > > > > > > > instantiate them on a running Connect cluster and then use > > various > > > > > > > handles > > > > > > > > to know how many times they've been polled, committed > records, > > etc. > > > > > If > > > > > > > > necessary we can tweak those classes or even write our own. > But > > > > > anyways, > > > > > > > > once that's all done, the test will be something like > "create a > > > > > > > connector, > > > > > > > > wait for it to produce N records (each of which contains some > > kind > > > > of > > > > > > > > predictable offset), and ensure that the offsets for it in > the > > REST > > > > > API > > > > > > > > match up with the ones we'd expect from N records". Does that > > > > answer > > > > > your > > > > > > > > question? > > > > > > > > > > > > > > > > Cheers, > > > > > > > > > > > > > > > > Chris > > > > > > > > > > > > > > > > On Tue, Oct 18, 2022 at 3:28 AM Yash Mayya < > > yash.ma...@gmail.com> > > > > > wrote: > > > > > > > > > > > > > > > > > Hi Chris, > > > > > > > > > > > > > > > > > > 1. Thanks a lot for elaborating on this, I'm now convinced > > about > > > > > the > > > > > > > > > usefulness of the new offset reset endpoint. Regarding the > > > > > follow-up > > > > > > > KIP > > > > > > > > > for a fine-grained offset write API, I'd be happy to take > > that on > > > > > once > > > > > > > > this > > > > > > > > > KIP is finalized and I will definitely look forward to your > > > > > feedback on > > > > > > > > > that one! > > > > > > > > > > > > > > > > > > 2. Gotcha, the motivation makes more sense to me now. So > the > > > > higher > > > > > > > level > > > > > > > > > partition field represents a Connect specific "logical > > partition" > > > > > of > > > > > > > > sorts > > > > > > > > > - i.e. the source partition as defined by a connector for > > source > > > > > > > > connectors > > > > > > > > > and a Kafka topic + partition for sink connectors. I like > the > > > > idea > > > > > of > > > > > > > > > adding a Kafka prefix to the lower level partition/offset > > (and > > > > > topic) > > > > > > > > > fields which basically makes it more clear (although > > implicitly) > > > > > that > > > > > > > the > > > > > > > > > higher level partition/offset field is Connect specific and > > not > > > > the > > > > > > > same > > > > > > > > as > > > > > > > > > what those terms represent in Kafka itself. However, this > > then > > > > > leads me > > > > > > > > to > > > > > > > > > wonder if we can make that explicit by including "connect" > or > > > > > > > "connector" > > > > > > > > > in the higher level field names? Or do you think this isn't > > > > > required > > > > > > > > given > > > > > > > > > that we're talking about a Connect specific REST API in the > > first > > > > > > > place? > > > > > > > > > > > > > > > > > > 3. Thanks, I think the response structure definitely looks > > better > > > > > now! > > > > > > > > > > > > > > > > > > 4. Interesting, I'd be curious to learn why we might want > to > > > > change > > > > > > > this > > > > > > > > in > > > > > > > > > the future but that's probably out of scope for this > > discussion. > > > > > I'm > > > > > > > not > > > > > > > > > sure I followed why the unresolved writes to the config > topic > > > > > would be > > > > > > > an > > > > > > > > > issue - wouldn't the delete offsets request be added to the > > > > > herder's > > > > > > > > > request queue and whenever it is processed, we'd anyway > need > > to > > > > > check > > > > > > > if > > > > > > > > > all the prerequisites for the request are satisfied? > > > > > > > > > > > > > > > > > > 5. Thanks for elaborating that just fencing out the > producer > > > > still > > > > > > > leaves > > > > > > > > > many cases where source tasks remain hanging around and > also > > that > > > > > we > > > > > > > > anyway > > > > > > > > > can't have similar data production guarantees for sink > > connectors > > > > > right > > > > > > > > > now. I agree that it might be better to go with ease of > > > > > implementation > > > > > > > > and > > > > > > > > > consistency for now. > > > > > > > > > > > > > > > > > > 6. Right, that does make sense but I still feel like the > two > > > > states > > > > > > > will > > > > > > > > > end up being confusing to end users who might not be able > to > > > > > discern > > > > > > > the > > > > > > > > > (fairly low-level) differences between them (also the > > nuances of > > > > > state > > > > > > > > > transitions like STOPPED -> PAUSED or PAUSED -> STOPPED > with > > the > > > > > > > > > rebalancing implications as well). We can probably revisit > > this > > > > > > > potential > > > > > > > > > deprecation in the future based on user feedback and how > the > > > > > adoption > > > > > > > of > > > > > > > > > the new proposed stop endpoint looks like, what do you > think? > > > > > > > > > > > > > > > > > > 7. Aha, that is completely my bad, I missed that the v1/v2 > > state > > > > is > > > > > > > only > > > > > > > > > applicable to the connector's target state and that we > don't > > need > > > > > to > > > > > > > > worry > > > > > > > > > about the tasks since we will have an empty set of tasks. I > > > > think I > > > > > > > was a > > > > > > > > > little confused by "pause the parts of the connector that > > they > > > > are > > > > > > > > > assigned" from the KIP. Thanks for clarifying that! > > > > > > > > > > > > > > > > > > > > > > > > > > > Some more thoughts and questions that I had - > > > > > > > > > > > > > > > > > > 8. Could you elaborate on what the implementation for > offset > > > > reset > > > > > for > > > > > > > > > source connectors would look like? Currently, it doesn't > look > > > > like > > > > > we > > > > > > > > track > > > > > > > > > all the partitions for a source connector anywhere. Will we > > need > > > > to > > > > > > > > > book-keep this somewhere in order to be able to emit a > > tombstone > > > > > record > > > > > > > > for > > > > > > > > > each source partition? > > > > > > > > > > > > > > > > > > 9. The KIP describes the offset reset endpoint as only > being > > > > > usable on > > > > > > > > > existing connectors that are in a `STOPPED` state. Why > > wouldn't > > > > we > > > > > want > > > > > > > > to > > > > > > > > > allow resetting offsets for a deleted connector which seems > > to > > > > be a > > > > > > > valid > > > > > > > > > use case? Or do we plan to handle this use case only via > the > > item > > > > > > > > outlined > > > > > > > > > in the future work section - "Automatically delete offsets > > with > > > > > > > > > connectors"? > > > > > > > > > > > > > > > > > > 10. The KIP mentions that source offsets will be reset > > > > > transactionally > > > > > > > > for > > > > > > > > > each topic (worker global offset topic and connector > specific > > > > > offset > > > > > > > > topic > > > > > > > > > if it exists). While it obviously isn't possible to > > atomically do > > > > > the > > > > > > > > > writes to two topics which may be on different Kafka > > clusters, > > > > I'm > > > > > > > > > wondering about what would happen if the first transaction > > > > > succeeds but > > > > > > > > the > > > > > > > > > second one fails. I think the order of the two transactions > > > > matters > > > > > > > here > > > > > > > > - > > > > > > > > > if we successfully emit tombstones to the connector > specific > > > > offset > > > > > > > topic > > > > > > > > > and fail to do so for the worker global offset topic, we'll > > > > > presumably > > > > > > > > fail > > > > > > > > > the offset delete request because the KIP mentions that "A > > > > request > > > > > to > > > > > > > > reset > > > > > > > > > offsets for a source connector will only be considered > > successful > > > > > if > > > > > > > the > > > > > > > > > worker is able to delete all known offsets for that > > connector, on > > > > > both > > > > > > > > the > > > > > > > > > worker's global offsets topic and (if one is used) the > > > > connector's > > > > > > > > > dedicated offsets topic.". However, this will lead to the > > > > connector > > > > > > > only > > > > > > > > > being able to read potentially older offsets from the > worker > > > > global > > > > > > > > offset > > > > > > > > > topic on resumption (based on the combined offset view > > presented > > > > as > > > > > > > > > described in KIP-618 [1]). So, I think we should make sure > > that > > > > the > > > > > > > > worker > > > > > > > > > global offset topic tombstoning is attempted first, right? > > Note > > > > > that in > > > > > > > > the > > > > > > > > > current implementation of > > `ConnectorOffsetBackingStore::set`, the > > > > > > > > primary / > > > > > > > > > connector specific offset store is written to first. > > > > > > > > > > > > > > > > > > 11. This probably isn't necessary to elaborate on in the > KIP > > > > > itself, > > > > > > > but > > > > > > > > I > > > > > > > > > was wondering what the second offset test - "verify that > that > > > > those > > > > > > > > offsets > > > > > > > > > reflect an expected level of progress for each connector > > (i.e., > > > > > they > > > > > > > are > > > > > > > > > greater than or equal to a certain value depending on how > the > > > > > > > connectors > > > > > > > > > are configured and how long they have been running)" - > would > > look > > > > > like? > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > Yash > > > > > > > > > > > > > > > > > > [1] - > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=153817402#KIP618:ExactlyOnceSupportforSourceConnectors-Smoothmigration > > > > > > > > > > > > > > > > > > On Tue, Oct 18, 2022 at 12:42 AM Chris Egerton > > > > > <chr...@aiven.io.invalid > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > Hi Yash, > > > > > > > > > > > > > > > > > > > > Thanks for your detailed thoughts. > > > > > > > > > > > > > > > > > > > > 1. In KAFKA-4107 [1], the primary request is exactly > what's > > > > > proposed > > > > > > > in > > > > > > > > > the > > > > > > > > > > KIP right now: a way to reset offsets for connectors. > Sure, > > > > > there's > > > > > > > an > > > > > > > > > > extra step of stopping the connector, but renaming a > > connector > > > > > isn't > > > > > > > as > > > > > > > > > > convenient of an alternative as it may seem since in many > > cases > > > > > you'd > > > > > > > > > also > > > > > > > > > > want to delete the older one, so the complete sequence of > > steps > > > > > would > > > > > > > > be > > > > > > > > > > something like delete the old connector, rename it > > (possibly > > > > > > > requiring > > > > > > > > > > modifications to its config file, depending on which API > is > > > > > used), > > > > > > > then > > > > > > > > > > create the renamed variant. It's also just not a great > user > > > > > > > > > > experience--even if the practical impacts are limited > > (which, > > > > > IMO, > > > > > > > they > > > > > > > > > are > > > > > > > > > > not), people have been asking for years about why they > > have to > > > > > employ > > > > > > > > > this > > > > > > > > > > kind of a workaround for a fairly common use case, and we > > don't > > > > > > > really > > > > > > > > > have > > > > > > > > > > a good answer beyond "we haven't implemented something > > better > > > > > yet". > > > > > > > On > > > > > > > > > top > > > > > > > > > > of that, you may have external tooling that needs to be > > tweaked > > > > > to > > > > > > > > > handle a > > > > > > > > > > new connector name, you may have strict authorization > > policies > > > > > around > > > > > > > > who > > > > > > > > > > can access what connectors, you may have other ACLs > > attached to > > > > > the > > > > > > > > name > > > > > > > > > of > > > > > > > > > > the connector (which can be especially common in the case > > of > > > > sink > > > > > > > > > > connectors, whose consumer group IDs are tied to their > > names by > > > > > > > > default), > > > > > > > > > > and leaving around state in the offsets topic that can > > never be > > > > > > > cleaned > > > > > > > > > up > > > > > > > > > > presents a bit of a footgun for users. It may not be a > > silver > > > > > bullet, > > > > > > > > but > > > > > > > > > > providing some mechanism to reset that state is a step in > > the > > > > > right > > > > > > > > > > direction and allows responsible users to more carefully > > > > > administer > > > > > > > > their > > > > > > > > > > cluster without resorting to non-public APIs. That said, > I > > do > > > > > agree > > > > > > > > that > > > > > > > > > a > > > > > > > > > > fine-grained reset/overwrite API would be useful, and I'd > > be > > > > > happy to > > > > > > > > > > review a KIP to add that feature if anyone wants to > tackle > > it! > > > > > > > > > > > > > > > > > > > > 2. Keeping the two formats symmetrical is motivated > mostly > > by > > > > > > > > aesthetics > > > > > > > > > > and quality-of-life for programmatic interaction with the > > API; > > > > > it's > > > > > > > not > > > > > > > > > > really a goal to hide the use of consumer groups from > > users. I > > > > do > > > > > > > agree > > > > > > > > > > that the format is a little strange-looking for sink > > > > connectors, > > > > > but > > > > > > > it > > > > > > > > > > seemed like it would be easier to work with for UIs, > > casual jq > > > > > > > queries, > > > > > > > > > and > > > > > > > > > > CLIs than a more Kafka-specific alternative such as > > > > > > > > > {"<topic>-<partition>": > > > > > > > > > > "<offset>"}, and although it is a little strange, I don't > > think > > > > > it's > > > > > > > > any > > > > > > > > > > less readable or intuitive. That said, I've made some > > tweaks to > > > > > the > > > > > > > > > format > > > > > > > > > > that should make programmatic access even easier; > > specifically, > > > > > I've > > > > > > > > > > removed the "source" and "sink" wrapper fields and > instead > > > > moved > > > > > them > > > > > > > > > into > > > > > > > > > > a top-level object with a "type" and "offsets" field, > just > > like > > > > > you > > > > > > > > > > suggested in point 3 (thanks!). We might also consider > > changing > > > > > the > > > > > > > > field > > > > > > > > > > names for sink offsets from "topic", "partition", and > > "offset" > > > > to > > > > > > > > "Kafka > > > > > > > > > > topic", "Kafka partition", and "Kafka offset" > > respectively, to > > > > > reduce > > > > > > > > the > > > > > > > > > > stuttering effect of having a "partition" field inside a > > > > > "partition" > > > > > > > > > field > > > > > > > > > > and the same with an "offset" field; thoughts? One final > > > > > point--by > > > > > > > > > equating > > > > > > > > > > source and sink offsets, we probably make it easier for > > users > > > > to > > > > > > > > > understand > > > > > > > > > > exactly what a source offset is; anyone who's familiar > with > > > > > consumer > > > > > > > > > > offsets can see from the response format that we > identify a > > > > > logical > > > > > > > > > > partition as a combination of two entities (a topic and a > > > > > partition > > > > > > > > > > number); it should make it easier to grok what a source > > offset > > > > > is by > > > > > > > > > seeing > > > > > > > > > > what the two formats have in common. > > > > > > > > > > > > > > > > > > > > 3. Great idea! Done. > > > > > > > > > > > > > > > > > > > > 4. Yes, I'm thinking right now that a 409 will be the > > response > > > > > status > > > > > > > > if > > > > > > > > > a > > > > > > > > > > rebalance is pending. I'd rather not add this to the KIP > > as we > > > > > may > > > > > > > want > > > > > > > > > to > > > > > > > > > > change it at some point and it doesn't seem vital to > > establish > > > > > it as > > > > > > > > part > > > > > > > > > > of the public contract for the new endpoint right now. > > Also, > > > > > small > > > > > > > > > > point--yes, a 409 is useful to avoid forwarding requests > > to an > > > > > > > > incorrect > > > > > > > > > > leader, but it's also useful to ensure that there aren't > > any > > > > > > > unresolved > > > > > > > > > > writes to the config topic that might cause issues with > the > > > > > request > > > > > > > > (such > > > > > > > > > > as deleting the connector). > > > > > > > > > > > > > > > > > > > > 5. That's a good point--it may be misleading to call a > > > > connector > > > > > > > > STOPPED > > > > > > > > > > when it has zombie tasks lying around on the cluster. I > > don't > > > > > think > > > > > > > > it'd > > > > > > > > > be > > > > > > > > > > appropriate to do this synchronously while handling > > requests to > > > > > the > > > > > > > PUT > > > > > > > > > > /connectors/{connector}/stop since we'd want to give all > > > > > > > > > currently-running > > > > > > > > > > tasks a chance to gracefully shut down, though. I'm also > > not > > > > sure > > > > > > > that > > > > > > > > > this > > > > > > > > > > is a significant problem, either. If the connector is > > resumed, > > > > > then > > > > > > > all > > > > > > > > > > zombie tasks will be automatically fenced out by their > > > > > successors on > > > > > > > > > > startup; if it's deleted, then we'll have wasted effort > by > > > > > performing > > > > > > > > an > > > > > > > > > > unnecessary round of fencing. It may be nice to guarantee > > that > > > > > source > > > > > > > > > task > > > > > > > > > > resources will be deallocated after the connector > > transitions > > > > to > > > > > > > > STOPPED, > > > > > > > > > > but realistically, it doesn't do much to just fence out > > their > > > > > > > > producers, > > > > > > > > > > since tasks can be blocked on a number of other > operations > > such > > > > > as > > > > > > > > > > key/value/header conversion, transformation, and task > > polling. > > > > > It may > > > > > > > > be > > > > > > > > > a > > > > > > > > > > little strange if data is produced to Kafka after the > > connector > > > > > has > > > > > > > > > > transitioned to STOPPED, but we can't provide the same > > > > > guarantees for > > > > > > > > > sink > > > > > > > > > > connectors, since their tasks may be stuck on a > > long-running > > > > > > > > > SinkTask::put > > > > > > > > > > that emits data even after the Connect framework has > > abandoned > > > > > them > > > > > > > > after > > > > > > > > > > exhausting their graceful shutdown timeout. Ultimately, > I'd > > > > > prefer to > > > > > > > > err > > > > > > > > > > on the side of consistency and ease of implementation for > > now, > > > > > but I > > > > > > > > may > > > > > > > > > be > > > > > > > > > > missing a case where a few extra records from a task > that's > > > > slow > > > > > to > > > > > > > > shut > > > > > > > > > > down may cause serious issues--let me know. > > > > > > > > > > > > > > > > > > > > 6. I'm hesitant to propose deprecation of the PAUSED > state > > > > right > > > > > now > > > > > > > as > > > > > > > > > it > > > > > > > > > > does serve a few purposes. Leaving tasks idling-but-ready > > makes > > > > > > > > resuming > > > > > > > > > > them less disruptive across the cluster, since a > rebalance > > > > isn't > > > > > > > > > necessary. > > > > > > > > > > It also reduces latency to resume the connector, > > especially for > > > > > ones > > > > > > > > that > > > > > > > > > > have to do a lot of state gathering on initialization to, > > e.g., > > > > > read > > > > > > > > > > offsets from an external system. > > > > > > > > > > > > > > > > > > > > 7. There should be no risk of mixed tasks after a > > downgrade, > > > > > thanks > > > > > > > to > > > > > > > > > the > > > > > > > > > > empty set of task configs that gets published to the > config > > > > > topic. > > > > > > > Both > > > > > > > > > > upgraded and downgraded workers will render an empty set > of > > > > > tasks for > > > > > > > > the > > > > > > > > > > connector, and keep that set of empty tasks until the > > connector > > > > > is > > > > > > > > > resumed. > > > > > > > > > > Does that address your concerns? > > > > > > > > > > > > > > > > > > > > You're also correct that the linked Jira ticket was > wrong; > > > > > thanks for > > > > > > > > > > pointing that out! Yes, KAFKA-4107 is the intended > ticket, > > and > > > > > I've > > > > > > > > > updated > > > > > > > > > > the link in the KIP accordingly. > > > > > > > > > > > > > > > > > > > > Cheers, > > > > > > > > > > > > > > > > > > > > Chris > > > > > > > > > > > > > > > > > > > > [1] - https://issues.apache.org/jira/browse/KAFKA-4107 > > > > > > > > > > > > > > > > > > > > On Sun, Oct 16, 2022 at 10:42 AM Yash Mayya < > > > > > yash.ma...@gmail.com> > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > Hi Chris, > > > > > > > > > > > > > > > > > > > > > > Thanks a lot for this KIP, I think something like this > > has > > > > been > > > > > > > long > > > > > > > > > > > overdue for Kafka Connect :) > > > > > > > > > > > > > > > > > > > > > > Some thoughts and questions that I had - > > > > > > > > > > > > > > > > > > > > > > 1. I'm wondering if you could elaborate a little more > on > > the > > > > > use > > > > > > > case > > > > > > > > > for > > > > > > > > > > > the `DELETE /connectors/{connector}/offsets` API. I > > think we > > > > > can > > > > > > > all > > > > > > > > > > agree > > > > > > > > > > > that a fine grained reset API that allows setting > > arbitrary > > > > > offsets > > > > > > > > for > > > > > > > > > > > partitions would be quite useful (which you talk about > > in the > > > > > > > Future > > > > > > > > > work > > > > > > > > > > > section). But for the `DELETE > > > > /connectors/{connector}/offsets` > > > > > API > > > > > > > in > > > > > > > > > its > > > > > > > > > > > described form, it looks like it would only serve a > > seemingly > > > > > niche > > > > > > > > use > > > > > > > > > > > case where users want to avoid renaming connectors - > > because > > > > > this > > > > > > > new > > > > > > > > > way > > > > > > > > > > > of resetting offsets actually has more steps (i.e. stop > > the > > > > > > > > connector, > > > > > > > > > > > reset offsets via the API, resume the connector) than > > simply > > > > > > > deleting > > > > > > > > > and > > > > > > > > > > > re-creating the connector with a different name? > > > > > > > > > > > > > > > > > > > > > > 2. The KIP talks about taking care that the response > > formats > > > > > > > > > (presumably > > > > > > > > > > > only talking about the new GET API here) are > symmetrical > > for > > > > > both > > > > > > > > > source > > > > > > > > > > > and sink connectors - is the end goal to have users of > > Kafka > > > > > > > Connect > > > > > > > > > not > > > > > > > > > > > even be aware that sink connectors use Kafka consumers > > under > > > > > the > > > > > > > hood > > > > > > > > > > (i.e. > > > > > > > > > > > have that as purely an implementation detail abstracted > > away > > > > > from > > > > > > > > > users)? > > > > > > > > > > > While I understand the value of uniformity here, the > > response > > > > > > > format > > > > > > > > > for > > > > > > > > > > > sink connectors currently looks a little odd with the > > > > > "partition" > > > > > > > > field > > > > > > > > > > > having "topic" and "partition" as sub-fields, > especially > > to > > > > > users > > > > > > > > > > familiar > > > > > > > > > > > with Kafka semantics. Thoughts? > > > > > > > > > > > > > > > > > > > > > > 3. Another little nitpick on the response format - why > > do we > > > > > need > > > > > > > > > > "source" > > > > > > > > > > > / "sink" as a field under "offsets"? Users can query > the > > > > > connector > > > > > > > > type > > > > > > > > > > via > > > > > > > > > > > the existing `GET /connectors` API. If it's deemed > > important > > > > > to let > > > > > > > > > users > > > > > > > > > > > know that the offsets they're seeing correspond to a > > source / > > > > > sink > > > > > > > > > > > connector, maybe we could have a top level field "type" > > in > > > > the > > > > > > > > response > > > > > > > > > > for > > > > > > > > > > > the `GET /connectors/{connector}/offsets` API similar > to > > the > > > > > `GET > > > > > > > > > > > /connectors` API? > > > > > > > > > > > > > > > > > > > > > > 4. For the `DELETE /connectors/{connector}/offsets` > API, > > the > > > > > KIP > > > > > > > > > mentions > > > > > > > > > > > that requests will be rejected if a rebalance is > pending > > - > > > > > > > presumably > > > > > > > > > > this > > > > > > > > > > > is to avoid forwarding requests to a leader which may > no > > > > > longer be > > > > > > > > the > > > > > > > > > > > leader after the pending rebalance? In this case, the > API > > > > will > > > > > > > > return a > > > > > > > > > > > `409 Conflict` response similar to some of the existing > > APIs, > > > > > > > right? > > > > > > > > > > > > > > > > > > > > > > 5. Regarding fencing out previously running tasks for a > > > > > connector, > > > > > > > do > > > > > > > > > you > > > > > > > > > > > think it would make more sense semantically to have > this > > > > > > > implemented > > > > > > > > in > > > > > > > > > > the > > > > > > > > > > > stop endpoint where an empty set of tasks is generated, > > > > rather > > > > > than > > > > > > > > the > > > > > > > > > > > delete offsets endpoint? This would also give the new > > > > `STOPPED` > > > > > > > > state a > > > > > > > > > > > higher confidence of sorts, with any zombie tasks being > > > > fenced > > > > > off > > > > > > > > from > > > > > > > > > > > continuing to produce data. > > > > > > > > > > > > > > > > > > > > > > 6. Thanks for outlining the issues with the current > > state of > > > > > the > > > > > > > > > `PAUSED` > > > > > > > > > > > state - I think a lot of users expect it to behave like > > the > > > > > > > `STOPPED` > > > > > > > > > > state > > > > > > > > > > > you outline in the KIP and are (unpleasantly) surprised > > when > > > > it > > > > > > > > > doesn't. > > > > > > > > > > > However, this does beg the question of what the > > usefulness of > > > > > > > having > > > > > > > > > two > > > > > > > > > > > separate `PAUSED` and `STOPPED` states is? Do we want > to > > > > > continue > > > > > > > > > > > supporting both these states in the future, or do you > > see the > > > > > > > > `STOPPED` > > > > > > > > > > > state eventually causing the existing `PAUSED` state to > > be > > > > > > > > deprecated? > > > > > > > > > > > > > > > > > > > > > > 7. I think the idea outlined in the KIP for handling a > > new > > > > > state > > > > > > > > during > > > > > > > > > > > cluster downgrades / rolling upgrades is quite clever, > > but do > > > > > you > > > > > > > > think > > > > > > > > > > > there could be any issues with having a mix of "paused" > > and > > > > > > > "stopped" > > > > > > > > > > tasks > > > > > > > > > > > for the same connector across workers in a cluster? At > > the > > > > very > > > > > > > > least, > > > > > > > > > I > > > > > > > > > > > think it would be fairly confusing to most users. I'm > > > > > wondering if > > > > > > > > this > > > > > > > > > > can > > > > > > > > > > > be avoided by stating clearly in the KIP that the new > > `PUT > > > > > > > > > > > /connectors/{connector}/stop` > > > > > > > > > > > can only be used on a cluster that is fully upgraded to > > an AK > > > > > > > version > > > > > > > > > > newer > > > > > > > > > > > than the one which ends up containing changes from this > > KIP > > > > and > > > > > > > that > > > > > > > > > if a > > > > > > > > > > > cluster needs to be downgraded to an older version, the > > user > > > > > should > > > > > > > > > > ensure > > > > > > > > > > > that none of the connectors on the cluster are in a > > stopped > > > > > state? > > > > > > > > With > > > > > > > > > > the > > > > > > > > > > > existing implementation, it looks like an > unknown/invalid > > > > > target > > > > > > > > state > > > > > > > > > > > record is basically just discarded (with an error > message > > > > > logged), > > > > > > > so > > > > > > > > > it > > > > > > > > > > > doesn't seem to be a disastrous failure scenario that > can > > > > bring > > > > > > > down > > > > > > > > a > > > > > > > > > > > worker. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > Yash > > > > > > > > > > > > > > > > > > > > > > On Fri, Oct 14, 2022 at 8:35 PM Chris Egerton > > > > > > > > <chr...@aiven.io.invalid > > > > > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > Hi Ashwin, > > > > > > > > > > > > > > > > > > > > > > > > Thanks for your thoughts. Regarding your questions: > > > > > > > > > > > > > > > > > > > > > > > > 1. The response would show the offsets that are > > visible to > > > > > the > > > > > > > > source > > > > > > > > > > > > connector, so it would combine the contents of the > two > > > > > topics, > > > > > > > > giving > > > > > > > > > > > > priority to offsets present in the connector-specific > > > > topic. > > > > > I'm > > > > > > > > > > > imagining > > > > > > > > > > > > a follow-up question that some people may have in > > response > > > > to > > > > > > > that > > > > > > > > is > > > > > > > > > > > > whether we'd want to provide insight into the > contents > > of a > > > > > > > single > > > > > > > > > > topic > > > > > > > > > > > at > > > > > > > > > > > > a time. It may be useful to be able to see this > > information > > > > > in > > > > > > > > order > > > > > > > > > to > > > > > > > > > > > > debug connector issues or verify that it's safe to > stop > > > > > using a > > > > > > > > > > > > connector-specific offsets topic (either explicitly, > or > > > > > > > implicitly > > > > > > > > > via > > > > > > > > > > > > cluster downgrade). What do you think about adding a > > URL > > > > > query > > > > > > > > > > parameter > > > > > > > > > > > > that allows users to dictate which view of the > > connector's > > > > > > > offsets > > > > > > > > > they > > > > > > > > > > > are > > > > > > > > > > > > given in the REST response, with options for the > > worker's > > > > > global > > > > > > > > > topic, > > > > > > > > > > > the > > > > > > > > > > > > connector-specific topic, and the combined view of > them > > > > that > > > > > the > > > > > > > > > > > connector > > > > > > > > > > > > and its tasks see (which would be the default)? This > > may be > > > > > too > > > > > > > > much > > > > > > > > > > for > > > > > > > > > > > V1 > > > > > > > > > > > > but it feels like it's at least worth exploring a > bit. > > > > > > > > > > > > > > > > > > > > > > > > 2. There is no option for this at the moment. Reset > > > > > semantics are > > > > > > > > > > > extremely > > > > > > > > > > > > coarse-grained; for source connectors, we delete all > > source > > > > > > > > offsets, > > > > > > > > > > and > > > > > > > > > > > > for sink connectors, we delete the entire consumer > > group. > > > > I'm > > > > > > > > hoping > > > > > > > > > > this > > > > > > > > > > > > will be enough for V1 and that, if there's sufficient > > > > demand > > > > > for > > > > > > > > it, > > > > > > > > > we > > > > > > > > > > > can > > > > > > > > > > > > introduce a richer API for resetting or even > modifying > > > > > connector > > > > > > > > > > offsets > > > > > > > > > > > in > > > > > > > > > > > > a follow-up KIP. > > > > > > > > > > > > > > > > > > > > > > > > 3. Good eye :) I think it's fine to keep the existing > > > > > behavior > > > > > > > for > > > > > > > > > the > > > > > > > > > > > > PAUSED state with the Connector instance, since the > > primary > > > > > > > purpose > > > > > > > > > of > > > > > > > > > > > the > > > > > > > > > > > > Connector is to generate task configs and monitor the > > > > > external > > > > > > > > system > > > > > > > > > > for > > > > > > > > > > > > changes. If there's no chance for tasks to be running > > > > > anyways, I > > > > > > > > > don't > > > > > > > > > > > see > > > > > > > > > > > > much value in allowing paused connectors to generate > > new > > > > task > > > > > > > > > configs, > > > > > > > > > > > > especially since each time that happens a rebalance > is > > > > > triggered > > > > > > > > and > > > > > > > > > > > > there's a non-zero cost to that. What do you think? > > > > > > > > > > > > > > > > > > > > > > > > Cheers, > > > > > > > > > > > > > > > > > > > > > > > > Chris > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Oct 14, 2022 at 12:59 AM Ashwin > > > > > > > > <apan...@confluent.io.invalid > > > > > > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for KIP Chris - I think this is a useful > > feature. > > > > > > > > > > > > > > > > > > > > > > > > > > Can you please elaborate on the following in the > KIP > > - > > > > > > > > > > > > > > > > > > > > > > > > > > 1. How would the response of GET > > > > > > > /connectors/{connector}/offsets > > > > > > > > > look > > > > > > > > > > > > like > > > > > > > > > > > > > if the worker has both global and connector > specific > > > > > offsets > > > > > > > > topic > > > > > > > > > ? > > > > > > > > > > > > > > > > > > > > > > > > > > 2. How can we pass the reset options like shift-by > , > > > > > > > to-date-time > > > > > > > > > > etc. > > > > > > > > > > > > > using a REST API like DELETE > > > > > /connectors/{connector}/offsets ? > > > > > > > > > > > > > > > > > > > > > > > > > > 3. Today PAUSE operation on a connector invokes its > > stop > > > > > > > method - > > > > > > > > > > will > > > > > > > > > > > > > there be a change here to reduce confusion with the > > new > > > > > > > proposed > > > > > > > > > > > STOPPED > > > > > > > > > > > > > state ? > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > > > Ashwin > > > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Oct 14, 2022 at 2:22 AM Chris Egerton > > > > > > > > > > <chr...@aiven.io.invalid > > > > > > > > > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi all, > > > > > > > > > > > > > > > > > > > > > > > > > > > > I noticed a fairly large gap in the first version > > of > > > > > this KIP > > > > > > > > > that > > > > > > > > > > I > > > > > > > > > > > > > > published last Friday, which has to do with > > > > accommodating > > > > > > > > > > connectors > > > > > > > > > > > > > > that target different Kafka clusters than the one > > that > > > > > the > > > > > > > > Kafka > > > > > > > > > > > > Connect > > > > > > > > > > > > > > cluster uses for its internal topics and source > > > > > connectors > > > > > > > with > > > > > > > > > > > > dedicated > > > > > > > > > > > > > > offsets topics. I've since updated the KIP to > > address > > > > > this > > > > > > > gap, > > > > > > > > > > which > > > > > > > > > > > > has > > > > > > > > > > > > > > substantially altered the design. Wanted to give > a > > > > > heads-up > > > > > > > to > > > > > > > > > > anyone > > > > > > > > > > > > > > that's already started reviewing. > > > > > > > > > > > > > > > > > > > > > > > > > > > > Cheers, > > > > > > > > > > > > > > > > > > > > > > > > > > > > Chris > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Oct 7, 2022 at 1:29 PM Chris Egerton < > > > > > > > chr...@aiven.io> > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi all, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > I'd like to begin discussion on a KIP to add > > offsets > > > > > > > support > > > > > > > > to > > > > > > > > > > the > > > > > > > > > > > > > Kafka > > > > > > > > > > > > > > > Connect REST API: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-875%3A+First-class+offsets+support+in+Kafka+Connect > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Cheers, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Chris > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
