Hey All, Bumping this discussion thread again to see how the modified KIP looks like.
Thanks! Sagar. On Mon, May 29, 2023 at 8:12 PM Sagar <sagarmeansoc...@gmail.com> wrote: > Hi, > > Bumping this thread again for further reviews. > > Thanks! > Sagar. > > On Fri, May 12, 2023 at 3:38 PM Sagar <sagarmeansoc...@gmail.com> wrote: > >> Hi All, >> >> Thanks for the comments/reviews. I have updated the KIP >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-910%3A+Update+Source+offsets+for+Source+Connectors+without+producing+records >> with a newer approach which shelves the need for an explicit topic. >> >> Please review again and let me know what you think. >> >> Thanks! >> Sagar. >> >> >> On Mon, Apr 24, 2023 at 3:35 PM Yash Mayya <yash.ma...@gmail.com> wrote: >> >>> Hi Sagar, >>> >>> Thanks for the KIP! I have a few questions and comments: >>> >>> 1) I agree with Chris' point about the separation of a connector >>> heartbeat >>> mechanism and allowing source connectors to generate offsets without >>> producing data. What is the purpose of the heartbeat topic here and are >>> there any concrete use cases for downstream consumers on this topic? Why >>> can't we instead simply introduce a mechanism to retrieve a list of >>> source >>> partition / source offset pairs from the source tasks? >>> >>> 2) With the currently described mechanism, the new >>> "SourceTask::produceHeartbeatRecords" method returns a >>> "List<SourceRecord>" >>> - what happens with the topic in each of these source records? Chris >>> pointed this out above, but it doesn't seem to have been addressed? The >>> "SourceRecord" class also has a bunch of other fields which will be >>> irrelevant here (partition, key / value schema, key / value data, >>> timestamp, headers). In fact, it seems like only the source partition and >>> source offset are relevant here, so we should either introduce a new >>> abstraction or simply use a data structure like a mapping from source >>> partitions to source offsets (adds to the above point)? >>> >>> 3) I'm not sure I fully follow why the heartbeat timer / interval is >>> needed? What are the downsides of >>> calling "SourceTask::produceHeartbeatRecords" in every execution loop >>> (similar to the existing "SourceTask::poll" method)? Is this only to >>> prevent the generation of a lot of offset records? Since Connect's >>> offsets >>> topics are log compacted (and source partitions are used as keys for each >>> source offset), I'm not sure if such concerns are valid and such a >>> heartbeat timer / interval mechanism is required? >>> >>> 4) The first couple of rejected alternatives state that the use of a null >>> topic / key / value are preferably avoided - but the current proposal >>> would >>> also likely require connectors to use such workarounds (null topic when >>> the >>> heartbeat topic is configured at a worker level and always for the key / >>> value)? >>> >>> 5) The third rejected alternative talks about subclassing the >>> "SourceRecord" class - this presumably means allowing connectors to pass >>> special offset only records via the existing poll mechanism? Why was this >>> considered a more invasive option? Was it because of the backward >>> compatibility issues that would be introduced for plugins using the new >>> public API class that still need to be deployed onto older Connect >>> workers? >>> >>> Thanks, >>> Yash >>> >>> On Fri, Apr 14, 2023 at 6:45 PM Sagar <sagarmeansoc...@gmail.com> wrote: >>> >>> > One thing I forgot to mention in my previous email was that the reason >>> I >>> > chose to include the opt-in behaviour via configs was that the users >>> of the >>> > connector know their workload patterns. If the workload is such that >>> the >>> > connector would receive regular valid updates then there’s ideally no >>> need >>> > for moving offsets since it would update automatically. >>> > >>> > This way they aren’t forced to use this feature and can use it only >>> when >>> > the workload is expected to be batchy or not frequent. >>> > >>> > Thanks! >>> > Sagar. >>> > >>> > >>> > On Fri, 14 Apr 2023 at 5:32 PM, Sagar <sagarmeansoc...@gmail.com> >>> wrote: >>> > >>> > > Hi Chris, >>> > > >>> > > Thanks for following up on the response. Sharing my thoughts further: >>> > > >>> > > If we want to add support for connectors to emit offsets without >>> > >> accompanying source records, we could (and IMO should) do that >>> without >>> > >> requiring users to manually enable that feature by adjusting worker >>> or >>> > >> connector configurations. >>> > > >>> > > >>> > > With the current KIP design, I have tried to implement this in an >>> opt-in >>> > > manner via configs. I guess what you are trying to say is that this >>> > doesn't >>> > > need a config of it's own and instead could be part of the poll -> >>> > > transform etc -> produce -> commit cycle. That way, the users don't >>> need >>> > to >>> > > set any config and if the connector supports moving offsets w/o >>> producing >>> > > SourceRecords, it should happen automatically. Is that correct? If >>> that >>> > > is the concern, then I can think of not exposing a config and try to >>> make >>> > > this process automatically. That should ease the load on connector >>> users, >>> > > but your point about cognitive load on Connector developers, I am >>> still >>> > not >>> > > sure how to address that. The offsets are privy to a connector and >>> the >>> > > framework at best can provide hooks to the tasks to update their >>> offsets. >>> > > Connector developers would still have to consider all cases before >>> > updating >>> > > offsets. And if I ignore the heartbeat topic and heartbeat interval >>> ms >>> > > configs, then what the KIP proposes currently isn't much different in >>> > that >>> > > regard. Just that it produces a List of SourceRecord which can be >>> changed >>> > > to a Map of SourcePartition and their offsets if you think that would >>> > > simplify things. Are there other cases in your mind which need >>> > addressing? >>> > > >>> > > Here's my take on the usecases: >>> > > >>> > > 1. Regarding the example about SMTs with Object Storage based >>> > > connectors, it was one of the scenarios identified. We have some >>> > connectors >>> > > that rely on the offsets topic to check if the next batch of files >>> > should >>> > > be processed and because of filtering of the last record from the >>> > files, >>> > > the eof supposedly is never reached and the connector can't >>> commit >>> > offsets >>> > > for that source partition(file). If there was a mechanism to >>> update >>> > offsets >>> > > for such a source file, then with some moderately complex state >>> > tracking, >>> > > the connector can mark that file as processed and proceed. >>> > > 2. There's another use case with the same class of connectors >>> where if >>> > > a file is malformed, then the connector couldn't produce any >>> offsets >>> > > because the file couldn't get processed completely. To handle such >>> > cases, >>> > > the connector developers have introduced a dev/null sort of topic >>> > where >>> > > they produce a record to this corrupted file topic and move the >>> offset >>> > > somehow. This topic ideally isn't needed and with a mechanism to >>> > update >>> > > offsets would have helped in this case as well. >>> > > 3. Coming to CDC based connectors, >>> > > 1. We had a similar issue with Oracle CDC source connector and >>> > > needed to employ the same heartbeat mechanism to get around it. >>> > > 2. MongoDB CDC source Connector has employed the same >>> heartbeat >>> > > mechanism Check `heartbeat.interval.ms` here ( >>> > > >>> > >>> https://www.mongodb.com/docs/kafka-connector/current/source-connector/configuration-properties/error-handling/ >>> > > ). >>> > > 3. Another CDC connector for ScyllaDB employs a similar >>> mechanism. >>> > > >>> > >>> https://github.com/scylladb/scylla-cdc-source-connector/search?q=heartbeat >>> > > 4. For CDC based connectors, you could argue that these >>> connectors >>> > > have been able to solve this error then why do we need >>> framework >>> > level >>> > > support. But the point I am trying to make is that this >>> limitation >>> > from the >>> > > framework is forcing CDC connector developers to implement >>> > per-connector >>> > > solutions/hacks(at times). And there could always be more CDC >>> > connectors in >>> > > the pipeline forcing them to take a similar route as well. >>> > > 4. There's also a case at times with CDC source connectors which >>> are >>> > > REST Api / Web Service based(Zendesk Source Connector for >>> example) . >>> > These >>> > > connectors typically use timestamps from the responses as >>> offsets. If >>> > > there's a long period of inactivity wherein the API invocations >>> don't >>> > > return any data, then the offsets won't move and the connector >>> would >>> > keep >>> > > using the same timestamp that it received from the last non-empty >>> > response. >>> > > If this period of inactivity keeps growing, and the API imposes >>> any >>> > limits >>> > > on how far back we can go in terms of window start, then this >>> could >>> > > potentially be a problem. In this case even though the connector >>> was >>> > caught >>> > > up with all the responses, it may need to snapshot again. In this >>> case >>> > > updating offsets can easily help since all the connector needs to >>> do >>> > is to >>> > > move the timestamp which would move the offset inherently. >>> > > >>> > > I still believe that this is something the framework should support >>> OOB >>> > > irrespective of whether the connectors have been able to get around >>> this >>> > > restriction or not. >>> > > >>> > > Lastly, about your comments here: >>> > > >>> > > I'm also not sure that it's worth preserving the current behavior >>> that >>> > >> offsets for records that have been filtered out via SMT are not >>> > committed. >>> > > >>> > > >>> > > Let me know if we need a separate JIRA to track this? This somehow >>> didn't >>> > > look related to this discussion. >>> > > >>> > > Thanks! >>> > > Sagar. >>> > > >>> > > >>> > > On Wed, Apr 12, 2023 at 9:34 PM Chris Egerton >>> <chr...@aiven.io.invalid> >>> > > wrote: >>> > > >>> > >> Hi Sagar, >>> > >> >>> > >> I'm sorry, I'm still not convinced that this design solves the >>> > problem(s) >>> > >> it sets out to solve in the best way possible. I tried to highlight >>> this >>> > >> in >>> > >> my last email: >>> > >> >>> > >> > In general, it seems like we're trying to solve two completely >>> > different >>> > >> problems with this single KIP: adding framework-level support for >>> > emitting >>> > >> heartbeat records for source connectors, and allowing source >>> connectors >>> > to >>> > >> emit offsets without also emitting source records. I don't mind >>> > addressing >>> > >> the two at the same time if the result is elegant and doesn't >>> compromise >>> > >> on >>> > >> the solution for either problem, but that doesn't seem to be the >>> case >>> > >> here. >>> > >> Of the two problems, could we describe one as the primary and one >>> as the >>> > >> secondary? If so, we might consider dropping the secondary problem >>> from >>> > >> this KIP and addressing it separately. >>> > >> >>> > >> If we wanted to add support for heartbeat records, we could (and IMO >>> > >> should) do that without requiring connectors to implement any new >>> > methods >>> > >> and only require adjustments to worker or connector configurations >>> by >>> > >> users >>> > >> in order to enable that feature. >>> > >> >>> > >> If we want to add support for connectors to emit offsets without >>> > >> accompanying source records, we could (and IMO should) do that >>> without >>> > >> requiring users to manually enable that feature by adjusting worker >>> or >>> > >> connector configurations. >>> > >> >>> > >> >>> > >> I'm also not sure that it's worth preserving the current behavior >>> that >>> > >> offsets for records that have been filtered out via SMT are not >>> > committed. >>> > >> I can't think of a case where this would be useful and there are >>> > obviously >>> > >> plenty where it isn't. There's also a slight discrepancy in how >>> these >>> > >> kinds >>> > >> of records are treated by the Connect runtime now; if a record is >>> > dropped >>> > >> because of an SMT, then its offset isn't committed, but if it's >>> dropped >>> > >> because exactly-once support is enabled and the connector chose to >>> abort >>> > >> the batch containing the record, then its offset is still committed. >>> > After >>> > >> thinking carefully about the aborted transaction behavior, we >>> realized >>> > >> that >>> > >> it was fine to commit the offsets for those records, and I believe >>> that >>> > >> the >>> > >> same logic can be applied to any record that we're done trying to >>> send >>> > to >>> > >> Kafka (regardless of whether it was sent correctly, dropped due to >>> > >> producer >>> > >> error, filtered via SMT, etc.). >>> > >> >>> > >> I also find the file-based source connector example a little >>> confusing. >>> > >> What about that kind of connector causes the offset for the last >>> record >>> > of >>> > >> a file to be treated differently? Is there anything different about >>> > >> filtering that record via SMT vs. dropping it altogether because of >>> an >>> > >> asynchronous producer error with "errors.tolerance" set to "all"? >>> And >>> > >> finally, how would such a connector use the design proposed here? >>> > >> >>> > >> Finally, I don't disagree that if there are other legitimate use >>> cases >>> > >> that >>> > >> would be helped by addressing KAFKA-3821, we should try to solve >>> that >>> > >> issue >>> > >> in the Kafka Connect framework instead of requiring individual >>> > connectors >>> > >> to implement their own solutions. But the cognitive load added by >>> the >>> > >> design proposed here, for connector developers and Connect cluster >>> > >> administrators alike, costs too much to justify by pointing to an >>> > >> already-solved problem encountered by a single group of connectors >>> > (i.e., >>> > >> Debezium). This is why I think it's crucial that we identify >>> realistic >>> > >> cases where this feature would actually be useful, and right now, I >>> > don't >>> > >> think any have been provided (at least, not ones that have already >>> been >>> > >> addressed or could be addressed with much simpler changes). >>> > >> >>> > >> Cheers, >>> > >> >>> > >> Chris >>> > >> >>> > >> On Tue, Apr 11, 2023 at 7:30 AM Sagar <sagarmeansoc...@gmail.com> >>> > wrote: >>> > >> >>> > >> > Hi Chris, >>> > >> > >>> > >> > Thanks for your detailed feedback! >>> > >> > >>> > >> > nits: I have taken care of them now. Thanks for pointing those >>> out. >>> > >> > >>> > >> > non-nits: >>> > >> > >>> > >> > 6) It seems (based on both the KIP and discussion on KAFKA-3821) >>> that >>> > >> the >>> > >> > > only use case for being able to emit offsets without also >>> emitting >>> > >> source >>> > >> > > records that's been identified so far is for CDC source >>> connectors >>> > >> like >>> > >> > > Debezium. >>> > >> > >>> > >> > >>> > >> > I am aware of atleast one more case where the non production of >>> > offsets >>> > >> > (due to non production of records ) leads to the failure of >>> connectors >>> > >> when >>> > >> > the source purges the records of interest. This happens in File >>> based >>> > >> > source connectors (like s3/blob storage ) in which if the last >>> record >>> > >> from >>> > >> > a file is fiterterd due to an SMT, then that particular file is >>> never >>> > >> > committed to the source partition and eventually when the file is >>> > >> deleted >>> > >> > from the source and the connector is restarted due to some >>> reason, it >>> > >> > fails. >>> > >> > Moreover, I feel the reason this support should be there in the >>> Kafka >>> > >> > Connect framework is because this is a restriction of the >>> framework >>> > and >>> > >> > today the framework provides no support for getting around this >>> > >> limitation. >>> > >> > Every connector has it's own way of handling offsets and having >>> each >>> > >> > connector handle this restriction in its own way can make it >>> complex. >>> > >> > Whether we choose to do it the way this KIP prescribes or any >>> other >>> > way >>> > >> is >>> > >> > up for debate but IMHO, the framework should provide a way of >>> > >> > getting around this limitation. >>> > >> > >>> > >> > 7. If a task produces heartbeat records and source records that >>> use >>> > the >>> > >> > > same source partition, which offset will ultimately be >>> committed? >>> > >> > >>> > >> > >>> > >> > The idea is to add the records returned by the >>> > `produceHeartbeatRecords` >>> > >> > to the same `toSend` list within >>> `AbstractWorkerSourceTask#execute`. >>> > >> The >>> > >> > `produceHeartbeatRecords` would be invoked before we make the >>> `poll` >>> > >> call. >>> > >> > Hence, the offsets committed would be in the same order in which >>> they >>> > >> would >>> > >> > be written. Note that, the onus is on the Connector >>> implementation to >>> > >> not >>> > >> > return records which can lead to data loss or data going out of >>> order. >>> > >> The >>> > >> > framework would just commit based on whatever is supplied. Also, >>> > AFAIK, >>> > >> 2 >>> > >> > `normal` source records can also produce the same source >>> partitions >>> > and >>> > >> > they are committed in the order in which they are written. >>> > >> > >>> > >> > 8. The SourceTask::produceHeartbeatRecords method returns a >>> > >> > > List<SourceRecord>, and users can control the heartbeat topic >>> for a >>> > >> > > connector via the (connector- or worker-level) >>> > >> "heartbeat.records.topic" >>> > >> > > property. Since every constructor for the SourceRecord class [2] >>> > >> > requires a >>> > >> > > topic to be supplied, what will happen to that topic? Will it be >>> > >> ignored? >>> > >> > > If so, I think we should look for a cleaner solution. >>> > >> > >>> > >> > >>> > >> > Sorry, I couldn't quite follow which topic will be ignored in this >>> > case. >>> > >> > >>> > >> > 9. A large concern raised in the discussion for KAFKA-3821 was the >>> > >> allowing >>> > >> > > connectors to control the ordering of these special >>> "offsets-only" >>> > >> > > emissions and the regular source records returned from >>> > >> SourceTask::poll. >>> > >> > > Are we choosing to ignore that concern? If so, can you add this >>> to >>> > the >>> > >> > > rejected alternatives section along with a rationale? >>> > >> > >>> > >> > >>> > >> > One thing to note is that the for every connector, the condition >>> to >>> > emit >>> > >> > the heartbeat record is totally up to the connector, For example, >>> for >>> > a >>> > >> > connector which is tracking transactions for an ordered log, if >>> there >>> > >> are >>> > >> > open transactions, it might not need to emit heartbeat records >>> when >>> > the >>> > >> > timer expires while for file based connectors, if the same file is >>> > being >>> > >> > processed again and again due to an SMT or some other reasons, >>> then it >>> > >> can >>> > >> > choose to emit that partition. The uber point here is that every >>> > >> connector >>> > >> > has it's own requirements and the framework can't really make an >>> > >> assumption >>> > >> > about it. What the KIP is trying to do is to provide a mechanism >>> to >>> > the >>> > >> > connector to commit new offsets. With this approach, as far as I >>> can >>> > >> think >>> > >> > so far, there doesn't seem to be a case of out of order >>> processing. If >>> > >> you >>> > >> > have other concerns/thoughts I would be happy to know them. >>> > >> > >>> > >> > 10. If, sometime in the future, we wanted to add framework-level >>> > support >>> > >> > > for sending heartbeat records that doesn't require connectors to >>> > >> > implement >>> > >> > > any new APIs... >>> > >> > >>> > >> > >>> > >> > The main purpose of producing heartbeat records is to be able to >>> emit >>> > >> > offsets w/o any new records. We are using heartbeat records to >>> solve >>> > the >>> > >> > primary concern of offsets getting stalled. The reason to do that >>> was >>> > >> once >>> > >> > we get SourceRecords, then the rest of the code is already in >>> place to >>> > >> > write it to a topic of interest and commit offsets and that >>> seemed the >>> > >> most >>> > >> > non invasive in terms of framework level changes. If in the >>> future we >>> > >> want >>> > >> > to do a framework-only heartbeat record support, then this would >>> > create >>> > >> > confusion as you pointed out. Do you think the choice of the name >>> > >> heartbeat >>> > >> > records is creating confusion in this case? Maybe we can call >>> these >>> > >> special >>> > >> > records something else (not sure what at this point) which would >>> then >>> > >> > decouple the 2 logically and implementation wise as well? >>> > >> > >>> > >> > Thanks! >>> > >> > Sagar. >>> > >> > >>> > >> > On Tue, Mar 28, 2023 at 8:28 PM Chris Egerton >>> <chr...@aiven.io.invalid >>> > > >>> > >> > wrote: >>> > >> > >>> > >> > > Hi Sagar, >>> > >> > > >>> > >> > > Thanks for the KIP! I have some thoughts. >>> > >> > > >>> > >> > > Nits: >>> > >> > > >>> > >> > > 1. Shouldn't KAFKA-3821 [1] be linked as the Jira ticket on the >>> KIP? >>> > >> Or >>> > >> > is >>> > >> > > there a different ticket that should be associated with it? >>> > >> > > 2. The current state is listed as "Draft". Considering it's been >>> > >> brought >>> > >> > up >>> > >> > > for discussion, maybe the KIP should be updated to "Discussion"? >>> > >> > > 3. Can you add a link for the discussion thread to the KIP? >>> > >> > > 4. The KIP states that "In this process, offsets are written at >>> > >> regular >>> > >> > > intervals(driven by `offset.flush.interval.ms`)". This isn't >>> > strictly >>> > >> > > accurate since, when exactly-once support is enabled, offset >>> commits >>> > >> can >>> > >> > > also be performed for each record batch (which is the default) >>> or >>> > when >>> > >> > > explicitly requested by the task instance (if the connector >>> > implements >>> > >> > the >>> > >> > > API to define its own transactions and the user has configured >>> it to >>> > >> do >>> > >> > > so). Maybe better to just say "Offsets are written >>> periodically"? >>> > >> > > 5. The description for the (per-connector) >>> "heartbeat.records.topic >>> > " >>> > >> > > property states that it is "Only applicable in distributed >>> mode; in >>> > >> > > standalone mode, setting this property will have no effect". Is >>> this >>> > >> > > correct? >>> > >> > > >>> > >> > > Non-nits: >>> > >> > > >>> > >> > > 6. It seems (based on both the KIP and discussion on KAFKA-3821) >>> > that >>> > >> the >>> > >> > > only use case for being able to emit offsets without also >>> emitting >>> > >> source >>> > >> > > records that's been identified so far is for CDC source >>> connectors >>> > >> like >>> > >> > > Debezium. But Debezium already has support for this exact >>> feature >>> > >> > (emitting >>> > >> > > heartbeat records that include offsets that cannot be associated >>> > with >>> > >> > > other, "regular" source records). Why should we add this >>> feature to >>> > >> Kafka >>> > >> > > Connect when the problem it addresses is already solved in the >>> set >>> > >> > > connectors that (it seems) would have any need for it, and the >>> size >>> > of >>> > >> > that >>> > >> > > set is extremely small? If there are other practical use cases >>> for >>> > >> > > connectors that would benefit from this feature, please let me >>> know. >>> > >> > > >>> > >> > > 7. If a task produces heartbeat records and source records that >>> use >>> > >> the >>> > >> > > same source partition, which offset will ultimately be >>> committed? >>> > >> > > >>> > >> > > 8. The SourceTask::produceHeartbeatRecords method returns a >>> > >> > > List<SourceRecord>, and users can control the heartbeat topic >>> for a >>> > >> > > connector via the (connector- or worker-level) >>> > >> "heartbeat.records.topic" >>> > >> > > property. Since every constructor for the SourceRecord class [2] >>> > >> > requires a >>> > >> > > topic to be supplied, what will happen to that topic? Will it be >>> > >> ignored? >>> > >> > > If so, I think we should look for a cleaner solution. >>> > >> > > >>> > >> > > 9. A large concern raised in the discussion for KAFKA-3821 was >>> the >>> > >> > allowing >>> > >> > > connectors to control the ordering of these special >>> "offsets-only" >>> > >> > > emissions and the regular source records returned from >>> > >> SourceTask::poll. >>> > >> > > Are we choosing to ignore that concern? If so, can you add this >>> to >>> > the >>> > >> > > rejected alternatives section along with a rationale? >>> > >> > > >>> > >> > > 10. If, sometime in the future, we wanted to add framework-level >>> > >> support >>> > >> > > for sending heartbeat records that doesn't require connectors to >>> > >> > implement >>> > >> > > any new APIs (e.g., SourceTask::produceHeartbeatRecords), a lot >>> of >>> > >> this >>> > >> > > would paint us into a corner design-wise. We'd have to think >>> > carefully >>> > >> > > about which property names would be used, how to account for >>> > >> connectors >>> > >> > > that have already implemented the >>> > SourceTask::produceHeartbeatRecords >>> > >> > > method, etc. In general, it seems like we're trying to solve two >>> > >> > completely >>> > >> > > different problems with this single KIP: adding framework-level >>> > >> support >>> > >> > for >>> > >> > > emitting heartbeat records for source connectors, and allowing >>> > source >>> > >> > > connectors to emit offsets without also emitting source >>> records. I >>> > >> don't >>> > >> > > mind addressing the two at the same time if the result is >>> elegant >>> > and >>> > >> > > doesn't compromise on the solution for either problem, but that >>> > >> doesn't >>> > >> > > seem to be the case here. Of the two problems, could we >>> describe one >>> > >> as >>> > >> > the >>> > >> > > primary and one as the secondary? If so, we might consider >>> dropping >>> > >> the >>> > >> > > secondary problm from this KIP and addressing it separately. >>> > >> > > >>> > >> > > [1] - https://issues.apache.org/jira/browse/KAFKA-3821 >>> > >> > > [2] - >>> > >> > > >>> > >> > > >>> > >> > >>> > >> >>> > >>> https://kafka.apache.org/34/javadoc/org/apache/kafka/connect/source/SourceRecord.html >>> > >> > > >>> > >> > > Cheers, >>> > >> > > >>> > >> > > Chris >>> > >> > > >>> > >> > > On Sat, Mar 25, 2023 at 11:18 PM Sagar < >>> sagarmeansoc...@gmail.com> >>> > >> > wrote: >>> > >> > > >>> > >> > > > Hi John, >>> > >> > > > >>> > >> > > > Thanks for taking. look at the KIP! >>> > >> > > > >>> > >> > > > The point about stream time not advancing in case of >>> infrequent >>> > >> updates >>> > >> > > is >>> > >> > > > an interesting one. I can imagine if the upstream producer to >>> a >>> > >> Kafka >>> > >> > > > Streams application is a Source Connector which isn't sending >>> > >> records >>> > >> > > > frequently(due to the nature of the data ingestion for >>> example), >>> > >> then >>> > >> > the >>> > >> > > > downstream stream processing can land into the issues you >>> > described >>> > >> > > above. >>> > >> > > > >>> > >> > > > Which also brings me to the second point you made about how >>> this >>> > >> would >>> > >> > be >>> > >> > > > used by downstream consumers. IIUC, you are referring to the >>> > >> consumers >>> > >> > of >>> > >> > > > the newly added topic i.e the heartbeat topic. In my mind, the >>> > >> > heartbeat >>> > >> > > > topic is an internal topic (similar to offsets/config/status >>> topic >>> > >> in >>> > >> > > > connect), the main purpose of which is to trick the framework >>> to >>> > >> > produce >>> > >> > > > records to the offsets topic and advance the offsets. Since >>> every >>> > >> > > connector >>> > >> > > > could have a different definition of offsets(LSN, BinLogID >>> etc for >>> > >> > > > example), that logic to determine what the heartbeat records >>> > should >>> > >> be >>> > >> > > > would have to reside in the actual connector. >>> > >> > > > >>> > >> > > > Now that I think of it, it could very well be consumed by >>> > downstream >>> > >> > > > consumers/ Streams or Flink Applications and be further used >>> for >>> > >> some >>> > >> > > > decision making. A very crude example could be let's say if >>> the >>> > >> > heartbeat >>> > >> > > > records sent to the new heartbeat topic include timestamps, >>> then >>> > the >>> > >> > > > downstream streams application can use that timestamp to >>> close any >>> > >> time >>> > >> > > > windows. Having said that, it still appears to me that it's >>> > outside >>> > >> the >>> > >> > > > scope of the Connect framework and is something which is >>> difficult >>> > >> to >>> > >> > > > generalise because of the variety of Sources and the >>> definitions >>> > of >>> > >> > > > offsets. >>> > >> > > > >>> > >> > > > But, I would still be more than happy to add this example if >>> you >>> > >> think >>> > >> > it >>> > >> > > > can be useful in getting a better understanding of the idea >>> and >>> > also >>> > >> > its >>> > >> > > > utility beyond connect. Please let me know! >>> > >> > > > >>> > >> > > > Thanks! >>> > >> > > > Sagar. >>> > >> > > > >>> > >> > > > >>> > >> > > > On Fri, Mar 24, 2023 at 7:22 PM John Roesler < >>> vvcep...@apache.org >>> > > >>> > >> > > wrote: >>> > >> > > > >>> > >> > > > > Thanks for the KIP, Sagar! >>> > >> > > > > >>> > >> > > > > At first glance, this seems like a very useful feature. >>> > >> > > > > >>> > >> > > > > A common pain point in Streams is when upstream producers >>> don't >>> > >> send >>> > >> > > > > regular updates and stream time cannot advance. This causes >>> > >> > > > > stream-time-driven operations to appear to hang, like time >>> > windows >>> > >> > not >>> > >> > > > > closing, suppressions not firing, etc. >>> > >> > > > > >>> > >> > > > > From your KIP, I have a good idea of how the feature would >>> be >>> > >> > > integrated >>> > >> > > > > into connect, and it sounds good to me. I don't quite see >>> how >>> > >> > > downstream >>> > >> > > > > clients, such as a downstream Streams or Flink application, >>> or >>> > >> users >>> > >> > of >>> > >> > > > the >>> > >> > > > > Consumer would make use of this feature. Could you add some >>> > >> examples >>> > >> > of >>> > >> > > > > that nature? >>> > >> > > > > >>> > >> > > > > Thank you, >>> > >> > > > > -John >>> > >> > > > > >>> > >> > > > > On Fri, Mar 24, 2023, at 05:23, Sagar wrote: >>> > >> > > > > > Hi All, >>> > >> > > > > > >>> > >> > > > > > Bumping the thread again. >>> > >> > > > > > >>> > >> > > > > > Sagar. >>> > >> > > > > > >>> > >> > > > > > >>> > >> > > > > > On Fri, Mar 10, 2023 at 4:42 PM Sagar < >>> > >> sagarmeansoc...@gmail.com> >>> > >> > > > wrote: >>> > >> > > > > > >>> > >> > > > > >> Hi All, >>> > >> > > > > >> >>> > >> > > > > >> Bumping this discussion thread again. >>> > >> > > > > >> >>> > >> > > > > >> Thanks! >>> > >> > > > > >> Sagar. >>> > >> > > > > >> >>> > >> > > > > >> On Thu, Mar 2, 2023 at 3:44 PM Sagar < >>> > >> sagarmeansoc...@gmail.com> >>> > >> > > > wrote: >>> > >> > > > > >> >>> > >> > > > > >>> Hi All, >>> > >> > > > > >>> >>> > >> > > > > >>> I wanted to create a discussion thread for KIP-910: >>> > >> > > > > >>> >>> > >> > > > > >>> >>> > >> > > > > >>> >>> > >> > > > > >>> > >> > > > >>> > >> > > >>> > >> > >>> > >> >>> > >>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-910%3A+Update+Source+offsets+for+Source+Connectors+without+producing+records >>> > >> > > > > >>> >>> > >> > > > > >>> Thanks! >>> > >> > > > > >>> Sagar. >>> > >> > > > > >>> >>> > >> > > > > >> >>> > >> > > > > >>> > >> > > > >>> > >> > > >>> > >> > >>> > >> >>> > > >>> > >>> >>
