You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Chris Egerton (Jira)" <ji...@apache.org> on 2021/07/06 13:14:00 UTC
[jira] [Commented] (KAFKA-12857) Using Connect Sink with
CooperativeStickyAssignor results in commit offsets failure
[ https://issues.apache.org/jira/browse/KAFKA-12857?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17375714#comment-17375714 ]
Chris Egerton commented on KAFKA-12857:
---------------------------------------
I believe this is a duplicate of https://issues.apache.org/jira/browse/KAFKA-12487, which has a PR available and is awaiting review (possibly from [~kkonstantine]): [https://github.com/apache/kafka/pull/10563]
> Using Connect Sink with CooperativeStickyAssignor results in commit offsets failure
> -----------------------------------------------------------------------------------
>
> Key: KAFKA-12857
> URL: https://issues.apache.org/jira/browse/KAFKA-12857
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
> Affects Versions: 2.7.1
> Environment: Linux
> Reporter: Oliver Hsu
> Assignee: dgd_contributor
> Priority: Major
>
> We are attempting to use a Kafka Connect Sink Connector with {{CooperativeStickyAssignor}} assignment strategy. When we use {{CooperativeStickyAssignor}} offset commits sometimes fail with
> {{[2021-05-26 22:03:36,435] WARN WorkerSinkTask\{id=sink-connector-7} Ignoring invalid task provided offset mytopic-0/OffsetAndMetadata\{offset=16305575, leaderEpoch=null, metadata=''} – partition not assigned, assignment=[mytopic-0] (org.apache.kafka.connect.runtime.WorkerSinkTask:434)}}
> Note that the invalid partition in the warning message matches the partition assignment.
> *Config changes*
> {{consumer.partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor}}
> *Cooperative vs Eager Assignment Strategy background*
> [https://cwiki.apache.org/confluence/display/KAFKA/KIP-429%3A+Kafka+Consumer+Incremental+Rebalance+Protocol#KIP429:KafkaConsumerIncrementalRebalanceProtocol-ConsumerRebalanceListenerandConsumerPartitionAssignorSemantics]
> With eager assignment:
> {quote}Listener#onPartitionsAssigned: called on the full set of assigned partitions (may have overlap with the partitions passed to #onPartitionsRevoked
> {quote}
> With cooperative assignment:
> {quote}Listener#onPartitionsAssigned: called on the subset of assigned partitions that were not previously owned before this rebalance. There should be no overlap with the revoked partitions (if any). This will always be called, even if there are no new partitions being assigned to a given member.
> {quote}
> This means with cooperative assignment, `onPartitionsAssigned` may be called with a partial assignment or an empty collection.
> However, the [WorkerSinkTask.HandleRebalance|https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java#L669-L680] class makes the assumption that `onPartitionsAssigned` is called with the full set of assigned partitions which is true for eager but not coooperative.
> {code:java|title=WorkerSinkTask.HandleRebalance.java|borderStyle=solid}
> public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
> log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
> lastCommittedOffsets = new HashMap<>();
> currentOffsets = new HashMap<>();
> for (TopicPartition tp : partitions) {
> long pos = consumer.position(tp);
> lastCommittedOffsets.put(tp, new OffsetAndMetadata(pos));
> currentOffsets.put(tp, new OffsetAndMetadata(pos));
> log.debug("{} Assigned topic partition {} with offset {}", WorkerSinkTask.this, tp, pos);
> }
> {code}
> The {{onPartitionsAssigned}} creates a new empty {{HashMap}} and puts the offsets of the {{partitions}} in that {{HashMap}}.
> In the logs we see
> {{[2021-05-26 22:02:09,785] DEBUG WorkerSinkTask\{id=sink-connector-7} Partitions assigned [myTopic-0] (org.apache.kafka.connect.runtime.WorkerSinkTask:677)}}
> {{[2021-05-26 22:02:13,063] DEBUG WorkerSinkTask\{id=sink-connector-7} Partitions assigned [] (org.apache.kafka.connect.runtime.WorkerSinkTask:677)}}
> {{[2021-05-26 22:02:16,074] DEBUG WorkerSinkTask\{id=sink-connector-7} }} Partitions assigned [] (org.apache.kafka.connect.runtime.WorkerSinkTask:677)}}
> These logs show that the {{CooperativeStickyAssignor}} calls {{onPartitionsAssigned}} first with the partition assigned to it followed by additional calls with an empty {{partitions}} collection.
> When {{HandleRebalance.onPartitionsAssigned}} is called first with the assigned partition followed by empty collections, the result will be {{lastCommittedOffsets}} initialized to an empty {{HashMap}}.
> Inside [WorkerSinkTask.commitOffsets|https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java#L415-L419], the current {{committableOffsets}} are based on the {{lastCommittedOffsets}}, which is an empty {{HashMap}}:
> {code:java|title=WorkerSinkTask.java|borderStyle=solid}
> private void commitOffsets(long now, boolean closing) {
> ...
> final Map<TopicPartition, OffsetAndMetadata> commitableOffsets = new HashMap<>(lastCommittedOffsets);
> for (Map.Entry<TopicPartition, OffsetAndMetadata> taskProvidedOffsetEntry : taskProvidedOffsets.entrySet()) {
> final TopicPartition partition = taskProvidedOffsetEntry.getKey();
> final OffsetAndMetadata taskProvidedOffset = taskProvidedOffsetEntry.getValue();
> if (commitableOffsets.containsKey(partition)) {
> long taskOffset = taskProvidedOffset.offset();
> long currentOffset = currentOffsets.get(partition).offset();
> if (taskOffset <= currentOffset) {
> commitableOffsets.put(partition, taskProvidedOffset);
> } else {
> log.warn("{} Ignoring invalid task provided offset {}/{} -- not yet consumed, taskOffset={} currentOffset={}",
> this, partition, taskProvidedOffset, taskOffset, currentOffset);
> }
> } else {
> log.warn("{} Ignoring invalid task provided offset {}/{} -- partition not assigned, assignment={}",
> this, partition, taskProvidedOffset, consumer.assignment());
> }
> }
> {code}
> {{if (commitableOffsets.containsKey(partition))}} is {{false}} because {{commitableOffsets} is an empty HashMap}} based on the empty {{lastCommittedOffsets HashMap}} . This causes {{{} Ignoring invalid task provided offset {}/{} – partition not assigned, assignment=}} to occur even though the task is assigned the partition.
> e.g.
> {{[2021-05-26 22:03:36,435] WARN WorkerSinkTask\{id=sink-connector-7} Ignoring invalid task provided offset mytopic-0/OffsetAndMetadata\{offset=16305575, leaderEpoch=null, metadata=''} – partition not assigned, assignment=[mytopic-0] (org.apache.kafka.connect.runtime.WorkerSinkTask:434)}}
> *Recommended Changes*
> {{WorkerSinkTask.HandleRebalance.onPartitionsAssigned}} needs to handle the new cooperative assignment strategy which may call {{onPartitionsAssigned}} with a subset of assigned partitions or an empty collection if no new partitions being assigned to a given member when initializing {{lastCommitedOffsets}}.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)