sink: do not error on successful batch
@@ -618,9 +618,11 @@ func (s *sink) handleReqResp(req *produceRequest, resp kmsg.Response, err error)
reqRetry.addSeqBatch(topic, partition, batch)
case err == kerr.DuplicateSequenceNumber: // ignorable, but we should not get
+ s.cl.cfg.logger.Log(LogLevelInfo, "received unexpected duplicate sequence number, ignoring and treating batch as successful")
err = nil
fallthrough
default:
+ if err != nil {
s.cl.cfg.logger.Log(LogLevelInfo, "batch in a produce request failed",
"topic", topic,
"partition", partition,
@@ -628,6 +630,7 @@ func (s *sink) handleReqResp(req *produceRequest, resp kmsg.Response, err error)
"err_is_retriable", kerr.IsRetriable(err),
"max_retries_reached", batch.tries == s.cl.cfg.retries,
)
+ }
s.cl.finishBatch(batch.recBatch, partition, rPartition.BaseOffset, err)
}
}
