创建kafka.Consumer后调整偏移量，ReadMessage超时问题求助

你的核心问题是：仅通过CommitOffsets将调整后的偏移量提交到Kafka集群，但当前消费者实例的本地消费位置并未同步更新，导致ReadMessage仍然从旧的偏移位置开始读取，自然无法获取到预期的消息，最终触发超时。

Kafka消费者的本地消费位置和集群存储的提交偏移量是两个独立概念：

• CommitOffsets仅负责把偏移量持久化到Kafka的__consumer_offsets主题，供消费者重启时恢复使用
• 正在运行的消费者不会自动从集群拉取提交的偏移量来更新本地位置，必须手动通过Seek操作定位到目标偏移量

修改代码，在提交偏移量后，对需要调整偏移量的分区执行Seek操作，将消费者本地的消费位置同步到目标偏移量：

import "github.com/confluentinc/confluent-kafka-go/kafka"
import "time"

func AdjustConsumerOffset(c *kafka.Consumer) error { 
    const TimeOut = 100

    balancing := true
    for balancing {
        c.Poll(TimeOut)
        assignments, err := c.Assignment()
        if err != nil {
            return err
        }
        if len(assignments) > 0 {
            balancing = false
        }
    }

    // 获取已分配分区
    assignments, err := c.Assignment()
    if err != nil {
        return err
    }

    // 获取当前已提交的偏移量
    assignments, err = c.Committed(assignments, TimeOut)
    if err != nil {
        return err
    }

    var decreasePartitions []kafka.TopicPartition 
    partitionsMap := make(map[int32]kafka.TopicPartition)
    
    // 调整偏移量并记录需要Seek的分区
    for _, partition := range assignments {
        originalOffset := partition.Offset
        if originalOffset > 0 {
            partition.Offset--
            partitionsMap[partition.Partition] = partition
        }
        decreasePartitions = append(decreasePartitions, partition)
    }
    
    // 提交新偏移量到集群
    if _, err := c.CommitOffsets(decreasePartitions); err != nil {
        return err
    }

    // 关键操作：将消费者本地位置Seek到调整后的偏移量
    for _, tp := range partitionsMap {
        if err := c.Seek(tp, TimeOut); err != nil {
            return err
        }
    }

    // 现在读取消息即可获取到预期内容
    for len(partitionsMap) > 0 {
        // 注意：原代码超时单位错误，100*time.Second会变成100秒，修正为毫秒
        msg, err := c.ReadMessage(TimeOut * time.Millisecond) 
        if err != nil {
            if kafkaErr, ok := err.(kafka.Error); ok && kafkaErr.Code() == kafka.ErrTimedOut {
                break
            }
            return err
        }
        // 处理消息逻辑...
        // 处理完成后从map中移除对应分区，避免循环无法退出
        delete(partitionsMap, msg.TopicPartition.Partition)
    }

    return nil
}

• 超时单位修正：原代码中TimeOut定义为100，后续ReadMessage误用time.Second会导致100秒超时，建议确认单位为毫秒并使用time.Millisecond
• 分区循环退出逻辑：消息处理完成后，必须从partitionsMap中移除对应分区，否则循环会一直执行直到超时
• Seek操作的必要性：任何时候需要在运行中的消费者实例上调整消费位置，都必须使用Seek，仅提交偏移量不会改变当前消费者的读取位置

内容的提问来源于stack exchange，提问作者Maryam