Merge cockroachdb#65751

65751: changefeedccl: Refactor errors and error wrapper event buffer. r=miretskiy a=miretskiy

This is a pure code move/refactoring to address comments from cockroachdb#65746:
  * Move errors to changefeedbase so that they are accessible.
  * Move error wrapper buffer to buffer.go
  * Unexport mem buffer.

Release Notes: None

Co-authored-by: Yevgeniy Miretskiy <[email protected]>

pkg/ccl/changefeedccl/BUILD.bazel

@@ -10,7 +10,6 @@ go_library(
         "changefeed_stmt.go",
         "doc.go",
         "encoder.go",
-        "errors.go",
         "metrics.go",
         "name.go",
         "rowfetcher_cache.go",

pkg/ccl/changefeedccl/bench_test.go

@@ -216,16 +216,15 @@ func createBenchmarkChangefeed(
 	}
 	_, withDiff := details.Opts[changefeedbase.OptDiff]
 	kvfeedCfg := kvfeed.Config{
-		Settings: settings,
-		DB:       s.DB(),
-		Clock:    feedClock,
-		Gossip:   gossip.MakeOptionalGossip(s.GossipI().(*gossip.Gossip)),
-		Spans:    spans,
-		Targets:  details.Targets,
-		Sink:     buf,
-		EventBufferFactory: func() kvfeed.EventBuffer {
-			return kvfeed.MakeMemBuffer(mm.MakeBoundAccount(), &metrics.KVFeedMetrics)
-		},
+		Settings:         settings,
+		DB:               s.DB(),
+		Clock:            feedClock,
+		Gossip:           gossip.MakeOptionalGossip(s.GossipI().(*gossip.Gossip)),
+		Spans:            spans,
+		Targets:          details.Targets,
+		Sink:             buf,
+		Metrics:          &metrics.KVFeedMetrics,
+		MM:               mm,
 		InitialHighWater: initialHighWater,
 		WithDiff:         withDiff,
 		NeedsInitialScan: needsInitialScan,

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -244,7 +244,7 @@ func (ca *changeAggregator) Start(ctx context.Context) {
 		ca.spec.User(), kvFeedMemMon.MakeBoundAccount(), ca.spec.JobID)
 
 	if err != nil {
-		err = MarkRetryableError(err)
+		err = changefeedbase.MarkRetryableError(err)
 		// Early abort in the case that there is an error creating the sink.
 		ca.MoveToDraining(err)
 		ca.cancel()
@@ -355,33 +355,6 @@ func (f doNothingSchemaFeed) Pop(
 	return nil, nil
 }
 
-type errorWrapperEventBuffer struct {
-	kvfeed.EventBuffer
-}
-
-func (e errorWrapperEventBuffer) AddKV(
-	ctx context.Context, kv roachpb.KeyValue, prevVal roachpb.Value, backfillTimestamp hlc.Timestamp,
-) error {
-	if err := e.EventBuffer.AddKV(ctx, kv, prevVal, backfillTimestamp); err != nil {
-		return MarkRetryableError(err)
-	}
-	return nil
-}
-
-func (e errorWrapperEventBuffer) AddResolved(
-	ctx context.Context,
-	span roachpb.Span,
-	ts hlc.Timestamp,
-	boundaryType jobspb.ResolvedSpan_BoundaryType,
-) error {
-	if err := e.EventBuffer.AddResolved(ctx, span, ts, boundaryType); err != nil {
-		return MarkRetryableError(err)
-	}
-	return nil
-}
-
-var _ kvfeed.EventBuffer = (*errorWrapperEventBuffer)(nil)
-
 func makeKVFeedCfg(
 	cfg *execinfra.ServerConfig,
 	mm *mon.BytesMonitor,
@@ -397,11 +370,7 @@ func makeKVFeedCfg(
 		spec.Feed.Opts[changefeedbase.OptSchemaChangePolicy])
 	_, withDiff := spec.Feed.Opts[changefeedbase.OptDiff]
 	initialHighWater, needsInitialScan := getKVFeedInitialParameters(spec)
-	bf := func() kvfeed.EventBuffer {
-		return &errorWrapperEventBuffer{
-			EventBuffer: kvfeed.MakeMemBuffer(mm.MakeBoundAccount(), &metrics.KVFeedMetrics),
-		}
-	}
+
 	kvfeedCfg := kvfeed.Config{
 		Sink:               buf,
 		Settings:           cfg.Settings,
@@ -411,7 +380,8 @@ func makeKVFeedCfg(
 		Gossip:             cfg.Gossip,
 		Spans:              spans,
 		Targets:            spec.Feed.Targets,
-		EventBufferFactory: bf,
+		Metrics:            &metrics.KVFeedMetrics,
+		MM:                 mm,
 		InitialHighWater:   initialHighWater,
 		WithDiff:           withDiff,
 		NeedsInitialScan:   needsInitialScan,
@@ -1052,7 +1022,7 @@ func (cf *changeFrontier) Start(ctx context.Context) {
 		cf.spec.User(), mm.MakeBoundAccount(), cf.spec.JobID)
 
 	if err != nil {
-		err = MarkRetryableError(err)
+		err = changefeedbase.MarkRetryableError(err)
 		cf.MoveToDraining(err)
 		return
 	}
@@ -1170,7 +1140,7 @@ func (cf *changeFrontier) Next() (rowenc.EncDatumRow, *execinfrapb.ProducerMetad
 				if cf.EvalCtx.Settings.Version.IsActive(
 					cf.Ctx, clusterversion.ChangefeedsSupportPrimaryIndexChanges,
 				) {
-					err = MarkRetryableError(err)
+					err = changefeedbase.MarkRetryableError(err)
 				} else {
 					err = errors.Wrap(err, "primary key change occurred")
 				}

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -293,7 +293,7 @@ func changefeedPlanHook(
 			if err != nil {
 				telemetry.Count(`changefeed.core.error`)
 			}
-			return MaybeStripRetryableErrorMarker(err)
+			return changefeedbase.MaybeStripRetryableErrorMarker(err)
 		}
 
 		// Changefeeds are based on the Rangefeed abstraction, which requires the
@@ -322,7 +322,7 @@ func changefeedPlanHook(
 				jobspb.InvalidJobID,
 			)
 			if err != nil {
-				return MaybeStripRetryableErrorMarker(err)
+				return changefeedbase.MaybeStripRetryableErrorMarker(err)
 			}
 			if err := canarySink.Close(); err != nil {
 				return err
@@ -607,7 +607,7 @@ func (b *changefeedResumer) Resume(ctx context.Context, execCtx interface{}) err
 			}
 		}
 
-		if !IsRetryableError(err) {
+		if !changefeedbase.IsRetryableError(err) {
 			if ctx.Err() != nil {
 				return ctx.Err()
 			}

pkg/ccl/changefeedccl/changefeed_test.go

@@ -2016,7 +2016,7 @@ func TestChangefeedRetryableError(t *testing.T) {
 		knobs.BeforeEmitRow = func(_ context.Context) error {
 			switch atomic.LoadInt64(&failEmit) {
 			case 1:
-				return MarkRetryableError(fmt.Errorf("synthetic retryable error"))
+				return changefeedbase.MarkRetryableError(fmt.Errorf("synthetic retryable error"))
 			case 2:
 				return fmt.Errorf("synthetic terminal error")
 			default:
@@ -3302,7 +3302,7 @@ func TestChangefeedMemBufferCapacityErrorRetryable(t *testing.T) {
 				// This function is invoked form a different go routine -- and calling
 				// t.Fatal will likely deadlock the test.
 				distErrCh <- err
-				return MaybeStripRetryableErrorMarker(err)
+				return changefeedbase.MaybeStripRetryableErrorMarker(err)
 			}
 
 			sqlDB := sqlutils.MakeSQLRunner(db)
@@ -3350,7 +3350,7 @@ func TestChangefeedMemBufferCapacityErrorRetryable(t *testing.T) {
 
 			err := <-distErrCh
 			require.Regexp(t, `memory budget exceeded`, err)
-			require.True(t, IsRetryableError(err))
+			require.True(t, changefeedbase.IsRetryableError(err))
 		}
 	}
 
@@ -3443,7 +3443,7 @@ func TestChangefeedRestartDuringBackfill(t *testing.T) {
 		require.NoError(t, feedJob.Pause())
 
 		// Make extra sure that the zombie changefeed can't write any more data.
-		beforeEmitRowCh <- MarkRetryableError(errors.New(`nope don't write it`))
+		beforeEmitRowCh <- changefeedbase.MarkRetryableError(errors.New(`nope don't write it`))
 
 		// Insert some data that we should only see out of the changefeed after it
 		// re-runs the backfill.

pkg/ccl/changefeedccl/changefeedbase/BUILD.bazel

@@ -4,13 +4,15 @@ go_library(
     name = "changefeedbase",
     srcs = [
         "avro.go",
+        "errors.go",
         "options.go",
         "settings.go",
         "validate.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/ccl/utilccl",
         "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/settings",

pkg/ccl/changefeedccl/errors.go → pkg/ccl/changefeedccl/changefeedbase/errors.go

@@ -6,7 +6,7 @@
 //
 //     https://github.com/cockroachdb/cockroach/blob/master/licenses/CCL.txt
 
-package changefeedccl
+package changefeedbase
 
 import (
 	"fmt"

pkg/ccl/changefeedccl/kvfeed/buffer.go

@@ -12,6 +12,7 @@ import (
 	"context"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
@@ -231,9 +232,7 @@ type memBuffer struct {
 	}
 }
 
-// MakeMemBuffer returns an EventBuffer backed by memory, limited
-// as specified by bound account.
-func MakeMemBuffer(acc mon.BoundAccount, metrics *Metrics) EventBuffer {
+func makeMemBuffer(acc mon.BoundAccount, metrics *Metrics) EventBuffer {
 	b := &memBuffer{
 		metrics:  metrics,
 		signalCh: make(chan struct{}, 1),
@@ -363,3 +362,30 @@ func (b *memBuffer) getRow(ctx context.Context) (tree.Datums, error) {
 		}
 	}
 }
+
+type errorWrapperEventBuffer struct {
+	EventBuffer
+}
+
+func (e errorWrapperEventBuffer) AddKV(
+	ctx context.Context, kv roachpb.KeyValue, prevVal roachpb.Value, backfillTimestamp hlc.Timestamp,
+) error {
+	if err := e.EventBuffer.AddKV(ctx, kv, prevVal, backfillTimestamp); err != nil {
+		return changefeedbase.MarkRetryableError(err)
+	}
+	return nil
+}
+
+func (e errorWrapperEventBuffer) AddResolved(
+	ctx context.Context,
+	span roachpb.Span,
+	ts hlc.Timestamp,
+	boundaryType jobspb.ResolvedSpan_BoundaryType,
+) error {
+	if err := e.EventBuffer.AddResolved(ctx, span, ts, boundaryType); err != nil {
+		return changefeedbase.MarkRetryableError(err)
+	}
+	return nil
+}
+
+var _ EventBuffer = (*errorWrapperEventBuffer)(nil)

pkg/ccl/changefeedccl/kvfeed/kv_feed.go

@@ -28,6 +28,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"github.com/cockroachdb/cockroach/pkg/util/span"
 	"github.com/cockroachdb/errors"
 )
@@ -42,7 +43,8 @@ type Config struct {
 	Spans              []roachpb.Span
 	Targets            jobspb.ChangefeedTargets
 	Sink               EventBufferWriter
-	EventBufferFactory func() EventBuffer
+	Metrics            *Metrics
+	MM                 *mon.BytesMonitor
 	WithDiff           bool
 	SchemaChangeEvents changefeedbase.SchemaChangeEventClass
 	SchemaChangePolicy changefeedbase.SchemaChangePolicy
@@ -78,14 +80,18 @@ func Run(ctx context.Context, cfg Config) error {
 		pff = rangefeedFactory(distSender.RangeFeed)
 	}
 
+	bf := func() EventBuffer {
+		return &errorWrapperEventBuffer{makeMemBuffer(cfg.MM.MakeBoundAccount(), cfg.Metrics)}
+	}
+
 	f := newKVFeed(
 		cfg.Sink, cfg.Spans,
 		cfg.SchemaChangeEvents, cfg.SchemaChangePolicy,
 		cfg.NeedsInitialScan, cfg.WithDiff,
 		cfg.InitialHighWater,
 		cfg.Codec,
 		cfg.SchemaFeed,
-		sc, pff, cfg.EventBufferFactory)
+		sc, pff, bf)
 
 	g := ctxgroup.WithContext(ctx)
 	g.GoCtx(cfg.SchemaFeed.Run)

pkg/ccl/changefeedccl/kvfeed/kv_feed_test.go

@@ -98,7 +98,7 @@ func TestKVFeed(t *testing.T) {
 		)
 		metrics := MakeMetrics(time.Minute)
 		bufferFactory := func() EventBuffer {
-			return MakeMemBuffer(mm.MakeBoundAccount(), &metrics)
+			return makeMemBuffer(mm.MakeBoundAccount(), &metrics)
 		}
 		scans := make(chan physicalConfig)
 		sf := scannerFunc(func(ctx context.Context, sink EventBufferWriter, cfg physicalConfig) error {

pkg/ccl/changefeedccl/rowfetcher_cache.go

@@ -11,6 +11,7 @@ package changefeedccl
 import (
 	"context"
 
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -86,7 +87,7 @@ func (c *rowFetcherCache) TableDescForKey(
 		if err != nil {
 			// Manager can return all kinds of errors during chaos, but based on
 			// its usage, none of them should ever be terminal.
-			return nil, MarkRetryableError(err)
+			return nil, changefeedbase.MarkRetryableError(err)
 		}
 		tableDesc = desc.Desc().(catalog.TableDescriptor)
 		// Immediately release the lease, since we only need it for the exact
@@ -110,7 +111,7 @@ func (c *rowFetcherCache) TableDescForKey(
 			}); err != nil {
 				// Manager can return all kinds of errors during chaos, but based on
 				// its usage, none of them should ever be terminal.
-				return nil, MarkRetryableError(err)
+				return nil, changefeedbase.MarkRetryableError(err)
 			}
 			// Immediately release the lease, since we only need it for the exact
 			// timestamp requested.

pkg/ccl/changefeedccl/schema_registry.go

@@ -197,7 +197,7 @@ func (r *confluentSchemaRegistry) doWithRetry(ctx context.Context, fn func() err
 		}
 		log.VInfof(ctx, 2, "retrying schema registry operation: %s", err.Error())
 	}
-	return MarkRetryableError(err)
+	return changefeedbase.MarkRetryableError(err)
 }
 
 func gracefulClose(ctx context.Context, toClose io.ReadCloser) {

pkg/ccl/changefeedccl/sink.go

@@ -154,7 +154,7 @@ func (s errorWrapperSink) EmitRow(
 	ctx context.Context, topicDescr TopicDescriptor, key, value []byte, updated hlc.Timestamp,
 ) error {
 	if err := s.wrapped.EmitRow(ctx, topicDescr, key, value, updated); err != nil {
-		return MarkRetryableError(err)
+		return changefeedbase.MarkRetryableError(err)
 	}
 	return nil
 }
@@ -164,23 +164,23 @@ func (s errorWrapperSink) EmitResolvedTimestamp(
 	ctx context.Context, encoder Encoder, resolved hlc.Timestamp,
 ) error {
 	if err := s.wrapped.EmitResolvedTimestamp(ctx, encoder, resolved); err != nil {
-		return MarkRetryableError(err)
+		return changefeedbase.MarkRetryableError(err)
 	}
 	return nil
 }
 
 // Flush implements Sink interface.
 func (s errorWrapperSink) Flush(ctx context.Context) error {
 	if err := s.wrapped.Flush(ctx); err != nil {
-		return MarkRetryableError(err)
+		return changefeedbase.MarkRetryableError(err)
 	}
 	return nil
 }
 
 // Close implements Sink interface.
 func (s errorWrapperSink) Close() error {
 	if err := s.wrapped.Close(); err != nil {
-		return MarkRetryableError(err)
+		return changefeedbase.MarkRetryableError(err)
 	}
 	return nil
 }