changefeedccl: add roachtest and parallel test for enriched envelope

Add a roachtest and a parallelized test for enriched envelope feeds.

Part of: #139660

Release note: None

Author: asg0451

pkg/ccl/changefeedccl/cdctest/validator.go

@@ -1017,21 +1017,26 @@ func ParseJSONValueTimestamps(v []byte) (updated, resolved hlc.Timestamp, err er
 	var valueRaw struct {
 		Resolved string `json:"resolved"`
 		Updated  string `json:"updated"`
+		Source   struct {
+			TsHLC string `json:"ts_hlc"`
+		} `json:"source"`
 	}
 	if err := gojson.Unmarshal(v, &valueRaw); err != nil {
 		return hlc.Timestamp{}, hlc.Timestamp{}, errors.Wrapf(err, "parsing [%s] as json", v)
 	}
+
 	if valueRaw.Updated != `` {
-		var err error
-		updated, err = hlc.ParseHLC(valueRaw.Updated)
-		if err != nil {
+		if updated, err = hlc.ParseHLC(valueRaw.Updated); err != nil {
 			return hlc.Timestamp{}, hlc.Timestamp{}, err
 		}
 	}
 	if valueRaw.Resolved != `` {
-		var err error
-		resolved, err = hlc.ParseHLC(valueRaw.Resolved)
-		if err != nil {
+		if resolved, err = hlc.ParseHLC(valueRaw.Resolved); err != nil {
+			return hlc.Timestamp{}, hlc.Timestamp{}, err
+		}
+	}
+	if valueRaw.Source.TsHLC != `` {
+		if updated, err = hlc.ParseHLC(valueRaw.Source.TsHLC); err != nil {
 			return hlc.Timestamp{}, hlc.Timestamp{}, err
 		}
 	}

pkg/ccl/changefeedccl/changefeed_test.go

@@ -4193,6 +4193,99 @@ func TestChangefeedEnriched(t *testing.T) {
 	}
 }
 
+// TestChangefeedsParallelEnriched tests that multiple changefeeds can run in
+// parallel with the enriched envelope. It is most useful under race, to ensure
+// that there is no accidental data races in the encoders and source providers.
+func TestChangefeedsParallelEnriched(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+
+	const numFeeds = 10
+	const maxIterations = 1_000_000_000
+	const maxRows = 100
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		db := sqlutils.MakeSQLRunner(s.DB)
+		db.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
+
+		ctx, cancel := context.WithCancel(ctx)
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			db := sqlutils.MakeSQLRunner(s.Server.SQLConn(t))
+			var i int
+			for i = 0; i < maxIterations && ctx.Err() == nil; i++ {
+				db.Exec(t, `UPSERT INTO d.foo VALUES ($1, $2)`, i%maxRows, fmt.Sprintf("hello %d", i))
+			}
+		}()
+
+		opts := `envelope='enriched'`
+
+		_, isKafka := f.(*kafkaFeedFactory)
+		useAvro := isKafka && rand.Intn(2) == 0
+		if useAvro {
+			t.Logf("using avro")
+			opts += `, format='avro'`
+		}
+		var feeds []cdctest.TestFeed
+		for range numFeeds {
+			feed := feed(t, f, fmt.Sprintf(`CREATE CHANGEFEED FOR foo WITH %s`, opts))
+			feeds = append(feeds, feed)
+		}
+
+		// consume from the feeds
+		for _, feed := range feeds {
+			feed := feed
+			msgCount := 0
+
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				for ctx.Err() == nil {
+					_, err := feed.Next()
+					if err != nil {
+						if errors.Is(err, context.Canceled) {
+							t.Errorf("error reading from feed: %v", err)
+						}
+						break
+					}
+					msgCount++
+				}
+				assert.GreaterOrEqual(t, msgCount, 0)
+			}()
+		}
+
+		// let the feeds run for a few seconds
+		select {
+		case <-time.After(5 * time.Second):
+		case <-ctx.Done():
+			t.Fatalf("%v", ctx.Err())
+		}
+
+		cancel()
+
+		for _, feed := range feeds {
+			closeFeed(t, feed)
+		}
+
+		doneWaiting := make(chan struct{})
+		go func() {
+			defer close(doneWaiting)
+			wg.Wait()
+		}()
+		select {
+		case <-doneWaiting:
+		case <-time.After(5 * time.Second):
+			t.Fatalf("timed out waiting for goroutines to finish")
+		}
+	}
+	// Sinkless testfeeds have some weird shutdown behaviours, so exclude them for now.
+	cdcTest(t, testFn, feedTestRestrictSinks("kafka", "pubsub", "webhook"))
+}
+
 func TestChangefeedEnrichedAvro(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/cmd/roachtest/tests/cdc.go

@@ -517,6 +517,7 @@ func makeDefaultFeatureFlags() cdcFeatureFlags {
 type feedArgs struct {
 	sinkType        sinkType
 	targets         []string
+	envelope        string
 	opts            map[string]string
 	assumeRole      string
 	tolerateErrors  bool
@@ -547,15 +548,15 @@ func (ct *cdcTester) newChangefeed(args feedArgs) changefeedJob {
 
 	targetsStr := strings.Join(args.targets, ", ")
 
+	if args.envelope == "" {
+		args.envelope = "wrapped"
+	}
+
 	feedOptions := make(map[string]string)
 	feedOptions["min_checkpoint_frequency"] = "'10s'"
+	feedOptions["envelope"] = args.envelope
 	if args.sinkType == cloudStorageSink || args.sinkType == webhookSink {
-		// Webhook and cloudstorage don't have a concept of keys and therefore
-		// require envelope=wrapped
-		feedOptions["envelope"] = "wrapped"
-
 		feedOptions["resolved"] = "'10s'"
-
 	} else {
 		feedOptions["resolved"] = ""
 	}
@@ -2311,6 +2312,40 @@ func registerCDC(r registry.Registry) {
 		Timeout:          1 * time.Hour,
 		Run:              runCDCMultipleSchemaChanges,
 	})
+	r.Add(registry.TestSpec{
+		Name:             "cdc/tpcc-100/10min/sink=kafka/envelope=enriched",
+		Owner:            registry.OwnerCDC,
+		Benchmark:        true,
+		Cluster:          r.MakeClusterSpec(4, spec.WorkloadNode(), spec.CPU(16)),
+		Leases:           registry.MetamorphicLeases,
+		CompatibleClouds: registry.AllClouds,
+		Suites:           registry.Suites(registry.Nightly),
+		Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
+			ct := newCDCTester(ctx, t, c)
+			defer ct.Close()
+
+			ct.runTPCCWorkload(tpccArgs{warehouses: 100, duration: "10m"})
+
+			feed := ct.newChangefeed(feedArgs{
+				sinkType: kafkaSink,
+				envelope: "enriched",
+				targets:  allTpccTargets,
+				kafkaArgs: kafkaFeedArgs{
+					validateOrder: true,
+				},
+				opts: map[string]string{
+					"initial_scan":        "'no'",
+					"updated":             "",
+					"enriched_properties": "source",
+				},
+			})
+			ct.runFeedLatencyVerifier(feed, latencyTargets{
+				initialScanLatency: 3 * time.Minute,
+				steadyLatency:      10 * time.Minute,
+			})
+			ct.waitForWorkload()
+		},
+	})
 }
 
 const (