kafka retry many times when i download large file

Issue

I am newbie in kafka, i try build a service send mail with attach files.
Execution flow:

• Kafka will receive a message to send mail
• function get file will download file from url , scale image, and save file
• when send mail i will get files from folder and attach to form
  Issues:
• when i send mail with large files many times , kafka retry many times, i will receive many mail

kafka error: "kafka server: The provided member is not known in the current generation"

I listened MaxProcessingTime , but i try to test a mail with large file, it still work fine

Kafka info : 1 broker , 3 consumer

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)}

function download file :

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}

function send mail

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
}

Thank

Solution

My team found my problem when I redeploy k8s pods, which lead to conflict leader partition causing rebalance. It will try to process the remaining messages in buffer of pods again.

Solution: I don’t fetch many messages saved in buffer , I just get a message and process it by config :

ChannelBufferSize = 0

Example conflict leader parition:

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

Answered By – Hưng hoàng

Answer Checked By – Timothy Miller (GoLangFix Admin)