[jira] [Commented] (KAFKA-5084) Allow Kafka connect source tasks to commit offsets without messages being sent

["Randall Hauch (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-5084?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15973498#comment-15973498 ] 

Randall Hauch commented on KAFKA-5084:
--------------------------------------

A bit more background may be in order. The Debezium MySQL connector can be configured to ignore binlog events that don't meet a certain criteria, so the connector is successfully making progress in the binlog event but will only produce {{SourceRecord}} objects when the binlog events satisfy the criteria. In discussions with [~criccomini], it seems that some MySQL connectors are configured such that very few of the binlog events satisfy the criteria, and so that connector very rarely outputs a {{SourceRecord}} even though large numbers of binlog events have been successfully processed.

The challenge is that the only way for a connector to supply a source offset to Kafka Connect is by including it in a {{SourceRecord}}. However, if the connector has no need to produce a {{SourceRecord}}, the connector can't tell Kafka Connect that it has made progress and it has a new offset that should be committed (at the next appropriate time).

> Allow Kafka connect source tasks to commit offsets without messages being sent
> ------------------------------------------------------------------------------
>
>                 Key: KAFKA-5084
>                 URL: https://issues.apache.org/jira/browse/KAFKA-5084
>             Project: Kafka
>          Issue Type: New Feature
>          Components: KafkaConnect
>    Affects Versions: 0.10.2.0
>            Reporter: Chris Riccomini
>
> We are currently running [Debezium|http://debezium.io/] connectors in Kafka connect. These connectors consume from MySQL's binlog, and produce into Kafka.
> One of the things we've observed is that some of our Debezium connectors are not honoring the {{offset.flush.interval.ms}} setting (which is set to 60 seconds). Some of our connectors seem to be committing only sporadically. For low-volume connectors, the commits seem to happen once every hour or two, and sometimes even longer.
> It sounds like the issue is that Kafka connect will only commit source task offsets when the source task produces new source records. This is because Kafka connect gets the offset to commit from an incoming source record. The problem with this approach is that there are (in my opinion) valid reasons to want to commit consumed offsets WITHOUT sending any new messages. Taking Debezium as an example, there are cases where Debezium consumes messages, but filters out messages based on a regex, or filter rule (e.g. table black lists). In such a case, Debezium is consuming messages from MySQL's binlog, and dropping them before they get to the Kafka connect framework. As such, Kafka connect never sees these messages, and doesn't commit any progress. This results in several problems:
> # In the event of a failure, the connector could fall WAY back, since the last committed offset might be from hours ago, even thought it *has* processed all recent messages--it just hasn't sent anything to Kafka.
> # For connectors like Debezium that consume from a source that has a *limited* window to fetch messages (MySQL's binlog has time/size based retention), you can actually fall off the edge of the binlog because the last commit can actually happen farther back than the binlog goes, even though Debezium has fetched every single message in the binlog--it just hasn't produced anything due to filtering.
> Again, I don't see this as a Debezium-specific issue. I could imagine a similar scenario with an [SST-based Cassandra source|https://github.com/datamountaineer/stream-reactor/issues/162].
> It would be nice if Kafka connect allowed us a way to commit offsets for source tasks even when messages haven't been sent recently. This would allow source tasks to log their progress even if they're opting not to send messages to Kafka due to filtering or for some other reason.
> (See https://issues.jboss.org/browse/DBZ-220 for more context.)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)