diff --git a/internal/flushcommon/syncmgr/growing_source.go b/internal/flushcommon/syncmgr/growing_source.go index 5f82403967..08f49fb2f4 100644 --- a/internal/flushcommon/syncmgr/growing_source.go +++ b/internal/flushcommon/syncmgr/growing_source.go @@ -564,6 +564,9 @@ func (t *GrowingSourceSyncTask) TargetOffset() int64 { } func (t *GrowingSourceSyncTask) HandleError(err error) { + if errors.IsAny(err, merr.ErrSegmentNotFound, merr.ErrChannelNotFound) { + return + } if t.failureCallback != nil { t.failureCallback(err) } diff --git a/internal/flushcommon/syncmgr/growing_source_test.go b/internal/flushcommon/syncmgr/growing_source_test.go index bba3a4a7eb..7ba721f692 100644 --- a/internal/flushcommon/syncmgr/growing_source_test.go +++ b/internal/flushcommon/syncmgr/growing_source_test.go @@ -5,6 +5,7 @@ import ( "testing" "github.com/cockroachdb/errors" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -18,6 +19,7 @@ import ( "github.com/milvus-io/milvus/internal/storage" "github.com/milvus-io/milvus/internal/storagecommon" "github.com/milvus-io/milvus/internal/storagev2/packed" + "github.com/milvus-io/milvus/pkg/v3/metrics" "github.com/milvus-io/milvus/pkg/v3/proto/datapb" "github.com/milvus-io/milvus/pkg/v3/util/merr" "github.com/milvus-io/milvus/pkg/v3/util/paramtable" @@ -82,6 +84,47 @@ func (s *fakeCommitGrowingFlushSource) CommitGrowingFlush(targetOffset int64) { s.commits = append(s.commits, targetOffset) } +func TestGrowingSourceSyncTaskHandleErrorSkipsFailureCallbackForStaleMetaErrors(t *testing.T) { + paramtable.Get().Init(paramtable.NewBaseTable()) + + testCases := []struct { + name string + err error + }{ + { + name: "channel_not_found", + err: merr.WrapErrChannelNotFound("ch"), + }, + { + name: "segment_not_found", + err: merr.WrapErrSegmentNotFound(1), + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + metrics.DataNodeFlushBufferCount.Reset() + callbackCalled := false + task := NewGrowingSourceSyncTask(). + WithFailureCallback(func(error) { + callbackCalled = true + panic("failure callback should not be called") + }) + + require.NotPanics(t, func() { + task.HandleError(tc.err) + task.HandleError(tc.err) + }) + require.False(t, callbackCalled) + require.Zero(t, testutil.ToFloat64(metrics.DataNodeFlushBufferCount.WithLabelValues( + paramtable.GetStringNodeID(), + metrics.FailLabel, + task.level.String(), + ))) + }) + } +} + func TestGrowingSourceSyncTaskBuildFlushConfigBM25(t *testing.T) { paramtable.Get().Init(paramtable.NewBaseTable()) paramtable.Get().Save(paramtable.Get().DataNodeCfg.TextInlineThreshold.Key, "12345")