You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Oliver Hsu (Jira)" <ji...@apache.org> on 2021/05/28 00:33:00 UTC
[jira] [Updated] (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:all-tabpanel ]
Oliver Hsu updated KAFKA-12857:
-------------------------------
Description:
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 partition that the warning message says is not assigned 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 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 was 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 the cooperative assignor calls {{HandleRebalance.onPartitionsAssigned}} 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}}.
was:
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 partition that the warning message says is not assigned 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 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 was 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 [scalyr-logstaging-v1-21050_1129447869326033792-AAAAAFulBCBPQjUB-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.
The {{HandleRebalance.onPartitionsAssigned}} code will then initialize {{lastCommittedOffsets}} 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}}.
> 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
> 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 partition that the warning message says is not assigned 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 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 was 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 the cooperative assignor calls {{HandleRebalance.onPartitionsAssigned}} 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)