From 6004c923a658729dff622cd890307e9133f00aa2 Mon Sep 17 00:00:00 2001 From: dongdong Date: Wed, 14 Aug 2024 10:36:08 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20transfer=20=E4=BF=AE=E5=A4=8D=E7=B4=A2?= =?UTF-8?q?=E5=BC=95=E5=88=87=E5=89=B2=E9=97=AE=E9=A2=98=20--story=3D11917?= =?UTF-8?q?9630=20(#485)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/transfer/elasticsearch/backend.go | 2 +- pkg/transfer/template/etl/flat_batch.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/transfer/elasticsearch/backend.go b/pkg/transfer/elasticsearch/backend.go index 0047b7153..12dddae83 100644 --- a/pkg/transfer/elasticsearch/backend.go +++ b/pkg/transfer/elasticsearch/backend.go @@ -195,7 +195,7 @@ func (b *BulkHandler) Flush(ctx context.Context, results []interface{}) (count i // 处理跨时间间隔 if index != lastIndex && lastIndex != "" { - cnt, err := b.flush(ctx, index, records) + cnt, err := b.flush(ctx, lastIndex, records) records = records[:0] count += cnt errs.Add(err) diff --git a/pkg/transfer/template/etl/flat_batch.go b/pkg/transfer/template/etl/flat_batch.go index 115240a9a..1a57da761 100644 --- a/pkg/transfer/template/etl/flat_batch.go +++ b/pkg/transfer/template/etl/flat_batch.go @@ -86,7 +86,7 @@ func (p *FlatBatchHandler) Process(d define.Payload, outputChan chan<- define.Pa err := d.To(&originMap) if err != nil { p.CounterFails.Inc() - logging.Errorf("%v convert payload %#v error %v", p, d, err) + logging.MinuteErrorfSampling(p.String(), "%v convert payload %#v error %v", p, d, err) return }