On Fri, Jan 29, 2016 at 7:06 AM, Randall Hauch <rha...@gmail.com> wrote:
> On January 28, 2016 at 7:07:02 PM, Ewen Cheslack-Postava ( > e...@confluent.io) wrote: > > Randall, > > Great question. Ideally you wouldn't need this type of state since it > should really be available in the source system. In your case, it might > actually make sense to be able to grab that information from the DB itself, > > although that will also have issues if, for example, there have been > multiple schema changes and you can no longer get a previous schema from > the current state of the tables. > > I agree that ideally connectors would be stateless, or at least have no > need for maintaining state across restarts. Unfortunately, that’s not > always possible. > > Reading the log but using the current schema does pose a problem if/when > the schema has evolved since the point in the log that we’re currently > reading. This is far more of an issue if you’re playing catch up and > there’s been non-compatible schema changes. > > Case in point: when MySQL inserts/updates/removes a row from a table, it > writes an event in the log that includes (a) a table identifier and (b) the > row values in column-order. There is no other information. Column renames > might be okay, but adding or removing columns will likely result in > mismatching the row values to the appropriate columns. > > Luckily, MySQL includes the DDL statements in the log, so my connector > parses these as part of processing and builds up the schema state as it > goes along. This works beautifully, with the only issue being how to > persist and recover this after restarts. > Yeah, this is a common complaint about the MySQL binlog. I know James mentioned this as well. It's a bit crazy that you need a full parser for the DDL to make this work :/ > > The offset storage is probably pretty close to what you're looking for, > although we obviously structure that very narrowly. Adding in some sort of > > other state store is an interesting idea, though I'd be curious how many > other systems encounter similar challenges. I think one way to do this > without huge changes and in a way that properly handles offset commits > would be to expose a small API for setting local state and have Connect > store that state right in the same topic (and message) as offsets. To > handle offset commits and reviving tasks that hit a fault, we would just > grab the current state as part of the process of committing offsets. Then > offsets would just be a special case of that more general state. > > However, I'm also wary of doing something like this. Right now every worker > > has to consume the entire offsets topic. This works fine because offsets, > while capable of being pretty complex, are generally pretty small such that > > there's no concern having to tail it on all workers (and no concern for the > > load on brokers leading those partitions). Once you provide a generic state > > storage mechanism without clear constraints on how it should be used, > someone will come along and abuse it. Also, with offsets it is very clear > (to the connector developer) which task should write to which keys (where > the key here is the partition of the source partitioned stream). With > support for arbitrary state, ownership of different subsets of the key > space is very unclear. I think you may not have that particular problem > because you probably only have 1 partition anyway since you are reading a > transaction log. > > In any case, you're right that this doesn't exist today. There is one very > > hacky way to get around this, which is to store that schema information in > > your "offsets". This may not be *too* bad -- it'll increase the size of > offset data, but probably doesn't affect much else. The data size may not > be that bad as long as offsets aren't committed too frequently. In terms of > > performance, I'm assuming these schema changes are relatively rare, and you > > can just include the same schema object in every offset you create during > the periods between schema changes so you (and the GC) are probably only > doing a negligible amount of extra work. > > > Hmm, it sound like hammering accumulated state into the offsets could be > pretty problematic and potentially risky, especially if the state has very > different size and frequency characteristics than the offsets. > Yes, as I siad, it is a hack. I mainly mentioned it here because I think for the amount of metadata you need to carry through, it could be a workable solution. > > Re: creating a consumer, Connect doesn't provide any utilities to do that > since the goal is to handle everything Kafka-related for the connector > developer so they can just focus on getting the data from the other system! > > We could consider exposing some of the worker config though, which I > imagine is all you really need -- it'd just be convenient to have the > connection info for the Kafka brokers. > > Having a way to get the worker config would be awesome, and IMO it a nice > minimalistic approach. If you think this is a good idea, I can log a JIRA > and take it to the dev list. I’m willing to work on it, too. > I think this is not going to be commonly used, but I think it'd be fine to expose it. > I’m starting to think that storing state on a separate dedicated topic is > the best option, at least for me. First, connector tasks can easily record > their state by simply adding more SourceRecord instances during polling. > Second, that information might be useful for downstream consumers. And > third, recording state this way requires no changes to the current Kafka > Connect. I’d hate to add a feature to Kafka Connect that is not useful to > others. > > Recovery would require consuming this topic upon restart. If the state > were incremental, then a restart might require consuming the entire topic. > If the state were snapshots, then simply reading the last entry might be > sufficient. There’s also the option of doing primarily incremental with > periodic snapshots. > > For my connector, I can easily store each DDL statement, making it > incremental. Consuming the whole topic upon restart shouldn’t be that > intensive. And if it does, then that only adds to the restart time a bit — > no big deal. > This all sounds like it should work fine, just remember that you'll have to store enough info with the DDL statement to indicate when it should be applied. The case you want to be careful of is if your connector processes the change and crashes before offsets are committed. When the process comes back up, you need to make sure that if it recovers and starts at an earlier offset that it will still be able to get the previous schema. That's the drawback of not being able to tie writes of state in with offset commits -- you need to handle potential inconsistencies. > Finally, I'd love to know which DB you're reading the transaction log from > > and if you're planning on open sourcing the connector:) > > > As I mentioned, it’s MySQL at the moment and part of a very (!) new OSS > project called Debezium (http://debezium.io), and the MySQL connector is > the first of hopefully many connectors for a variety of databases. > (Contributors welcome!) > Looks cool! I see you have at least skeletons for a couple of different DBs. Will the Kafka Connect adapter work with all of them? -Ewen > > Best regards, > > Randall > -- Thanks, Ewen
