Hi all,
Good news everyone! I've reworked the design one more time and hopefully
some of the improvements here should make the proposal more palatable.
TL;DR:
- Rolling upgrades are now possible, in at most two phases; workers will
first have to be given a binary upgrade to 3.0 (the targeted version for
this feature) which can be a rolling upgrade, and then a rolling upgrade to
enable exactly-once source support in a cluster should be possible with no
anticipated downtime for source connectors or their tasks
- Offset topic migration is completely removed in favor of fallback to the
global offsets topic (much simpler!)
- One backwards-incompatible change is introduced: the leader will be
required to use a transactional producer for writes to the config topic
regardless of whether exactly-once support is enabled on the cluster.
Technically we could gate this behind a config property but since the
benefits of a transactional producer actually extend beyond exactly-once
source support (we can now ensure that there's only one writer to the
config topic at any given time, which isn't guaranteed with the current
model) and the cost to accommodate it is fairly low (a handful of
well-defined and limited-scope ACLs), I erred on the side of keeping things
simple
Looking forward to the next round of review and really hoping we can get
the ball rolling in time for this to land with 3.0!
Cheers,
Chris
On Mon, Apr 12, 2021 at 7:51 AM Chris Egerton <chr...@confluent.io> wrote:
> Hi Randall,
>
> After thinking things over carefully, I've done some reworking of the
> design. Instead of performing zombie fencing during rebalance, the leader
> will expose an internal REST endpoint that will allow workers to request a
> round of zombie fencing on demand, at any time. Workers will then hit this
> endpoint after starting connectors and after task config updates for
> connectors are detected; the precise details of this are outlined in the
> KIP. If a round of fencing should fail for any reason, the worker will be
> able to mark its Connector failed and, if the user wants to retry, they can
> simply restart the Connector via the REST API (specifically, the POST
> /connectors/{connector}/restart endpoint).
>
> The idea I'd been playing with to allow workers to directly write to the
> config topic seemed promising at first, but it allowed things to get pretty
> hairy for users if any kind of rebalancing bug took place and two workers
> believed they owned the same Connector object.
>
> I hope this answers any outstanding questions and look forward to your
> thoughts.
>
> Cheers,
>
> Chris
>
> On Mon, Mar 22, 2021 at 4:38 PM Chris Egerton <chr...@confluent.io> wrote:
>
>> Hi Randall,
>>
>> No complaints about email size from me. Let's dive in!
>>
>> 1. Sure! Especially important in my mind is that this is already possible
>> with Connect as it is today, and users can benefit from this with or
>> without the expanded exactly-once souce support we're trying to add with
>> this KIP. I've added that info to the "Motivation" section and included a
>> brief overview of the idempotent producer in the "Background and
>> References" section.
>>
>> 2. I actually hadn't considered enabling exactly-once source support by
>> default. Thinking over it now, I'm a little hesitant to do so just because,
>> even with the best testing out there, it's a pretty large change and it
>> seems safest to try to make it opt-in in case there's unanticipated
>> fallout. Then again, when incremental cooperative rebalancing was
>> introduced, it was made opt-out instead of opt-in. However, ICR came with
>> lower known risk of breaking existing users' setups; we know for a fact
>> that, if you haven't granted your worker or connector principals some ACLs
>> on Kafka, your connectors will fail. In an ideal world people would
>> carefully read upgrade notes and either grant those permissions or disable
>> the feature before upgrading their Connect cluster to 3.0, but if they
>> don't, they'll be in for a world of hurt. Overall I'd be more comfortable
>> letting this feature incubate for a little bit to let everyone get familiar
>> with it before possibly enabling it in 4.0 by default; what do you think?
>>
>> 3. I didn't think too long about the name for the offsets topic property;
>> it seemed pretty unlikely that it'd conflict with existing connector
>> property names. One alternative could be to follow the idiom established by
>> KIP-458 and include the word "override" in there somewhere, but none of the
>> resulting names seem very intuitive ("offsets.storage.override.topic" seems
>> the best to me but even that isn't super clear). Happy to field suggestions
>> if anyone has any alternatives they'd like to propose.
>>
>> 4. I _really_ wanted to enable per-connector toggling of this feature for
>> the exact reasons you've outlined. There's already a couple cases where
>> some piece of functionality was introduced that users could only control at
>> the worker config level and, later, effort had to be made in order to add
>> per-connector granularity: key/value converters and connector Kafka clients
>> are the two at the top of my head, and there may be others. So if at all
>> possible, it'd be great if we could support this. The only thing standing
>> in the way is that it allows exactly-once delivery to be compromised, which
>> in my estimation was unacceptable. I'm hoping we can make this feature
>> great enough that it'll work with most if not all source connectors out
>> there, and users won't even want to toggle it on a per-connector basis.
>> Otherwise, we'll have to decide between forcing users to split their
>> connectors across two Connect clusters (one with exactly-once source
>> enabled, one with it disabled), which would be awful, or potentially seeing
>> duplicate record delivery for exactly-once connectors, which is also awful.
>>
>> 5. Existing source connectors won't necessarily have to be configured
>> with "consumer.override" properties, but there are a couple of cases where
>> that will be necessary:
>>
>> a. Connector is running against against a secured Kafka cluster and the
>> principal configured in the worker via the various "consumer." properties
>> (if there is one) doesn't have permission to access the offsets topic that
>> the connector will use. If no "consumer.override." properties are specified
>> in this case, the connector and its tasks will fail.
>>
>> b. Connector is running against a separate Kafka cluster from the one
>> specified in the worker's config with the "bootstrap.servers" property (or
>> the "consumer.bootstrap.servers" property, if specified). For an example,
>> the connector is configured with a "producer.override.boostrap.servers"
>> property. If no corresponding "consumer.override.bootstrap.servers"
>> property is specified by the connector (or on is provided but doesn't match
>> the one used for the producer), bad things will happen--best case, the
>> connector refuses to start up because it's configured with a separate
>> offsets topic and no offsets migration is able to take place; worst case,
>> the connector starts reading offsets from a stale topic and every time a
>> task is restarted, duplicate records ensue. Since this can compromise
>> delivery guarantees I think action is worth taking here, and I'm glad
>> you've raised this scenario. Proposed approach: If the user hasn't
>> specified a "consumer.override.bootstrap.servers" property in their
>> connector config and the "bootstrap.servers" property that we'd configure
>> its consumer with (based on either the "consumer.bootstrap.servers" worker
>> property or, if not provided, the "bootstrap.servers" worker property) is
>> different from the "bootstrap.servers" that we'd configure its producer
>> with, we can log a warning but then automatically configure the consumer
>> with the "bootstrap.servers" that are going to be used for the producer. If
>> the user has specified a "consumer.override.bootstrap.servers" property in
>> their connector config and it differs from the one that the connector's
>> producer will use, we can fail the connector and all of its tasks, in order
>> to prevent delivery guarantees from being compromised. Thoughts?
>>
>> 6. Sure, I've updated the KIP with that change in all the places it
>> applies.
>>
>> 7. Sure, I've updated the KIP to be clear that these methods will also be
>> invoked for the end-of-life offset commit of a task scheduled for shutdown.
>>
>> 8.
>>
>> > Won't there also be transaction contention just within a single
>> connector using multiple tasks? Even if that connector has all offsets go
>> to a dedicated topic (separate from all other connectors), then isn't it
>> still possible/likely that a transaction from task 1 blocks a transaction
>> from task 2?
>>
>> Yes, this is completely correct.
>>
>> > How will this affect latency perceived by consumers (e.g., the
>> records from task 2's tranasction don't become visible until the
>> transactions from task 1 and task 2 commit)?
>>
>> Worst case, task startup and requests from existing tasks to read from
>> the offsets topic will block for up to the producer transaction timeout for
>> the hung transaction. However, this is fairly unlikely as it would require
>> two things to happen (or not happen):
>>
>> a. The task would have to die in between writing at least one record to
>> the offsets topic and committing the transaction, which in most cases
>> should be a fairly small window.
>>
>> b. The task would have to not be restarted (either manually by the user,
>> or automatically by some monitoring system) before then. If the task were
>> to be restarted, it'd cause any hanging transactions from older task
>> instances to be aborted immediately.
>>
>> I can add some of this to the KIP if you think it's worth calling out,
>> just let me know.
>>
>> 9. I'll try to clear it up but I'm hoping that the text very soon after
>> that section helps clarify this point already: "the worker will continue to
>> read to the end of the topic until either offset information for the
>> connector or a sentinel value is found". Is the muddiness here perhaps
>> because it's not explicitly stated that sentinel values will apply on a
>> per-connector basis (as opposed to, e.g., a per-offsets-topic basis)?
>>
>> 10.
>>
>> > This section says "then workers will not bring up tasks for the
>> connector." What does this mean? Does it mean the tasks fail?
>>
>> I'm not in favor of failing tasks since that would either require manual
>> intervention on the part of the user or go against the current flow of
>> failed task recovery (i.e., requiring manual intervention) by automatically
>> transitioning the task from the "FAILED" state to "RUNNING" once the
>> condition for safe startup was met. Task restarts would also have no effect
>> in this case, and putting the task into a state where a restart is
>> frequently the appropriate remedial action for users could be misleading.
>>
>> I was hoping we could leave existing tasks in the UNASSIGNED state.
>> Warning log messages can be emitted notifying the user that the worker was
>> assigned tasks but is not able to start them yet due to constraints
>> introduced by exactly-once source support.
>>
>> I suspect there may be some more discussion on this topic before a design
>> is settled on, so I'll hold off on updating the KIP for now.
>>
>> > How do users recover from this?
>>
>> (Addressed with response to 11)
>>
>> 11.
>>
>> > What exactly does "refuse to bring the task up" mean?
>>
>> I hope I've covered this with the information from my response to part 10.
>>
>> > Also, "until a rebalance occurs" suggests that another rebalance will
>> be forthcoming? What would
>> cause that subsequent rebalance? Isn't it possible that there is no
>> subsequent rebalance for a potentially significant amount of time?
>>
>> (And, from 10):
>>
>> > How do users recover from this?
>>
>> This area is a little rough and more detail is definitely warranted.
>> Unfortunately, there's also some open questions that, after some thought,
>> aren't really covered by the design I had in mind.
>>
>> I was originally thinking it'd go something like this: leader fails to
>> fence out producers in time during rebalance, rebalance continues, workers
>> all notice that latest task configs for the connector are after the latest
>> task count record, workers don't bring up tasks (leaving them UNASSIGNED),
>> then eventually leader is able to finish fencing out producers, writes a
>> new task count record to the config topic, and after the ensuing rebalance,
>> all workers begin to bring up tasks for that connector.
>>
>> Unfortunately, a key assumption here is that the leader will eventually
>> succeed at fencing out producers for the connector, which isn't always
>> going to be the case if, for example, it lacks the right ACLs on the Kafka
>> cluster. If the leader fails to perform zombie fencing with a non-retriable
>> error, it's important that this information be surfaced to the user, and
>> log messages and even JMX metrics don't seem powerful enough for this case.
>>
>> I've toyed with the idea of allowing non-leader workers to take on the
>> work of zombie fencing and producing task count records to the config
>> topic. We could then assign responsibility for this action to the worker
>> that owns the Connector object for a connector. This would also require
>> task configs to be written to the config topic by those same workers, since
>> otherwise the worker that owns the Connector object might accidentally
>> write a task count record to the config topic immediately after the leader
>> has written a new set of task configs, and it would look like the new set
>> of task configs is ready to use even though the task count record actually
>> corresponded to an earlier set of task configs (in fact, this is actually a
>> possibility with the current design if a zombie leader writes new task
>> configs to the config topic during a rolling upgrade).
>>
>> With that in mind, I think an improved design might be something like
>> this:
>>
>> a. Connector is reconfigured
>> b. The worker that owns the Connector object for that connector uses a
>> transactional producer whose ID corresponds to the connector name to write
>> those task configs to the config topic
>> c. Workers pick up those new task configs and preemptively attempt to
>> shut down all tasks for that connector
>> c. Rebalance ensues
>> d. Worker that owns the Connector object for that connector performs a
>> round of zombie fencing and then, using a transactional producer whose ID
>> corresponds to the connector name, writes the task count record for that
>> connector to the config topic
>> e. Workers pick up the task count record and start tasks for that
>> connector
>>
>> If step d) fails with a non-retriable error, the worker can mark the
>> Connector object "FAILED" in the status API with a stack trace
>> corresponding to the error encountered while trying to perform the round of
>> zombie fencing, and a message stating that no new tasks for the connector
>> can be started until this problem is addressed. And, if the user tries to
>> restart the connector, another round of zombie fencing can be attempted.
>>
>> How does this sound to you, especially in terms of correctness (i.e., not
>> adding potential holes in delivery guarantees) and user experience? If it
>> looks good I'll update the KIP accordingly, but I suspect there might still
>> be some discussion on this topic.
>>
>> Further discussion on the bug where a zombie leader writes task configs
>> to the config topic after a task count record has been produced during a
>> rolling upgrade to come later as a response to 15.
>>
>> 12.
>>
>> > What exactly does "abort startup of the task" mean?
>> How do users recover from this? Again, I more explicit behavior would be
>> helpful.
>>
>> "Abort" is probably the wrong term to use here. "Abandon and re-attempt"
>> might be better. Essentially, we'd discard the current task, but with the
>> assumption that a new one has either already been brought up in its place,
>> or will be soon as a response to the rebalance caused by the new set of
>> task configurations for the connector. No action on the part of the user
>> should be required here. I've updated the section to add this info.
>>
>> 13.
>>
>> > Isn't it possible given N producer transactional IDs for there to be up
>> to
>> N transaction coordinators?
>>
>> I believe there is only one transaction coordinator per broker, so we'd
>> only have up to B transactional coordinators where B is the number of
>> brokers.
>>
>> > Does this reduce the likelihood that it can be
>> completed successfully? What risk does this play in rebalances being
>> unsuccessful?
>>
>> Regardless of the info above, these are still valid questions. I think
>> the short answer to the first part is "yes"--the more transaction
>> coordinators you have to reach at once, the more likely it becomes that
>> something goes wrong somewhere. With regards to "successful" or
>> "unsuccessful" rebalances, I'm guessing you're really asking about the
>> round of zombie fencing being performed during them, since the only impact
>> on the rebalance itself will be that things might take a little longer with
>> the leader may blocking during assignment. As far as the viability of
>> zombie fencing goes, it does introduce a point of failure that will prevent
>> tasks from being brought up for a connector, which isn't great. However, if
>> one or more transaction coordinators associated with a connector's tasks
>> aren't available, those tasks can't be brought up anyways, since it can't
>> be guaranteed that the old generation has been fenced out. I'm optimistic
>> that, with the improvements (and the retry path offered) in response to 11,
>> this won't be a showstopper and won't cause too much pain for users.
>>
>> 14. Thanks, added.
>>
>> 15.
>>
>> > IIUC, the net effect is that rolling upgrades don't really work for
>> source
>> connectors.
>>
>> Yep, that's correct. If you get unlucky and the worker that owns the
>> Connector object for your connector is the last one to be upgraded, it
>> won't be able to perform the offset migration until the upgrade is
>> complete, and all of the tasks for that connector will be stuck idling in
>> the meantime waiting for that to complete.
>>
>> > Wouldn't it just be easier and maybe faster to stop and restart
>> the whole cluster?
>>
>> Sure, if you're unconcerned with any sink connectors that might be
>> running on the cluster. I've also realized that, with the possibility of
>> zombie leaders writing task configs to the config topic during a rolling
>> upgrade (and possibly doing so right before a task count record is
>> written), there's some vulnerability and confusion that might come about
>> during a rolling upgrade and turn users off of the feature. We can
>> definitely engineer around this (possibly by including some kind of flag
>> with task configs that indicates that they've been written by a worker
>> using a transactional producer, and ignoring any task configs that don't
>> use that flag), but at this point I'm kind of leaning in the direction of
>> requiring (or at least, strongly recommending) a mass upgrade for clusters
>> when enabling exactly-once source support, since no matter what, they're
>> still going to see downtime with source connectors. Curious what your
>> thoughts are on this, though.
>>
>> > Also, what are the states of the source tasks during
>> this time?
>>
>> Covered above in response to another question, but to reiterate: they
>> will be UNASSIGNED.
>>
>> 16. Good call, added.
>>
>> 17. Sure, I'll add a "Limitations" section with the points you've
>> described and any others that seem applicable.
>>
>>
>> Thanks again for taking a look!
>>
>> Cheers,
>>
>> Chris
>>
>> On Mon, Mar 22, 2021 at 11:39 AM Chris Egerton <chr...@confluent.io>
>> wrote:
>>
>>> Hi Randall,
>>>
>>> Thanks for taking a look! Responses inline:
>>>
>>> > This would seem to make the transactions based upon time rather than
>>> record
>>> count. Is that really the intent? Did you consider using one transaction
>>> per "batch" of records, whatever number of records that means? One issue
>>> with the time-based boundary is that `offset.flush.interval.ms`
>>> defaults to
>>> 60 seconds, and that could mean many hundreds of thousands of records on
>>> a
>>> high throughput task. (BTW, I do agree with the KIP's rejected
>>> alternative
>>> to not expose to the task or let the task define the transaction
>>> boundaries.)
>>>
>>> Yep, this was the intent. I covered a few alternatives in the "Finer
>>> control over offset commits" section in the rejected alternatives all the
>>> way at the bottom, including allowing source tasks to define their own
>>> transaction boundaries, allowing per-connector source offset intervals, and
>>> allowing offsets to be committed after a certain number of records. While
>>> writing this up I also realized that, besides establishing batch boundaries
>>> based on the number of records, another possibility could also be to count
>>> each collection of records returned from a single call to SourceTask::poll
>>> as its own batch, although this might seriously hamper throughput for tasks
>>> that provide a small number of records at a time, but do so with high
>>> frequency.
>>>
>>> I was hoping we could save this for future work since it's unclear that
>>> it would be necessary to operate Connect with exactly-once source support.
>>> It almost seems like a bit of a leaky abstraction for users to have to
>>> worry about the details of offset commit logic, and I'm hoping we can do
>>> some magic behind the curtains to avoid this. Here's a few thoughts on how
>>> we can keep fixed-time-interval offset flushing with exactly-once source
>>> support without making things worse for users:
>>>
>>> 1. Ignore the offset flush timeout parameter for exactly-once source
>>> tasks. They should be allowed to take as long as they want to complete an
>>> offset commit, since the cost of failure at that point is to fail the
>>> source task. This might require a change to how Connect manages source task
>>> offset commits right now since a single worker-global thread is used to
>>> schedule and perform offset commits for all source tasks, and any blocked
>>> task will also block offset commits for all other tasks, but I'm confident
>>> that we can find a way to solve this problem and would like to leave it as
>>> an implementation detail that doesn't need to be covered extensively in
>>> this KIP.
>>>
>>> 2. The only other problem I foresee with high-throughput tasks like the
>>> ones you described is that it may take longer than the transaction timeout
>>> for a task to flush all of its records to Kafka. In this case, there are
>>> two actions that users can take to nurse their connector back to health:
>>> reconfigure it to produce records with a lower throughput, or increase the
>>> transaction timeout for the producers used by the connector. We can include
>>> include these steps in the error message for a task that fails due to
>>> producer transaction timeout.
>>>
>>> Does this seem reasonable? If so, I'll add this info to the design doc.
>>>
>>>
>>> > Gwen's #5 question was about rebalances of a source connector. Your
>>> answer
>>> made sense, but while the KIP talks about source tasks being stopped
>>> before
>>> the connector is rebalanced, I'm wondering about the delay or other
>>> impact
>>> this has on the rebalance itself. The "Cannot fence out producers during
>>> rebalance" section talks about fencing out producers, but IIUC this is
>>> not
>>> the same as:
>>>
>>> > ... all workers will preemptively stop all tasks of all source
>>> connectors
>>> for which task configurations.
>>>
>>> > If this is already addressed in the KIP, please just point me to that
>>> section.
>>>
>>> Not addressed yet :) I haven't called out a potential impact on
>>> rebalance because, although there will probably be one, I don't believe
>>> it'll be substantial. After a connector is reconfigured and its new task
>>> configs are written to the config topic, the sequence of events for a
>>> single worker would be:
>>>
>>> 1. New task configs are read from the config topic
>>> 2. All tasks for that connector are stopped in parallel (which is
>>> already how en-masse task start/stop is performed by a distributed worker)
>>> 3. Any tasks that take longer than the graceful task shutdown timeout
>>> are abandoned and allowed to continue running
>>> 4. Worker rejoins the group
>>>
>>> The only new impact on rebalance might come from steps 2-3, but it's
>>> unlikely to be significant. Distributed workers currently use a pool of
>>> eight threads to handle en-masse start/stop of tasks and connectors, have a
>>> default graceful task shutdown timeout of five seconds, and a default
>>> rebalance timeout of sixty seconds. In order to block the worker for an
>>> extra thirty seconds (only half of the rebalance timeout), there would have
>>> to be (30 / 5) * 8 = 48 tasks that are either completely hung or drag their
>>> feet for five seconds each during shutdown. This would have to either come
>>> from a single connector, or a collection of connectors that were
>>> reconfigured in extremely-rapid succession. The likelihood of this scenario
>>> doesn't seem high enough to worry about initially, but if you feel
>>> differently, we can consider some (hopefully lightweight) optimizations,
>>> such as expanding the size of the distributed herder's start/stop thread
>>> pool.
>>>
>>> One final point--this is also similar to the logic that was used for
>>> eager rebalancing, except that with eager rebalancing, all tasks from all
>>> connectors were stopped before rejoining a group, instead of just the tasks
>>> for recently-updated connectors. Someone correct me if I'm wrong, but the
>>> stopping of tasks before rejoining the group wasn't a pain point with eager
>>> rebalancing; the biggest pain point was just that a change to one connector
>>> would temporarily disable all connectors on the cluster.
>>>
>>> I'll update the KIP with a (brief) note on this in the "Fencing during
>>> rebalance" section; let me know if the reasoning here seems valid.
>>>
>>>
>>> Responses to your second email coming soon after I update the KIP with
>>> the info from above.
>>>
>>> Cheers,
>>>
>>> Chris
>>>
>>> On Thu, Mar 18, 2021 at 7:28 PM Randall Hauch <rha...@apache.org> wrote:
>>>
>>>> Hey, Chris.
>>>>
>>>> Thanks again for all your hard work figuring out this complicated
>>>> problem.
>>>> While there are still issues to iron out, at the moment it seems pretty
>>>> sensible.
>>>>
>>>> Here's my follow up email with additional comments/questions. I've
>>>> organized these by section to help provide context, and I've numbered
>>>> the
>>>> questions to make it easier to reply/discuss. I apologize for the
>>>> length of
>>>> this email -- I thought it better to do this all at once rather than
>>>> string
>>>> out in several emails.
>>>>
>>>> Motivation
>>>> 1. This section does not mention another important source of duplicate
>>>> records: producer retries. Using the idempotent producer can prevent
>>>> duplicates due to retries, but it can't help with anything else. Perhaps
>>>> it's worth mentioning?
>>>>
>>>> Public Interfaces
>>>> 2. Have we considered `exactly.once.source.enabled=true` by default in
>>>> the
>>>> worker config? We're currently on track for considering this KIP for
>>>> inclusion in AK 3.0, so we are allowed backward incompatible changes.
>>>> 3. It is possible (but maybe unlikely) that existing source connectors
>>>> might already have a connector config property that matches the proposed
>>>> new connector config `offsets.storage.topic`. It's good that this
>>>> matches
>>>> the worker's config, but were any other property names considered?
>>>> 4. I see that one of the rejected alternatives was a connector-level
>>>> option
>>>> to disable EOS. For example, what happens if some source connectors
>>>> don't
>>>> work well with EOS (e.g., the transaction blocking behavior of a
>>>> multi-task
>>>> source connector is unacceptable)? What would we say if a user wants to
>>>> use
>>>> EOS for some source connectors but not others? Is it really viable to
>>>> not
>>>> allow that?
>>>> 5. The worker source task will create a consumer for the offset topic
>>>> used
>>>> by the connector. Currently, Connect does not use or do anything with
>>>> `consumer.override.*` properties, but won't they now likely be needed
>>>> with
>>>> a source connector using EOS? What happens with existing source
>>>> connector
>>>> configurations that don't define these?
>>>>
>>>> Proposed Changes
>>>> Offset reads
>>>> 6. The KIP says "a message will be logged notifying them of this fact."
>>>> Should it be more specific, namely "a warning message..."?
>>>>
>>>> SourceTask record commit API
>>>> 7. The second paragraph talks about why we cannot guarantee the methods
>>>> be
>>>> invoked after every successful commit, and I think that makes sense
>>>> because
>>>> there is no other practical way to handle worker failures. It does seem
>>>> like it would be more useful to describe the expectation about calling
>>>> these methods during normal task shutdown. Specifically, should
>>>> connector
>>>> developers expect that these methods will be called following a
>>>> successful
>>>> commit immediately prior to stopping the task (barring worker failures)?
>>>>
>>>> Per-connector offsets topic
>>>> This section says:
>>>>
>>>> > Finally, it allows users to limit the effect that hanging
>>>> transactions on
>>>> > an offsets topic will have. If tasks A and B use the same offsets
>>>> topic,
>>>> > and task A initiates a transaction on that offsets topic right before
>>>> task
>>>> > B starts up, then task A dies suddenly without committing its
>>>> transaction,
>>>> > task B will have to wait for that transaction to time out before it
>>>> can
>>>> > read to the end of the offsets topic. If the transaction timeout is
>>>> set
>>>> > very high for task A (to accommodate bursts of high throughput, for
>>>> > example), this will block task B from processing any data for a long
>>>> time.
>>>> > Although this scenario may be unavoidable in some cases, using a
>>>> dedicated
>>>> > offsets topic for each connector should allow cluster administrators
>>>> to
>>>> > isolate the blast radius of a hanging transaction on an offsets
>>>> topic. This
>>>> > way, although tasks of the same connector may still interfere with
>>>> each
>>>> > other, they will at least not interfere with tasks of other
>>>> connectors.
>>>> > This should be sufficient for most multitenant environments.
>>>>
>>>>
>>>> 8. Won't there also be transaction contention just within a single
>>>> connector using multiple tasks? Even if that connector has all offsets
>>>> go
>>>> to a dedicated topic (separate from all other connectors), then isn't it
>>>> still possible/likely that a transaction from task 1 blocks a
>>>> transaction
>>>> from task 2? How will this affect latency perceived by consumers (e.g.,
>>>> the
>>>> records from task 2's tranasction don't become visible until the
>>>> transactions from task 1 and task 2 commit)?
>>>>
>>>> Migration
>>>> 9. This section says:
>>>>
>>>> > the worker will read to the end of the offsets topic before starting
>>>> the
>>>> > task. If there are no offsets for the connector present in the topic
>>>> and
>>>> > there is no sentinel value...
>>>>
>>>> This sounds like the worker will know that a sentinel value is expected,
>>>> but IIUC a worker cannot know that. Instead, the worker consuming the
>>>> topic
>>>> must read to the end and expect _either_ a sentinel value _or_ committed
>>>> offsets. Can the text make this more clear?
>>>>
>>>> Task count records
>>>> 10. This section says "then workers will not bring up tasks for the
>>>> connector." What does this mean? Does it mean the tasks fail? How do
>>>> users
>>>> recover from this? I think the "Addressed failure/degradation scenarios"
>>>> section tries to address some of these, but any link is implicit.
>>>>
>>>> Source task startup
>>>> 11. This section says "Otherwise, the worker will emit a warning log
>>>> message and refuse to bring the task up until a rebalance occurs." What
>>>> exactly does "refuse to bring the task up" mean? Also, "until a
>>>> rebalance
>>>> occurs" suggests that another rebalance will be forthcoming? What would
>>>> cause that subsequent rebalance? Isn't it possible that there is no
>>>> subsequent rebalance for a potentially significant amount of time?
>>>> Again,
>>>> any link between this section and the "Addressed failure/degradation
>>>> scenarios" section is implicit.
>>>>
>>>> 12. Also, "Once a worker has instantiated a producer for a source task,
>>>> it
>>>> will read to the end of the config topic once more, and if a new set of
>>>> task configurations for that connector has been generated, it will abort
>>>> startup of the task." What exactly does "abort startup of the task"
>>>> mean?
>>>> How do users recover from this? Again, I more explicit behavior would be
>>>> helpful.
>>>>
>>>> Addressed failure/degradation scenarios
>>>> Cannot fence out producer during rebalance
>>>> 13. This section says:
>>>>
>>>> > The leader may be unable to fence out producers during a rebalance
>>>> if, for
>>>> > example, it is unable to reach or locate a transaction coordinator.
>>>>
>>>> Isn't it possible given N producer transactional IDs for there to be up
>>>> to
>>>> N transaction coordinators? Does this reduce the likelihood that it can
>>>> be
>>>> completed successfully? What risk does this play in rebalances being
>>>> unsuccessful?
>>>>
>>>> Rolling upgrade
>>>> 14. This section says:
>>>>
>>>> > Users can stop each worker, upgrade to a later version if necessary,
>>>> set
>>>> > exactly.once.source.enabled to true in the config, then restart, one
>>>> by one.
>>>>
>>>> Nit: This is not as clear/precise as it probably needs to be. A subtle
>>>> change makes this more precise:
>>>>
>>>> > Users can then do the following for each worker, one by one: stop the
>>>> > worker, upgrade to a later version if necessary, set
>>>> > exactly.once.source.enabled to true in the config, then restart.
>>>>
>>>>
>>>> 15. A bit later in the same section:
>>>>
>>>> > This will have the effect that, until the leader is upgraded, no
>>>> upgraded
>>>> > workers will start source tasks.
>>>>
>>>> IIUC, the net effect is that rolling upgrades don't really work for
>>>> source
>>>> connectors. Wouldn't it just be easier and maybe faster to stop and
>>>> restart
>>>> the whole cluster? Also, what are the states of the source tasks during
>>>> this time?
>>>>
>>>> 16. And a bit later:
>>>>
>>>> > Some of these offsets may be out-of-date or not match the ones in the
>>>> > connector’s separate offsets topic
>>>>
>>>> The "not match" is probably looser than necessary. What about "older
>>>> than"?
>>>>
>>>>
>>>> Future Work
>>>> Standalone mode support
>>>> 17. The fact that EOS only works for distributed Connect was mentioned
>>>> only
>>>> once earlier in the document, but I missed it on my first pass and was a
>>>> bit surprised when I read this. Perhaps we should be clearer on this
>>>> earlier on in the document, perhaps by adding a "Limitations" section.
>>>> There are probably several limitations:
>>>>
>>>> - Only Connect's distributed worker supports EOS.
>>>> - EOS can be enabled only for all source connectors, and not only for
>>>> some.
>>>> - EOS works only for source connectors that assign source partitions
>>>> to
>>>> at most one task at a time.
>>>>
>>>> I suspect this last limitation is likely to not be that severe, since
>>>> such
>>>> a connector would have non-deterministic restart behavior with the
>>>> current
>>>> Connect runtime, as each task would commit different source offsets
>>>> using
>>>> the same source partitions. Of course, a following up question
>>>> then becomes: What is the impact of any source connector that doesn't
>>>> adhere to this last limitation? (This is kind of addressed in the KIP,
>>>> but
>>>> may need to be updated/moved if we add the Limitations section.
>>>>
>>>> Best regards,
>>>>
>>>> Randall
>>>>
>>>>
>>>> On Thu, Mar 18, 2021 at 5:26 PM Randall Hauch <rha...@apache.org>
>>>> wrote:
>>>>
>>>> > Chris,
>>>> >
>>>> > Thanks for the well-written KIP and the more recent updates. It's
>>>> exciting
>>>> > to envision EOS for Connect source connectors!
>>>> >
>>>> > I'll follow up this email with another that goes into specific
>>>> comments
>>>> > and questions not already touched on by others. This email addresses
>>>> > comments and questions previously discussed on this thread.
>>>> >
>>>> > I'd like to +1 your response to Gwen's question about
>>>> connector-specific
>>>> > offsets topic (#2 in her email). But importantly, a connector can
>>>> override
>>>> > any producer settings, including Kafka credentials that don't have
>>>> > permissions on the internal Connect offsets topic. For this reason, I
>>>> think
>>>> > it's necessary for this KIP to address those scenarios.
>>>> >
>>>> > Gwen's #4 question was about how transactions related to the batches
>>>> > source connector tasks return from their `poll()` method, and you
>>>> followed
>>>> > up by clarifying the document to make this more clear. IIUC, the KIP
>>>> > suggests still using the existing `offset.flush.interval.ms`
>>>> property and
>>>> > then committing transactions based (solely?) upon the value of the
>>>> > property. This would seem to make the transactions based upon time
>>>> rather
>>>> > than record count. Is that really the intent? Did you consider using
>>>> one
>>>> > transaction per "batch" of records, whatever number of records that
>>>> means?
>>>> > One issue with the time-based boundary is that `
>>>> offset.flush.interval.ms`
>>>> > defaults to 60 seconds, and that could mean many hundreds of
>>>> thousands of
>>>> > records on a high throughput task. (BTW, I do agree with the KIP's
>>>> > rejected alternative to not expose to the task or let the task define
>>>> the
>>>> > transaction boundaries.)
>>>> >
>>>> > Gwen's #5 question was about rebalances of a source connector. Your
>>>> answer
>>>> > made sense, but while the KIP talks about source tasks being stopped
>>>> before
>>>> > the connector is rebalanced, I'm wondering about the delay or other
>>>> impact
>>>> > this has on the rebalance itself. The "Cannot fence out producers
>>>> during
>>>> > rebalance" section talks about fencing out producers, but IIUC this
>>>> is not
>>>> > the same as:
>>>> >
>>>> > ... all workers will preemptively stop all tasks of all source
>>>> connectors
>>>> > for which task configurations.
>>>> >
>>>> >
>>>> > If this is already addressed in the KIP, please just point me to that
>>>> > section.
>>>> >
>>>> > Finally, regarding Guozhang's question about "partial rebalances"
>>>> rather
>>>> > than fencing out all producers. In your initial response, you said the
>>>> > following:
>>>> >
>>>> > Connectors can arbitrarily reassign source partitions (such as
>>>> > database tables, or Kafka topic partitions) across tasks, so even if
>>>> the
>>>> > assignment of tasks across workers remains unchanged, the assignment
>>>> of
>>>> > source partitions across those workers might.
>>>> >
>>>> >
>>>> > This is a very important point. In fact, there are definitely
>>>> connectors
>>>> > that shuffle around "source partition" assignments to different
>>>> tasks, and
>>>> > IMO we have to take that into consideration because Connect provides
>>>> no
>>>> > guidance on this and because there are good reasons to do this. So
>>>> I'm glad
>>>> > the KIP takes this into account. This is actually one of the issues I
>>>> ran
>>>> > into during an earlier attempt to POC EOS for source connectors, and
>>>> the
>>>> > ability to force fencing is a novel (albeit heavy-handed) way to deal
>>>> with
>>>> > that.
>>>> >
>>>> > As I said, I'll follow up in subsequent emails on other specific
>>>> questions
>>>> > I have.
>>>> >
>>>> > Best regards,
>>>> >
>>>> > Randall
>>>> >
>>>> > On Mon, Feb 22, 2021 at 6:45 PM Guozhang Wang <wangg...@gmail.com>
>>>> wrote:
>>>> >
>>>> >> This is a great explanation, thank you!
>>>> >>
>>>> >> On Mon, Feb 22, 2021 at 2:44 PM Chris Egerton <chr...@confluent.io>
>>>> >> wrote:
>>>> >>
>>>> >> > Hi Guozhang,
>>>> >> >
>>>> >> > Your understanding should be correct in most cases, but there are
>>>> two
>>>> >> finer
>>>> >> > points that may interest you:
>>>> >> >
>>>> >> > 1. It's technically dependent on the implementation of the
>>>> connector and
>>>> >> > how it chooses to allocate source partitions across its tasks;
>>>> even if
>>>> >> the
>>>> >> > number of tasks and source partitions remains completely
>>>> unchanged, a
>>>> >> > connector may still choose to shuffle around the partition->task
>>>> >> > allocation. I can't think of any cases where this might happen off
>>>> the
>>>> >> top
>>>> >> > of my head, but it seemed worth sharing given the educational
>>>> nature of
>>>> >> the
>>>> >> > question.
>>>> >> > 2. It's also possible that the number of source partitions remains
>>>> >> > unchanged, but the set of source partitions changes. One case
>>>> where this
>>>> >> > might happen is with a database connector monitoring for tables
>>>> that
>>>> >> match
>>>> >> > a given regex every five minutes; if a table that matched that
>>>> regex
>>>> >> during
>>>> >> > the last scan got assigned to a task and then dropped, and then
>>>> another
>>>> >> > table that matched the regex got added before the next scan, the
>>>> >> connector
>>>> >> > would see the same number of tables, but the actual set would be
>>>> >> different.
>>>> >> > At this point, it would again be connector-dependent for whether
>>>> the
>>>> >> > already-assigned tables stayed assigned to the same tasks. Is
>>>> anyone
>>>> >> else
>>>> >> > reminded of the various consumer partition assignment strategies
>>>> at this
>>>> >> > point?
>>>> >> >
>>>> >> > A general comment I should make here (not necessarily for your
>>>> benefit
>>>> >> but
>>>> >> > for anyone following along) is that it's important to keep in mind
>>>> that
>>>> >> > "source partitions" in Kafka Connect aren't Kafka topic partitions
>>>> >> (well,
>>>> >> > unless your connector is designed to replicate data across Kafka
>>>> >> clusters
>>>> >> > like MirrorMaker 2). As a result, we have to rely on
>>>> developer-written
>>>> >> code
>>>> >> > to a greater extent to define what the source partitions for a
>>>> source
>>>> >> > connector are, and how to divvy up that work amongst tasks.
>>>> >> >
>>>> >> > Hope this helps! If you have any further questions please hit me
>>>> with
>>>> >> them;
>>>> >> > I doubt you'll be the only one wondering about these things.
>>>> >> >
>>>> >> > Cheers,
>>>> >> >
>>>> >> > Chris
>>>> >> >
>>>> >> > On Mon, Feb 22, 2021 at 5:30 PM Guozhang Wang <wangg...@gmail.com>
>>>> >> wrote:
>>>> >> >
>>>> >> > > Thanks Chris, yeah I think I agree with you that this does not
>>>> >> > necessarily
>>>> >> > > have to be in the scope of this KIP.
>>>> >> > >
>>>> >> > > My understanding was that the source partitions -> tasks are not
>>>> >> static
>>>> >> > but
>>>> >> > > dynamic, but they are only changed when either the number of
>>>> >> partitions
>>>> >> > > changed or "tasks.max" config changed (please correct me if I'm
>>>> >> wrong),
>>>> >> > so
>>>> >> > > what I'm thinking that we can try to detect if either of these
>>>> things
>>>> >> > > happens, and if they do not happen we can assume the mapping from
>>>> >> > > partitions -> tasks does not change --- of course this requires
>>>> some
>>>> >> > > extension on the API, aligned with what you said. I would like
>>>> to make
>>>> >> > sure
>>>> >> > > that my understanding here is correct :)
>>>> >> > >
>>>> >> > > Guozhang
>>>> >> > >
>>>> >> > >
>>>> >> > > On Mon, Feb 22, 2021 at 11:29 AM Chris Egerton <
>>>> chr...@confluent.io>
>>>> >> > > wrote:
>>>> >> > >
>>>> >> > > > Hi Guozhang,
>>>> >> > > >
>>>> >> > > > Thanks for taking a look, and for your suggestion!
>>>> >> > > >
>>>> >> > > > I think there is room for more intelligent fencing strategies,
>>>> but I
>>>> >> > > think
>>>> >> > > > that it'd have to be more nuanced than one based on
>>>> task->worker
>>>> >> > > > assignments. Connectors can arbitrarily reassign source
>>>> partitions
>>>> >> > (such
>>>> >> > > as
>>>> >> > > > database tables, or Kafka topic partitions) across tasks, so
>>>> even if
>>>> >> > the
>>>> >> > > > assignment of tasks across workers remains unchanged, the
>>>> >> assignment of
>>>> >> > > > source partitions across those workers might. Connect doesn't
>>>> do any
>>>> >> > > > inspection of task configurations at the moment, and without
>>>> >> expansions
>>>> >> > > to
>>>> >> > > > the Connector/Task API, it'd likely be impossible to get
>>>> information
>>>> >> > from
>>>> >> > > > tasks about their source partition assignments. With that in
>>>> mind, I
>>>> >> > > think
>>>> >> > > > we may want to leave the door open for more intelligent task
>>>> fencing
>>>> >> > but
>>>> >> > > > not include that level of optimization at this stage. Does that
>>>> >> sound
>>>> >> > > fair
>>>> >> > > > to you?
>>>> >> > > >
>>>> >> > > > There is one case that I've identified where we can cheaply
>>>> optimize
>>>> >> > > right
>>>> >> > > > now: single-task connectors, such as the Debezium CDC source
>>>> >> > connectors.
>>>> >> > > If
>>>> >> > > > a connector is configured at some point with a single task,
>>>> then
>>>> >> some
>>>> >> > > other
>>>> >> > > > part of its configuration is altered but the single-task aspect
>>>> >> > remains,
>>>> >> > > > the leader doesn't have to worry about fencing out the older
>>>> task as
>>>> >> > the
>>>> >> > > > new task's producer will do that automatically. In this case,
>>>> the
>>>> >> > leader
>>>> >> > > > can skip the producer fencing round and just write the new task
>>>> >> count
>>>> >> > > > record straight to the config topic. I've added this case to
>>>> the
>>>> >> KIP;
>>>> >> > if
>>>> >> > > it
>>>> >> > > > overcomplicates things I'm happy to remove it, but seeing as it
>>>> >> serves
>>>> >> > a
>>>> >> > > > real use case and comes fairly cheap, I figured it'd be best to
>>>> >> include
>>>> >> > > > now.
>>>> >> > > >
>>>> >> > > > Thanks again for your feedback; if you have other thoughts I'd
>>>> love
>>>> >> to
>>>> >> > > hear
>>>> >> > > > them!
>>>> >> > > >
>>>> >> > > > Cheers,
>>>> >> > > >
>>>> >> > > > Chris
>>>> >> > > >
>>>> >> > > > On Mon, Feb 22, 2021 at 1:57 PM Chris Egerton <
>>>> chr...@confluent.io>
>>>> >> > > wrote:
>>>> >> > > >
>>>> >> > > > > Hi Gwen,
>>>> >> > > > >
>>>> >> > > > > Thanks for the feedback!
>>>> >> > > > >
>>>> >> > > > > 0.
>>>> >> > > > > That's a great point; I've updated the motivation section
>>>> with
>>>> >> that
>>>> >> > > > > rationale.
>>>> >> > > > >
>>>> >> > > > > 1.
>>>> >> > > > > This enables safe "hard downgrades" of clusters where,
>>>> instead of
>>>> >> > just
>>>> >> > > > > disabling exactly-once support on each worker, each worker is
>>>> >> rolled
>>>> >> > > back
>>>> >> > > > > to an earlier version of the Connect framework that doesn't
>>>> >> support
>>>> >> > > > > per-connector offsets topics altogether. Those workers would
>>>> go
>>>> >> back
>>>> >> > to
>>>> >> > > > all
>>>> >> > > > > using a global offsets topic, and any offsets stored in
>>>> >> per-connector
>>>> >> > > > > topics would be lost to those workers. This would cause a
>>>> large
>>>> >> > number
>>>> >> > > of
>>>> >> > > > > duplicates to flood the downstream system. While technically
>>>> >> > > permissible
>>>> >> > > > > given that the user in this case will have knowingly
>>>> switched to a
>>>> >> > > > version
>>>> >> > > > > of the Connect framework that doesn't support exactly-once
>>>> source
>>>> >> > > > > connectors (and is therefore susceptible to duplicate
>>>> delivery of
>>>> >> > > > records),
>>>> >> > > > > the user experience in this case could be pretty bad. A
>>>> similar
>>>> >> > > situation
>>>> >> > > > > is if users switch back from per-connector offsets topics to
>>>> the
>>>> >> > global
>>>> >> > > > > offsets topic.
>>>> >> > > > > I've tried to make this more clear in the KIP by linking to
>>>> the
>>>> >> "Hard
>>>> >> > > > > downgrade" section from the proposed design, and by
>>>> expanding on
>>>> >> the
>>>> >> > > > > rationale provided for redundant global offset writes in the
>>>> "Hard
>>>> >> > > > > downgrade" section. Let me know if you think this could be
>>>> >> improved
>>>> >> > or
>>>> >> > > > > think a different approach is warranted.
>>>> >> > > > >
>>>> >> > > > > 2.
>>>> >> > > > > I think the biggest difference between Connect and Streams
>>>> comes
>>>> >> from
>>>> >> > > the
>>>> >> > > > > fact that Connect allows users to create connectors that
>>>> target
>>>> >> > > different
>>>> >> > > > > Kafka clusters on the same worker. This hasn't been a
>>>> problem in
>>>> >> the
>>>> >> > > past
>>>> >> > > > > because workers use two separate producers to write offset
>>>> data
>>>> >> and
>>>> >> > > > source
>>>> >> > > > > connector records, but in order to write offsets and records
>>>> in
>>>> >> the
>>>> >> > > same
>>>> >> > > > > transaction, it becomes necessary to use a single producer,
>>>> which
>>>> >> > also
>>>> >> > > > > requires that the internal offsets topic be hosted on the
>>>> same
>>>> >> Kafka
>>>> >> > > > > cluster that the connector is targeting.
>>>> >> > > > > This may sound like a niche use case but it was actually one
>>>> of
>>>> >> the
>>>> >> > > > > driving factors behind KIP-458 (
>>>> >> > > > >
>>>> >> > > >
>>>> >> > >
>>>> >> >
>>>> >>
>>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-458%3A+Connector+Client+Config+Override+Policy
>>>> >> > > > ),
>>>> >> > > > > and it's a feature that we rely on heavily today.
>>>> >> > > > > If there's too much going on in this KIP and we'd prefer to
>>>> drop
>>>> >> > > support
>>>> >> > > > > for running that type of setup with exactly-once source
>>>> connectors
>>>> >> > for
>>>> >> > > > now,
>>>> >> > > > > I can propose this in a separate KIP. I figured it'd be best
>>>> to
>>>> >> get
>>>> >> > > this
>>>> >> > > > > out of the way with the initial introduction of exactly-once
>>>> >> source
>>>> >> > > > support
>>>> >> > > > > in order to make adoption by existing Connect users as
>>>> seamless as
>>>> >> > > > > possible, and since we'd likely have to address this issue
>>>> before
>>>> >> > being
>>>> >> > > > > able to utilize the feature ourselves.
>>>> >> > > > > I switched around the ordering of the "motivation" section
>>>> for
>>>> >> > > > > per-connector offsets topics to put the biggest factor
>>>> first, and
>>>> >> > > called
>>>> >> > > > it
>>>> >> > > > > out as the major difference between Connect and Streams in
>>>> this
>>>> >> case.
>>>> >> > > > >
>>>> >> > > > > 3.
>>>> >> > > > > Fair enough, after giving it a little more thought I agree
>>>> that
>>>> >> > > allowing
>>>> >> > > > > users to shoot themselves in the foot is a bad idea here.
>>>> There's
>>>> >> > also
>>>> >> > > > some
>>>> >> > > > > precedent for handling this with the "enable.idempotence"
>>>> and "
>>>> >> > > > > transactional.id" producer properties; if you specify a
>>>> >> > transactional
>>>> >> > > ID
>>>> >> > > > > but don't specify a value for idempotence, the producer just
>>>> does
>>>> >> the
>>>> >> > > > right
>>>> >> > > > > thing for you by enabling idempotence and emitting a log
>>>> message
>>>> >> > > letting
>>>> >> > > > > you know that it's done so. I've adjusted the proposed
>>>> behavior to
>>>> >> > try
>>>> >> > > to
>>>> >> > > > > use a similar approach; let me know what you think.
>>>> >> > > > > There is the potential gap here where, sometime in the
>>>> future, a
>>>> >> > third
>>>> >> > > > > accepted value for the "isolation.level" property is added
>>>> to the
>>>> >> > > > consumer
>>>> >> > > > > API and users will be unable to use that new value for their
>>>> >> worker.
>>>> >> > > But
>>>> >> > > > > the likelihood of footgunning seems much greater than this
>>>> >> scenario,
>>>> >> > > and
>>>> >> > > > we
>>>> >> > > > > can always address expansions to the consumer API with
>>>> changes to
>>>> >> the
>>>> >> > > > > Connect framework as well if/when that becomes necessary.
>>>> >> > > > > I've also added a similar note to the source task's
>>>> transactional
>>>> >> ID
>>>> >> > > > > property; user overrides of it will also be disabled.
>>>> >> > > > >
>>>> >> > > > > 4.
>>>> >> > > > > Yeah, that's mostly correct. I tried to touch on this in the
>>>> >> > > "Motivation"
>>>> >> > > > > section with this bit:
>>>> >> > > > > > The Connect framework periodically writes source task
>>>> >> offsets
>>>> >> > to
>>>> >> > > an
>>>> >> > > > > internal Kafka topic at a configurable interval, once the
>>>> source
>>>> >> > > records
>>>> >> > > > > that they correspond to have been successfully sent to Kafka.
>>>> >> > > > > I've expanded on this in the "Offset (and record) writes"
>>>> section,
>>>> >> > and
>>>> >> > > > > I've tweaked the "Motivation" section a little bit to add a
>>>> link
>>>> >> to
>>>> >> > the
>>>> >> > > > > relevant config property and to make the language a little
>>>> more
>>>> >> > > accurate.
>>>> >> > > > >
>>>> >> > > > > 5.
>>>> >> > > > > This isn't quite as bad as stop-the-world; more like
>>>> >> > > stop-the-connector.
>>>> >> > > > > If a worker is running a dozen connectors and one of those
>>>> (that
>>>> >> > > happens
>>>> >> > > > to
>>>> >> > > > > be a source) is reconfigured, only the tasks for that
>>>> connector
>>>> >> will
>>>> >> > be
>>>> >> > > > > preemptively halted, and all other tasks and connectors will
>>>> >> continue
>>>> >> > > > > running. This is pretty close to the current behavior with
>>>> >> > incremental
>>>> >> > > > > rebalancing; the only difference is that, instead of waiting
>>>> for
>>>> >> the
>>>> >> > > > > rebalance to complete before halting the tasks for that
>>>> connector,
>>>> >> > the
>>>> >> > > > > worker will halt them in preparation for the rebalance. This
>>>> >> > increases
>>>> >> > > > the
>>>> >> > > > > time that the tasks will be down for, but is necessary if we
>>>> want
>>>> >> to
>>>> >> > > give
>>>> >> > > > > tasks time to gracefully shut down before being fenced out
>>>> by the
>>>> >> > > leader,
>>>> >> > > > > and shouldn't impact availability too much as it's only
>>>> triggered
>>>> >> on
>>>> >> > > > > connector reconfiguration and rebalances generally don't
>>>> take that
>>>> >> > long
>>>> >> > > > > with the new incremental protocol.
>>>> >> > > > >
>>>> >> > > > > 6.
>>>> >> > > > > Ah, thanks for the catch! That's a good point. I've adjusted
>>>> the
>>>> >> > > proposal
>>>> >> > > > > to treat the worker's transactional ID like the consumer's
>>>> >> isolation
>>>> >> > > > level
>>>> >> > > > > and the source task's transactional ID properties; can't be
>>>> >> modified
>>>> >> > by
>>>> >> > > > > users and, if an attempt is made to do so, will be discarded
>>>> after
>>>> >> > > > logging
>>>> >> > > > > a message to that effect. More footguns removed, hooray!
>>>> >> > > > >
>>>> >> > > > > Thanks again for taking a look; if you have any other
>>>> questions or
>>>> >> > > > > suggestions I'm all ears.
>>>> >> > > > >
>>>> >> > > > > Cheers,
>>>> >> > > > >
>>>> >> > > > > Chris
>>>> >> > > > >
>>>> >> > > > > p.s. - Guozhang--I saw your email; response to you coming
>>>> next :)
>>>> >> > > > >
>>>> >> > > > > On Sun, Feb 21, 2021 at 4:12 PM Guozhang Wang <
>>>> wangg...@gmail.com
>>>> >> >
>>>> >> > > > wrote:
>>>> >> > > > >
>>>> >> > > > >> Hello Chris,
>>>> >> > > > >>
>>>> >> > > > >> Thanks for the great write-up! I was mainly reviewing the
>>>> admin
>>>> >> > > > >> fenceProducers API of the KIP. I think it makes sense
>>>> overall.
>>>> >> I'm
>>>> >> > > just
>>>> >> > > > >> wondering if we can go one step further, that instead of
>>>> forcing
>>>> >> to
>>>> >> > > > fence
>>>> >> > > > >> out all producers of the previous generation, could we try
>>>> to
>>>> >> > achieve
>>>> >> > > > >> "partial rebalance" still by first generate the new
>>>> assignment,
>>>> >> and
>>>> >> > > then
>>>> >> > > > >> based on the new assignment only fence out producers
>>>> involved in
>>>> >> > tasks
>>>> >> > > > >> that
>>>> >> > > > >> are indeed migrated? Just a wild thought to bring up for
>>>> debate.
>>>> >> > > > >>
>>>> >> > > > >> Guozhang
>>>> >> > > > >>
>>>> >> > > > >>
>>>> >> > > > >>
>>>> >> > > > >> On Sat, Feb 20, 2021 at 10:20 PM Gwen Shapira <
>>>> g...@confluent.io
>>>> >> >
>>>> >> > > > wrote:
>>>> >> > > > >>
>>>> >> > > > >> > Hey Chris,
>>>> >> > > > >> >
>>>> >> > > > >> > Thank you for the proposal. Few questions / comments:
>>>> >> > > > >> >
>>>> >> > > > >> > 0. It may be worth pointing out in the motivation section
>>>> that
>>>> >> > > > >> > source-connector exactly once is more important than sink
>>>> >> > connector
>>>> >> > > > >> > exactly once, since many target systems will have unique
>>>> key
>>>> >> > > > >> > constraint mechanisms that will prevent duplicates. Kafka
>>>> does
>>>> >> not
>>>> >> > > > >> > have any such constraints, so without this KIP-618,
>>>> exactly
>>>> >> once
>>>> >> > > won't
>>>> >> > > > >> > be possible.
>>>> >> > > > >> > 1. I am not clear why we need the worker to async copy
>>>> offsets
>>>> >> > from
>>>> >> > > > >> > the connector-specific offset topic to a global offsets
>>>> topic
>>>> >> > > > >> > 2. While the reasoning you have for offset topic per
>>>> connector
>>>> >> > > appears
>>>> >> > > > >> > sound, it doesn't add up with the use of transactions in
>>>> >> > > KafkaStreams.
>>>> >> > > > >> > My understanding is that KafkaStreams uses shared offsets
>>>> topic
>>>> >> > with
>>>> >> > > > >> > all the other consumers, and (apparently) corrupting data
>>>> and
>>>> >> > delays
>>>> >> > > > >> > by other tenants is a non-issue. Perhaps you can comment
>>>> on how
>>>> >> > > > >> > Connect is different? In general much of the complexity
>>>> in the
>>>> >> KIP
>>>> >> > > is
>>>> >> > > > >> > related to the separate offset topic, and I'm wondering
>>>> if this
>>>> >> > can
>>>> >> > > be
>>>> >> > > > >> > avoided. The migration use-case is interesting, but not
>>>> >> related to
>>>> >> > > > >> > exactly-once and can be handled separately.
>>>> >> > > > >> > 3. Allowing users to override the isolation level for the
>>>> >> offset
>>>> >> > > > >> > reader, even when exactly-once is enabled, thereby
>>>> disabling
>>>> >> > > > >> > exactly-once in a non-obvious way. I get that connect
>>>> usually
>>>> >> > allows
>>>> >> > > > >> > users to shoot themselves in the foot, but are there any
>>>> actual
>>>> >> > > > >> > benefits for allowing it in this case? Maybe it is better
>>>> if we
>>>> >> > > don't?
>>>> >> > > > >> > I don't find the argument that we always did this to be
>>>> >> > particularly
>>>> >> > > > >> > compelling.
>>>> >> > > > >> > 4. It isn't stated explicitly, but it sounds like connect
>>>> or
>>>> >> > source
>>>> >> > > > >> > connectors already have some batching mechanism, and that
>>>> >> > > transaction
>>>> >> > > > >> > boundaries will match the batches (i.e. each batch will
>>>> be a
>>>> >> > > > >> > transaction?). If so, worth being explicit.
>>>> >> > > > >> > 5. "When a rebalance is triggered, before (re-)joining the
>>>> >> cluster
>>>> >> > > > >> > group, all workers will preemptively stop all tasks of all
>>>> >> source
>>>> >> > > > >> > connectors for which task configurations are present in
>>>> the
>>>> >> config
>>>> >> > > > >> > topic after the latest task count record" - how will this
>>>> play
>>>> >> > with
>>>> >> > > > >> > the incremental rebalances? isn't this exactly the
>>>> >> stop-the-world
>>>> >> > > > >> > rebalance we want to avoid?
>>>> >> > > > >> > 6. "the worker will instantiate a transactional producer
>>>> whose
>>>> >> > > > >> > transactional ID is, by default, the group ID of the
>>>> cluster
>>>> >> (but
>>>> >> > > may
>>>> >> > > > >> > be overwritten by users using the transactional.id worker
>>>> >> > > property)"
>>>> >> > > > -
>>>> >> > > > >> > If users change transactional.id property, zombie leaders
>>>> >> won't
>>>> >> > get
>>>> >> > > > >> > fenced (since they will have an older and different
>>>> >> transactional
>>>> >> > > id)
>>>> >> > > > >> >
>>>> >> > > > >> > Thanks,
>>>> >> > > > >> >
>>>> >> > > > >> > Gwen
>>>> >> > > > >> >
>>>> >> > > > >> > On Thu, May 21, 2020 at 11:21 PM Chris Egerton <
>>>> >> > chr...@confluent.io
>>>> >> > > >
>>>> >> > > > >> > wrote:
>>>> >> > > > >> > >
>>>> >> > > > >> > > Hi all,
>>>> >> > > > >> > >
>>>> >> > > > >> > > I know it's a busy time with the upcoming 2.6 release
>>>> and I
>>>> >> > don't
>>>> >> > > > >> expect
>>>> >> > > > >> > > this to get a lot of traction until that's done, but
>>>> I've
>>>> >> > > published
>>>> >> > > > a
>>>> >> > > > >> KIP
>>>> >> > > > >> > > for allowing atomic commit of offsets and records for
>>>> source
>>>> >> > > > >> connectors
>>>> >> > > > >> > and
>>>> >> > > > >> > > would appreciate your feedback:
>>>> >> > > > >> > >
>>>> >> > > > >> >
>>>> >> > > > >>
>>>> >> > > >
>>>> >> > >
>>>> >> >
>>>> >>
>>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-618%3A+Atomic+commit+of+source+connector+records+and+offsets
>>>> >> > > > >> > >
>>>> >> > > > >> > > This feature should make it possible to implement source
>>>> >> > > connectors
>>>> >> > > > >> with
>>>> >> > > > >> > > exactly-once delivery guarantees, and even allow a wide
>>>> >> range of
>>>> >> > > > >> existing
>>>> >> > > > >> > > source connectors to provide exactly-once delivery
>>>> guarantees
>>>> >> > with
>>>> >> > > > no
>>>> >> > > > >> > > changes required.
>>>> >> > > > >> > >
>>>> >> > > > >> > > Cheers,
>>>> >> > > > >> > >
>>>> >> > > > >> > > Chris
>>>> >> > > > >> >
>>>> >> > > > >> >
>>>> >> > > > >> >
>>>> >> > > > >> > --
>>>> >> > > > >> > Gwen Shapira
>>>> >> > > > >> > Engineering Manager | Confluent
>>>> >> > > > >> > 650.450.2760 | @gwenshap
>>>> >> > > > >> > Follow us: Twitter | blog
>>>> >> > > > >> >
>>>> >> > > > >>
>>>> >> > > > >>
>>>> >> > > > >> --
>>>> >> > > > >> -- Guozhang
>>>> >> > > > >>
>>>> >> > > > >
>>>> >> > > >
>>>> >> > >
>>>> >> > >
>>>> >> > > --
>>>> >> > > -- Guozhang
>>>> >> > >
>>>> >> >
>>>> >>
>>>> >>
>>>> >> --
>>>> >> -- Guozhang
>>>> >>
>>>> >
>>>>
>>>
