[jira] [Comment Edited] (KAFKA-14131) KafkaBasedLog#readToLogEnd() may accciedently falls into infinite loop

["Sagar Rao (Jira)" <ji...@apache.org>]

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

Sagar Rao edited comment on KAFKA-14131 at 8/4/22 6:20 PM:
-----------------------------------------------------------

[~Justinwins] i created a PR here => [https://github.com/apache/kafka/pull/12485.] I guess handling InterruptedException and throwing should make the hung consumer problem go away. WDYT?


was (Author: sagarrao):
[~Justinwins] i created a PR here => [https://github.com/apache/kafka/pull/12485.] It's pretty small...

> KafkaBasedLog#readToLogEnd() may accciedently falls into infinite loop
> ----------------------------------------------------------------------
>
>                 Key: KAFKA-14131
>                 URL: https://issues.apache.org/jira/browse/KAFKA-14131
>             Project: Kafka
>          Issue Type: Bug
>          Components: mirrormaker
>            Reporter: Justinwins
>            Assignee: Sagar Rao
>            Priority: Major
>
> When a herder starts ,its KafkaOffsetBackingStore will readToLogEnd() by
> DistributedHerder.herderExecutor of name "Distrubuted-connect-"  thread , e.g . Distrubuted-connect-28-1 , which may consume  a few minutes.
> If another thread tries to shut down this herder , it will block for "task.shutdown.graceful.timeout.ms ' before  the DistributedHerder.herderExecutor is interrupted. 
> And if thread in DistributedHerder.herderExecutor is interupted, KafkaOffsetBackingStore.readToLogEnd() will  poll(Integer.MAX_VALUE) and log "
> Error polling" as  the the read  has been interrupted, then "consumer.position" will not advance, readToLogEnd() falls into infinite loop.
>  
> {code:java}
> // code placeholder
> private void readToLogEnd() {
>     Set<TopicPartition> assignment = consumer.assignment();
>     Map<TopicPartition, Long> endOffsets = readEndOffsets(assignment);
>     log.trace("Reading to end of log offsets {}", endOffsets);
>     while (!endOffsets.isEmpty()) { // this loop will never jump out
>         Iterator<Map.Entry<TopicPartition, Long>> it = endOffsets.entrySet().iterator();
>         while (it.hasNext()) {
>             Map.Entry<TopicPartition, Long> entry = it.next();
>             TopicPartition topicPartition = entry.getKey();
>             long endOffset = entry.getValue();
>             long lastConsumedOffset = consumer.position(topicPartition);  // when thread was in interupted status ,consumer.position will not advance
>             if (lastConsumedOffset >= endOffset) {
>                 log.trace("Read to end offset {} for {}", endOffset, topicPartition);
>                 it.remove();
>             } else {
>                 log.trace("Behind end offset {} for {}; last-read offset is {}",
>                         endOffset, topicPartition, lastConsumedOffset);
>                 poll(Integer.MAX_VALUE); // here , poll() will catch InterruptedException and log it without throwing it up
>                 break;
>             }
>         }
>     }
> } {code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)