当我下载大文件时，kafka 重试多次

Question

我是kafka的新手，我尝试构建一个服务发送带有附件的邮件。执行流程：

Kafka会收到消息发送邮件
函数获取文件将从 url 下载文件，缩放图像，并保存文件
发送邮件时，我将从文件夹中获取文件并附加到表单中问题：
当我多次发送大文件邮件时，kafka重试多次，我会收到很多邮件

kafka 错误：“kafka 服务器：提供的成员在当前代中未知”

我听了 MaxProcessingTime ，但我尝试用大文件测试邮件，它仍然工作正常

Kafka 信息：1 个代理，3 个消费者

func (s *customerMailService) SendPODMail() error { filePaths, err := DownloadFiles(podURLs, orderInfo.OrderCode)

if err != nil{
    countRetry := 0
    for countRetry <= NUM_OF_RETRY{
        filePaths, err = DownloadFiles(podURLs, orderInfo.OrderCode)
        if err == nil{
            break
        }
        countRetry++
    }
}

    err = s.sendMailService.Send(ctx, orderInfo.CustomerEmail, tmsPod, content,filePaths)}

函数下载文件：

func DownloadFiles(files []string, orderCode string) ([]string, error) {
var filePaths []string

err := os.Mkdir(tempDir, 0750)
if err != nil && !os.IsExist(err) {
    return nil, err
}

tempDirPath := tempDir + "/" + orderCode
err = os.Mkdir(tempDirPath, 0750)
if err != nil && !os.IsExist(err) {
    return nil, err
}

for _, fileUrl := range files {
    fileUrlParsed, err := url.ParseRequestURI(fileUrl)
    if err != nil {
        logrus.WithError(err).Infof("Pod url is invalid %s", orderCode)
        return nil, err
    }

    extFile := filepath.Ext(fileUrlParsed.Path)
    dir, err := os.MkdirTemp(tempDirPath, "tempDir")

    if err != nil {
        return nil, err
    }

    f, err := os.CreateTemp(dir, "tmpfile-*"+extFile)
    if err != nil {
        return nil, err
    }
    defer f.Close()

    response, err := http.Get(fileUrl)
    if err != nil {
        return nil, err
    }
    defer response.Body.Close()

    contentTypes := response.Header["Content-Type"]
    isTypeAllow := false
    for _, contentType := range contentTypes {
        if contentType == "image/png" || contentType == "image/jpeg" {
            isTypeAllow = true
        }
    }

    if !isTypeAllow {
        logrus.WithError(err).Infof("Pod image type is invalid %s", orderCode)
        return nil, errors.New("Pod image type is invalid")
    }

    decodedImg, err := imaging.Decode(response.Body)
    if err != nil {
        return nil, err
    }

    resizedImg := imaging.Resize(decodedImg, 1024, 0, imaging.Lanczos)

    imaging.Save(resizedImg, f.Name())

    filePaths = append(filePaths, f.Name())
}
return filePaths, nil}

函数发送邮件

func (s *tikiMailService) SendFile(ctx context.Context, receiver string, templateCode string, data interface{}, filePaths []string) error {
path := "/v1/emails"
fullPath := fmt.Sprintf("%s%s", s.host, path)

formValue := &bytes.Buffer{}
writer := multipart.NewWriter(formValue)
_ = writer.WriteField("template", templateCode)
_ = writer.WriteField("to", receiver)

if data != nil {
    b, err := json.Marshal(data)
    if err != nil {
        return errors.Wrapf(err, "Cannot marshal mail data to json with object %+v", data)
    }

    _ = writer.WriteField("params", string(b))
}

for _, filePath := range filePaths {
    part, err := writer.CreateFormFile(filePath, filepath.Base(filePath))

    if err != nil {
        return err
    }


    pipeReader, pipeWriter := io.Pipe()

    go func() {
        defer pipeWriter.Close()

        file, err := os.Open(filePath)
        if err != nil {
            return 
        }
        defer file.Close()

        io.Copy(pipeWriter, file)
    }()

    io.Copy(part, pipeReader)
}

err := writer.Close()
if err != nil {
    return err
}

request, err := http.NewRequest("POST", fullPath, formValue)
if err != nil {
    return err
}
request.Header.Set("Content-Type", writer.FormDataContentType())

resp, err := s.doer.Do(request)
if err != nil {
    return errors.Wrap(err, "Cannot send request to send email")
}
defer resp.Body.Close()

b, err := ioutil.ReadAll(resp.Body)
if err != nil {
    return err
}

if resp.StatusCode != http.StatusOK {
    return errors.New(fmt.Sprintf("Send email with code %s error: status code %d, response %s",
        templateCode, resp.StatusCode, string(b)))
} else {
    logrus.Infof("Send email with attachment ,code %s success with response %s , box-code", templateCode, string(b),filePaths)
}
return nil
}

感谢

Answer 1

我的两分钱：如果附件非常大，用户需要花费大量时间来阅读文件并将其作为附件发送。

这会增加两次 poll() 调用之间的时间量。如果该时间大于max.poll.interval.ms，则认为消费者失败，分区偏移量未提交。结果，消息被再次处理，最终，如果执行时间碰巧保持在轮询间隔以下，则提交偏移量。效果是多封邮件发送。

尝试增加消费者端的 max.poll.interval.ms。

Answer 2

我的团队在我重新部署 k8s 时发现了我的问题 pods，导致领导分区冲突导致重新平衡。它将再次尝试处理 pods 缓冲区中剩余的消息。

解决方案：我没有获取缓冲区中保存的许多消息，我只是获取一条消息并通过配置处理它：

ChannelBufferSize = 0

冲突领导分区示例：

consumer A and B startup in the same time
consumer A registers itself as leader, and owns the topic with all partitions
consumer B registers itself as leader, and then begins to rebalance and owns all partitions
consumer A rebalance and obtains all partitions, but can not consume because the memberId is old and need a new one
consumer B rebalance again and owns the topic with all partitions, but it's already obtained by consumer A

当我下载大文件时，kafka 重试多次

kafka retry many times when i download large file

go

apache-kafka