iotpg: не возвращать ошибку при постоянной (poison) ошибке flush — данные уже обработаны построчно
This commit is contained in:
@@ -162,18 +162,20 @@ func (s *IoTPostgresStore) flushAll(ctx context.Context) error {
|
|||||||
for ns, rows := range batches {
|
for ns, rows := range batches {
|
||||||
if err := s.flushTenant(ctx, ns, rows); err != nil {
|
if err := s.flushTenant(ctx, ns, rows); err != nil {
|
||||||
s.log.Error("iotpg: flush batch", "namespace", ns, "rows", len(rows), "err", err)
|
s.log.Error("iotpg: flush batch", "namespace", ns, "rows", len(rows), "err", err)
|
||||||
if firstErr == nil {
|
|
||||||
firstErr = err
|
|
||||||
}
|
|
||||||
if isRetryableFlushErr(err) {
|
if isRetryableFlushErr(err) {
|
||||||
|
if firstErr == nil {
|
||||||
|
firstErr = err
|
||||||
|
}
|
||||||
// Сеть/PG временно недоступен — возвращаем строки в буфер, повторим.
|
// Сеть/PG временно недоступен — возвращаем строки в буфер, повторим.
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
s.batches[ns] = append(s.batches[ns], rows...)
|
s.batches[ns] = append(s.batches[ns], rows...)
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
} else {
|
} else {
|
||||||
// Постоянная ошибка: валидные строки уже вставлены построчным
|
// Постоянная ошибка: валидные строки уже вставлены построчным
|
||||||
// fallback'ом (см. flushTenant), битые отброшены с логом. Повтор
|
// fallback'ом (см. flushTenant), битые отброшены с логом. Ошибку
|
||||||
// бессмыслен — в буфер НЕ возвращаем (иначе залипание namespace).
|
// НЕ возвращаем и строки в буфер НЕ возвращаем — данные обработаны,
|
||||||
|
// иначе consumer увидел бы ошибку и бесконечно гонял poison-батч
|
||||||
|
// через visibility timeout (ревью 🔴).
|
||||||
s.log.Error("iotpg: permanent flush error (валидные строки — построчно, битые отброшены)", "namespace", ns, "rows", len(rows), "err", err)
|
s.log.Error("iotpg: permanent flush error (валидные строки — построчно, битые отброшены)", "namespace", ns, "rows", len(rows), "err", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user