kafka-jira mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Randall Hauch (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (KAFKA-6661) Sink connectors that explicitly 'resume' topic partitions can resume a paused task
Date Wed, 14 Mar 2018 23:02:00 GMT

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

Randall Hauch commented on KAFKA-6661:
--------------------------------------

The {{WorkerSinkTask}} and {{WorkerSinkTaskContext}} both deal with pausing and resuming the
consumer. Most of this logic is correct:

* {{WorkerSinkTaskContext}} is already tracking the topic partitions that have been explicitly
paused or resumed by the connector
* {{WorkerSinkTaskContext.pause(TopicPartition...)}} adds the topic partitions to its paused
set and always pauses those topic partitions in the consumer
* {{WorkerSinkTaskContext.resume(TopicPartition...)}} removes the topic partitions from its
paused set and *_always_* resumes those topic partitions in the consumer. *The _always_ part
is what is incorrect; it should still remove the topic partitions from its set but should
only tell the consumer to resume the topic partitions _when the consumer is not paused_.*
* {{WorkerSinkTask.pauseAll()}} currently pauses all of the partitions, but does not use or
change the context's set of paused partitions
* {{WorkerSinkTask.resumeAll()}} currently resumes all topic partitions *_except_* those that
are still explicitly paused in the context

So, I think the only change that needs to be made is that {{WorkerSinkTaskContext.resume(TopicPartition...)}}
should still remove the topic partitions from its set but should only tell the consumer to
resume the topic partitions _when the consumer is not paused_.

> Sink connectors that explicitly 'resume' topic partitions can resume a paused task
> ----------------------------------------------------------------------------------
>
>                 Key: KAFKA-6661
>                 URL: https://issues.apache.org/jira/browse/KAFKA-6661
>             Project: Kafka
>          Issue Type: Bug
>          Components: KafkaConnect
>    Affects Versions: 0.9.0.0, 0.10.0.0, 0.11.0.0, 1.0.0
>            Reporter: Randall Hauch
>            Assignee: Randall Hauch
>            Priority: Critical
>
> Sink connectors are allowed to use the {{SinkTaskContext}}'s methods to explicitly pause
and resume topic partitions. This is useful when connectors need additional time processing
the records for specific topic partitions (e.g., the external system has an outage).
> However, when the sink connector has been paused via the REST API, the worker for the
sink tasks pause the consumer. When the connector is polled, the poll request might timeout
and return no records. Connect then calls the task's {{put(...)}} method (with no records),
and this allows the task to optionally call any of the {{SinkTaskContext}}'s pause or resume
methods. If it calls resume, this will unexpectedly resume the paused consumer, causing the
consumer to return messages and the connector to process those messages --  despite the connector
still being paused.
> This is reported against 1.0, but the affected code has not been changed since at least
0.9.0.0.
> A workaround is to remove rather than pause a connector. It's inconvenient, but it works.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Mime
View raw message