> On 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. > > > 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. > > 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’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. > > 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!) >
Randall, I am also working on a MySQL connector. https://github.com/wushujames/kafka-mysql-connector It is a Kafka connector which uses Maxwell (https://github.com/zendesk/maxwell) to do MySQL DDL parsing, which uses open-replicator (https://github.com/zendesk/open-replicator), of which https://github.com/shyiko/mysql-binlog-connector-java is a fork. So it seems we're along the same path. Maxwell in particular is doing the things you are talking about in this thread. It captures MySQL schema changes and stores them in such a way that they are correlated with the binlog coordinates, and so they can be reloaded when the CDC process comes back up. Let me know if you'd like to talk more about this. It'd be great to collaborate. And if you're interested, here are a list of other MySQL CDC projects. https://github.com/wushujames/mysql-cdc-projects -James > Best regards, > > Randall ________________________________ This email and any attachments may contain confidential and privileged material for the sole use of the intended recipient. Any review, copying, or distribution of this email (or any attachments) by others is prohibited. If you are not the intended recipient, please contact the sender immediately and permanently delete this email and any attachments. No employee or agent of TiVo Inc. is authorized to conclude any binding agreement on behalf of TiVo Inc. by email. Binding agreements with TiVo Inc. may only be made by a signed written agreement.
