[jira] [Commented] (KAFKA-5696) SourceConnector does not commit offset on reconfiguration

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

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

Randall Hauch commented on KAFKA-5696:
--------------------------------------

One option for the standalone and distributed modes is to have the worker explicitly commit offsets for each of the tasks just after it [stops the tasks|https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java#L454] but before the classloader is swapped, so something like this:

{code:java}
        try {
            savedLoader = Plugins.compareAndSwapLoaders(task.loader());
            task.stop();
            if (task instanceof WorkerSourceTask) {
                ((WorkerSourceTask)task).commitOffsets();
            }
        } finally {
            Plugins.compareAndSwapLoaders(savedLoader);
        }
{code}

Ideally, the {{WorkerTest}} unit tests can be modified (or a new one added) that verifies that the offset backing store is called when the worker stops the source tasks.

> SourceConnector does not commit offset on reconfiguration
> ---------------------------------------------------------
>
>                 Key: KAFKA-5696
>                 URL: https://issues.apache.org/jira/browse/KAFKA-5696
>             Project: Kafka
>          Issue Type: Bug
>          Components: KafkaConnect
>            Reporter: Oleg Kuznetsov
>              Labels: newbie
>             Fix For: 0.10.0.2
>
>
> I'm running SourceConnector, that reads files from storage and put data in kafka. I want, in case of reconfiguration, offsets to be flushed. 
> Say, a file is completely processed, but source records are not yet committed and in case of reconfiguration their offsets might be missing in store.
> Is it possible to force committing offsets on reconfiguration?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)