changefeedccl: add helper to detect changefeed errors

Add a test helper to detect changefeed errors
rather than just observing timeouts.

Epic: None
Release note: None

Author: asg0451

pkg/ccl/changefeedccl/alter_changefeed_test.go

@@ -514,7 +514,7 @@ func TestAlterChangefeedDropTargetAfterTableDrop(t *testing.T) {
 		})
 	}
 
-	cdcTest(t, testFn, feedTestEnterpriseSinks, feedTestNoExternalConnection)
+	cdcTest(t, testFn, feedTestEnterpriseSinks, feedTestNoExternalConnection, withAllowChangefeedErr("error is expected when dropping"))
 }
 
 func TestAlterChangefeedDropTargetFamily(t *testing.T) {

pkg/ccl/changefeedccl/changefeed_test.go

@@ -416,7 +416,7 @@ func TestRLSBlocking(t *testing.T) {
 		require.Contains(t, err.Error(), expErrSubstr)
 	}
 
-	cdcTest(t, testFn)
+	cdcTest(t, testFn, withAllowChangefeedErr("expects terminal error"))
 }
 
 func TestToJSONAsChangefeed(t *testing.T) {
@@ -744,7 +744,7 @@ func TestChangefeedSendError(t *testing.T) {
 			`foo: [3]->{"after": {"a": 3}}`,
 			`foo: [4]->{"after": {"a": 4}}`,
 		})
-	}, feedTestEnterpriseSinks)
+	}, feedTestEnterpriseSinks, withAllowChangefeedErr("injects error"))
 }
 
 func TestChangefeedBasicConfluentKafka(t *testing.T) {
@@ -837,9 +837,9 @@ func TestChangefeedTenants(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	kvServer, kvSQLdb, cleanup := startTestFullServer(t, feedTestOptions{argsFn: func(args *base.TestServerArgs) {
+	kvServer, kvSQLdb, cleanup := startTestFullServer(t, makeOptions(t, withArgsFn(func(args *base.TestServerArgs) {
 		args.ExternalIODirConfig.DisableOutbound = true
-	}})
+	})))
 	defer cleanup()
 
 	tenantArgs := base.TestTenantArgs{
@@ -2404,7 +2404,7 @@ func TestChangefeedLaggingSpanCheckpointing(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	s, db, stopServer := startTestFullServer(t, feedTestOptions{})
+	s, db, stopServer := startTestFullServer(t, makeOptions(t, feedTestNoTenants))
 	defer stopServer()
 	sqlDB := sqlutils.MakeSQLRunner(db)
 
@@ -3381,7 +3381,7 @@ func TestChangefeedEachColumnFamily(t *testing.T) {
 		}
 	}
 
-	cdcTest(t, testFn)
+	cdcTest(t, testFn, withAllowChangefeedErr("expects terminal error"))
 }
 
 func TestChangefeedSingleColumnFamily(t *testing.T) {
@@ -3475,12 +3475,12 @@ func TestChangefeedSingleColumnFamilySchemaChanges(t *testing.T) {
 
 		// Removing all columns in a watched family fails the feed
 		waitForSchemaChange(t, sqlDB, `ALTER TABLE foo DROP column c`)
-		requireErrorSoon(context.Background(), t, fooRest,
+		requireTerminalErrorSoon(context.Background(), t, fooRest,
 			regexp.MustCompile(`CHANGEFEED targeting nonexistent or removed column family rest of table foo`))
 	}
 
 	runWithAndWithoutRegression141453(t, testFn, func(t *testing.T, testFn cdcTestFn) {
-		cdcTest(t, testFn)
+		cdcTest(t, testFn, withAllowChangefeedErr("expects terminal error"))
 	}, withMaybeUseLegacySchemaChanger())
 }
 
@@ -3811,9 +3811,9 @@ func TestChangefeedCustomKey(t *testing.T) {
 			`foo: ["dog"]->{"after": {"a": 1, "b": "dog", "c": "zebra"}}`,
 		})
 		sqlDB.Exec(t, `ALTER TABLE foo RENAME COLUMN b to b2`)
-		requireErrorSoon(context.Background(), t, foo, regexp.MustCompile(`required column b not present`))
+		requireTerminalErrorSoon(context.Background(), t, foo, regexp.MustCompile(`required column b not present`))
 	}
-	cdcTest(t, testFn, feedTestForceSink("kafka"))
+	cdcTest(t, testFn, feedTestForceSink("kafka"), withAllowChangefeedErr("expects error"))
 }
 
 // Reproduce issue for #114196. This test verifies that changefeed with custom
@@ -5184,12 +5184,14 @@ func TestChangefeedOutputTopics(t *testing.T) {
 	})
 }
 
-// requireErrorSoon polls for the test feed for an error and asserts that
-// the error matches the provided regex.
-func requireErrorSoon(
+// requireTerminalErrorSoon polls for the test feed for an error and asserts
+// that the error matches the provided regex. This can either be a terminal
+// error or an error encountered while parsing messages and doing testfeed
+// things.
+func requireTerminalErrorSoon(
 	ctx context.Context, t *testing.T, f cdctest.TestFeed, errRegex *regexp.Regexp,
 ) {
-	err := timeutil.RunWithTimeout(ctx, "requireErrorSoon", 30*time.Second, func(ctx context.Context) error {
+	err := timeutil.RunWithTimeout(ctx, "requireTerminalErrorSoon", 30*time.Second, func(ctx context.Context) error {
 		for {
 			select {
 			case <-ctx.Done():
@@ -5235,9 +5237,9 @@ func TestChangefeedFailOnTableOffline(t *testing.T) {
 			`for_import: [0]->{"after": {"a": 0, "b": null}}`,
 		})
 		sqlDB.Exec(t, `IMPORT INTO for_import CSV DATA ($1)`, dataSrv.URL)
-		requireErrorSoon(context.Background(), t, forImport,
+		requireTerminalErrorSoon(context.Background(), t, forImport,
 			regexp.MustCompile(`CHANGEFEED cannot target offline table: for_import \(offline reason: "importing"\)`))
-	})
+	}, withAllowChangefeedErr("expects terminal error"))
 
 	cdcTestNamedWithSystem(t, "reverted import fails changefeed with earlier cursor", func(t *testing.T, s TestServerWithSystem, f cdctest.TestFeedFactory) {
 		sysSQLDB := sqlutils.MakeSQLRunner(s.SystemDB)
@@ -5275,9 +5277,9 @@ func TestChangefeedFailOnTableOffline(t *testing.T) {
 		// Changefeed should fail regardless
 		forImport := feed(t, f, `CREATE CHANGEFEED FOR for_import WITH cursor=$1`, start)
 		defer closeFeed(t, forImport)
-		requireErrorSoon(context.Background(), t, forImport,
+		requireTerminalErrorSoon(context.Background(), t, forImport,
 			regexp.MustCompile(`CHANGEFEED cannot target offline table: for_import \(offline reason: "importing"\)`))
-	})
+	}, withAllowChangefeedErr("expects terminal error"))
 }
 
 func TestChangefeedRestartMultiNode(t *testing.T) {
@@ -5987,7 +5989,7 @@ func TestChangefeedTruncateOrDrop(t *testing.T) {
 		assertFailuresCounter(t, metrics, 3)
 	}
 
-	cdcTest(t, testFn, feedTestEnterpriseSinks)
+	cdcTest(t, testFn, feedTestEnterpriseSinks, withAllowChangefeedErr("expects errors"))
 	// will sometimes fail, non deterministic
 }
 
@@ -6183,7 +6185,7 @@ func TestChangefeedRetryableError(t *testing.T) {
 		}
 	}
 
-	cdcTest(t, testFn, feedTestEnterpriseSinks)
+	cdcTest(t, testFn, feedTestEnterpriseSinks, withAllowChangefeedErr("expects error"))
 }
 
 func TestChangefeedJobUpdateFailsIfNotClaimed(t *testing.T) {
@@ -6378,7 +6380,7 @@ func TestChangefeedDataTTL(t *testing.T) {
 	// timestamp before beginning their backfill.
 	// TODO(samiskin): Tenant test disabled because this test requires
 	// forceTableGC which doesn't work on tenants
-	cdcTestWithSystem(t, testFn, feedTestForceSink("sinkless"), feedTestNoTenants)
+	cdcTestWithSystem(t, testFn, feedTestForceSink("sinkless"), feedTestNoTenants, withAllowChangefeedErr("expects batch ts gc error"))
 }
 
 // TestChangefeedOutdatedCursor ensures that create changefeeds fail with an
@@ -6418,7 +6420,7 @@ func TestChangefeedCursorAgeWarning(t *testing.T) {
 	}
 
 	testutils.RunValues(t, "cursor age", cursorAges, func(t *testing.T, cursorAge time.Duration) {
-		s, stopServer := makeServer(t)
+		s, stopServer := makeServer(t, withAllowChangefeedErr("expects batch ts gc error"))
 		defer stopServer()
 		knobs := s.TestingKnobs.
 			DistSQL.(*execinfra.TestingKnobs).
@@ -6540,7 +6542,7 @@ func TestChangefeedSchemaTTL(t *testing.T) {
 
 	// TODO(samiskin): tenant tests skipped because of forceTableGC not working
 	// with a ApplicationLayerInterface
-	cdcTestWithSystem(t, testFn, feedTestNoTenants)
+	cdcTestWithSystem(t, testFn, feedTestNoTenants, withAllowChangefeedErr("expects batch ts gc error"))
 }
 
 func TestChangefeedErrors(t *testing.T) {
@@ -7215,7 +7217,7 @@ func TestChangefeedDescription(t *testing.T) {
 
 	// Intentionally don't use the TestFeedFactory because we want to
 	// control the placeholders.
-	s, stopServer := makeServer(t)
+	s, stopServer := makeServerWithOptions(t, makeOptions(t, withAllowChangefeedErr("create strange changefeeds that don't actually run")))
 	defer stopServer()
 
 	sqlDB := sqlutils.MakeSQLRunner(s.DB)
@@ -7335,7 +7337,7 @@ func TestChangefeedPanicRecovery(t *testing.T) {
 		defer closeFeed(t, foo)
 		err := waitForFeedErr(t, foo, 2*time.Minute)
 		require.ErrorContains(t, err, "error evaluating CDC expression", "expected panic recovery while evaluating WHERE clause")
-	}, feedTestAdditionalSystemPrivs("REPAIRCLUSTER"))
+	}, feedTestAdditionalSystemPrivs("REPAIRCLUSTER"), withAllowChangefeedErr("expects error"))
 
 	// Check that all panics while evaluating the SELECT clause in an expression are recovered from.
 	// NB: REPAIRCLUSTER is required to use crdb_internal.force_panic.
@@ -7347,7 +7349,7 @@ func TestChangefeedPanicRecovery(t *testing.T) {
 		defer closeFeed(t, foo)
 		err := waitForFeedErr(t, foo, 2*time.Minute)
 		require.ErrorContains(t, err, "error evaluating CDC expression", "expected panic recovery while evaluating SELECT clause")
-	}, feedTestAdditionalSystemPrivs("REPAIRCLUSTER"))
+	}, feedTestAdditionalSystemPrivs("REPAIRCLUSTER"), withAllowChangefeedErr("expects error"))
 }
 
 func TestChangefeedPauseUnpause(t *testing.T) {
@@ -7822,7 +7824,7 @@ func TestChangefeedHandlesDrainingNodes(t *testing.T) {
 	// We use feedTestUseRootUserConnection to prevent the
 	// feed factory from trying to create a test user. Because the registry is draining, creating the test user
 	// will fail and the test will fail prematurely.
-	f, closeSink := makeFeedFactory(t, randomSinkType(feedTestEnterpriseSinks), tc.Server(1), tc.ServerConn(0),
+	f, closeSink := makeFeedFactory(t, randomSinkType(t, feedTestEnterpriseSinks), tc.Server(1), tc.ServerConn(0),
 		feedTestUseRootUserConnection)
 	defer closeSink()
 
@@ -7859,7 +7861,7 @@ func TestChangefeedHandlesRollingRestart(t *testing.T) {
 
 	const numNodes = 4
 
-	opts := makeOptions()
+	opts := makeOptions(t)
 	opts.forceRootUserConnection = true
 	defer addCloudStorageOptions(t, &opts)()
 
@@ -8079,7 +8081,7 @@ func TestChangefeedTimelyResolvedTimestampUpdatePostRollingRestart(t *testing.T)
 	skip.UnderDeadlock(t)
 	skip.UnderRace(t)
 
-	opts := makeOptions()
+	opts := makeOptions(t)
 	defer addCloudStorageOptions(t, &opts)()
 	opts.forceRootUserConnection = true
 	defer changefeedbase.TestingSetDefaultMinCheckpointFrequency(testSinkFlushFrequency)()
@@ -8180,7 +8182,7 @@ func TestChangefeedPropagatesTerminalError(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	opts := makeOptions()
+	opts := makeOptions(t)
 	defer addCloudStorageOptions(t, &opts)()
 	defer changefeedbase.TestingSetDefaultMinCheckpointFrequency(testSinkFlushFrequency)()
 	defer testingUseFastRetry()()
@@ -9103,7 +9105,7 @@ func TestChangefeedOrderingWithErrors(t *testing.T) {
 
 	// only used for webhook sink for now since it's the only testfeed where
 	// we can control the ordering of errors
-	cdcTest(t, testFn, feedTestForceSink("webhook"), feedTestNoExternalConnection)
+	cdcTest(t, testFn, feedTestForceSink("webhook"), feedTestNoExternalConnection, withAllowChangefeedErr("expects error"))
 }
 
 func TestChangefeedOnErrorOption(t *testing.T) {
@@ -9198,7 +9200,7 @@ func TestChangefeedOnErrorOption(t *testing.T) {
 		})
 	}
 
-	cdcTest(t, testFn, feedTestEnterpriseSinks)
+	cdcTest(t, testFn, feedTestEnterpriseSinks, withAllowChangefeedErr("expects error"))
 }
 
 func TestDistSenderRangeFeedPopulatesVirtualTable(t *testing.T) {
@@ -9650,7 +9652,7 @@ func TestChangefeedInvalidPredicate(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	_, db, stopServer := startTestFullServer(t, feedTestOptions{})
+	_, db, stopServer := startTestFullServer(t, makeOptions(t, feedTestNoTenants))
 	defer stopServer()
 	sqlDB := sqlutils.MakeSQLRunner(db)
 
@@ -9927,7 +9929,11 @@ func TestChangefeedPredicateWithSchemaChange(t *testing.T) {
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			cdcTest(t, testFn(tc), feedTestEnterpriseSinks)
+			testOpts := []feedTestOption{feedTestEnterpriseSinks}
+			if tc.expectErr != "" {
+				testOpts = append(testOpts, withAllowChangefeedErr(tc.expectErr))
+			}
+			cdcTest(t, testFn(tc), testOpts...)
 		})
 	}
 }
@@ -10185,7 +10191,7 @@ func TestChangefeedMultiPodTenantPlanning(t *testing.T) {
 	// Ensure both pods can be assigned work
 	waitForTenantPodsActive(t, tenant1Server, 2)
 
-	feedFactory, cleanupSink := makeFeedFactory(t, randomSinkType(feedTestEnterpriseSinks), tenant1Server, tenant1DB)
+	feedFactory, cleanupSink := makeFeedFactory(t, randomSinkType(t, feedTestEnterpriseSinks), tenant1Server, tenant1DB)
 	defer cleanupSink()
 
 	// Run a changefeed across two tables to guarantee multiple spans that can be spread across the aggregators
@@ -10316,7 +10322,7 @@ func TestChangefeedFailedTelemetryLogs(t *testing.T) {
 	}
 
 	t.Run(`connection_closed`, func(t *testing.T) {
-		s, stopServer := makeServer(t)
+		s, stopServer := makeServer(t, withAllowChangefeedErr("expects error"))
 		defer stopServer()
 
 		sqlDB := sqlutils.MakeSQLRunner(s.DB)
@@ -10350,7 +10356,7 @@ func TestChangefeedFailedTelemetryLogs(t *testing.T) {
 		failLogs := waitForLogs(t, beforeCreate)
 		require.Equal(t, 1, len(failLogs))
 		require.Equal(t, failLogs[0].FailureType, changefeedbase.UserInput)
-	}, feedTestEnterpriseSinks)
+	}, feedTestEnterpriseSinks, withAllowChangefeedErr("expects error"))
 
 	cdcTestNamed(t, "unknown_error", func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
 		sqlDB := sqlutils.MakeSQLRunner(s.DB)
@@ -10375,7 +10381,7 @@ func TestChangefeedFailedTelemetryLogs(t *testing.T) {
 		require.Equal(t, failLogs[0].FailureType, changefeedbase.UnknownError)
 		require.Contains(t, []string{`gcpubsub`, `external`}, failLogs[0].SinkType)
 		require.Equal(t, failLogs[0].NumTables, int32(1))
-	}, feedTestForceSink("pubsub"))
+	}, feedTestForceSink("pubsub"), withAllowChangefeedErr("expects error"))
 }
 
 func TestChangefeedCanceledTelemetryLogs(t *testing.T) {
@@ -10632,7 +10638,7 @@ func TestChangefeedKafkaMessageTooLarge(t *testing.T) {
 		}
 	}
 
-	cdcTest(t, testFn, feedTestForceSink(`kafka`))
+	cdcTest(t, testFn, feedTestForceSink(`kafka`), withAllowChangefeedErr("expects kafka error"))
 }
 
 // Regression for #85902.
@@ -11003,7 +11009,7 @@ func TestHighwaterDoesNotRegressOnRetry(t *testing.T) {
 		case <-doneCh:
 		}
 	}
-	cdcTest(t, testFn, feedTestEnterpriseSinks)
+	cdcTest(t, testFn, feedTestEnterpriseSinks, withAllowChangefeedErr("injects error"))
 }
 
 // TestChangefeedPubsubResolvedMessages tests that the pubsub sink emits
@@ -11786,7 +11792,9 @@ func TestCloudstorageParallelCompression(t *testing.T) {
 	const numFeedsEach = 10
 
 	testutils.RunValues(t, "compression", []string{"zstd", "gzip"}, func(t *testing.T, compression string) {
-		s, cleanup := makeServer(t)
+		opts := makeOptions(t)
+		opts.externalIODir = t.TempDir()
+		s, cleanup := makeServerWithOptions(t, opts)
 		defer cleanup()
 
 		sqlDB := sqlutils.MakeSQLRunner(s.DB)

pkg/ccl/changefeedccl/encoder_test.go

@@ -726,7 +726,10 @@ func TestAvroSchemaNaming(t *testing.T) {
 
 	}
 
-	cdcTest(t, testFn, feedTestForceSink("kafka"), feedTestUseRootUserConnection)
+	// TODO(#150537): This test sometimes encounters errors like "CHANGEFEED
+	// created on a table with a single column family (drivers) cannot now
+	// target a table with 2 families". Why?
+	cdcTest(t, testFn, feedTestForceSink("kafka"), feedTestUseRootUserConnection, withAllowChangefeedErr("inexplicable errors"))
 }
 
 func TestAvroSchemaNamespace(t *testing.T) {
@@ -879,7 +882,7 @@ func TestAvroMigrateToUnsupportedColumn(t *testing.T) {
 		}
 	}
 
-	cdcTest(t, testFn, feedTestForceSink("kafka"))
+	cdcTest(t, testFn, feedTestForceSink("kafka"), withAllowChangefeedErr("checks error manually"))
 }
 
 func TestAvroLedger(t *testing.T) {