Merge #146476 #148683 #149439 #149483 #149488

146476: changefeedccl: fix split_column_families changefeeds r=asg0451 a=KeithCh

This fixes the following when creating changefeeds 
for tables that have multiple columns and with the 
option split_column_families, into sinks that have topics:

When using a Kafka V1 sink, we would always 
attempt to create a nonsense topic even if the
Kafka broker setting has
auto.create.topics.enable=false. This would cause
the changefeed to fail.

Release note(bug fix):
Remove unncessary Kafka topic creation that would
could cause changefeed start up to fail when using 
changefeed.new_kafka_sink_enabled=false.

Fixes: #128973
Epic: CRDB-41784

148683: sql: add infrastructure for new inspect job r=spilchen a=spilchen

This adds infrastructure for a new job that will handle data consistency checking (tentatively called inspect). It will be initially invoked via the `EXPERIMENTAL SCRUB` command and a new session variable (`enable_scrub_job`). It will eventually be invoked under its own SQL, but that will be handled separately.

I split this into 4 commits to make reviewing easier. I will squash them prior to merging to master.

## jobs: add stub implementation of inspect job

This commit introduces a stub for the new INSPECT job type. The Resumer doesn't do anything useful yet. It simply returns as a no-op.

Informs: #148289
Epic: CRDB-30356
Release note: None

## sql: support starting inspect job from SCRUB

This change adds support for initiating a INSPECT job when running SCRUB TABLE, controlled by the session variable enable_scrub_job.

When enable_scrub_job is true, SCRUB TABLE will schedule or start a consistency check job instead of running inline checks.

Informs: #148289
Epic: CRDB-30356
Release note: None

##  sql: add logic tests for inspect job via SCRUB

This change introduces logic tests for the new inspect job path when enable_scrub_job is enabled.

Informs: #148289
Epic: CRDB-30356
Release note: None

## sql: add unit test for SCRUB job execution semantics

This change adds a unit test (TestInspectJobImplicitTxnSemantics) to verify the behavior of inspect jobs started via EXPERIMENTAL SCRUB TABLE in implicit transactions.

Includes test hooks via InspectTestingKnobs to simulate and control job execution behavior.

Informs: #148289
Epic: CRDB-30356
Release note: None




149439: roachtest: capture metrics for index backfill in schemachange/bulkingest r=rafiss a=rafiss

This will allow us to make a roachperf chart to track changes over time.

informs #146571
Release note: None

149483: opentelemetry: add raftlog size metrics r=arulajmani a=pav-kv

Part of #141126

149488: tracing: fix tests for simple flight recorder r=angles-n-daemons a=dhartunian

We weren't returning errors correctly in SucceedsSoon. First failure would fail the test instead of retrying.

Resolves: #149450

Release note: None

Co-authored-by: Keith Chow <[email protected]>
Co-authored-by: Matt Spilchen <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>
Co-authored-by: Pavel Kalinnikov <[email protected]>
Co-authored-by: David Hartunian <[email protected]>

Author: 6 people

docs/generated/metrics/metrics.yaml

@@ -5086,6 +5086,114 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.currently_idle
+      exported_name: jobs_inspect_currently_idle
+      labeled_name: 'jobs{type: inspect, status: currently_idle}'
+      description: Number of inspect jobs currently considered Idle and can be freely shut down
+      y_axis_label: jobs
+      type: GAUGE
+      unit: COUNT
+      aggregation: AVG
+      derivative: NONE
+    - name: jobs.inspect.currently_paused
+      exported_name: jobs_inspect_currently_paused
+      labeled_name: 'jobs{name: inspect, status: currently_paused}'
+      description: Number of inspect jobs currently considered Paused
+      y_axis_label: jobs
+      type: GAUGE
+      unit: COUNT
+      aggregation: AVG
+      derivative: NONE
+    - name: jobs.inspect.currently_running
+      exported_name: jobs_inspect_currently_running
+      labeled_name: 'jobs{type: inspect, status: currently_running}'
+      description: Number of inspect jobs currently running in Resume or OnFailOrCancel state
+      y_axis_label: jobs
+      type: GAUGE
+      unit: COUNT
+      aggregation: AVG
+      derivative: NONE
+    - name: jobs.inspect.expired_pts_records
+      exported_name: jobs_inspect_expired_pts_records
+      labeled_name: 'jobs.expired_pts_records{type: inspect}'
+      description: Number of expired protected timestamp records owned by inspect jobs
+      y_axis_label: records
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.fail_or_cancel_completed
+      exported_name: jobs_inspect_fail_or_cancel_completed
+      labeled_name: 'jobs.fail_or_cancel{name: inspect, status: completed}'
+      description: Number of inspect jobs which successfully completed their failure or cancelation process
+      y_axis_label: jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.fail_or_cancel_failed
+      exported_name: jobs_inspect_fail_or_cancel_failed
+      labeled_name: 'jobs.fail_or_cancel{name: inspect, status: failed}'
+      description: Number of inspect jobs which failed with a non-retriable error on their failure or cancelation process
+      y_axis_label: jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.fail_or_cancel_retry_error
+      exported_name: jobs_inspect_fail_or_cancel_retry_error
+      labeled_name: 'jobs.fail_or_cancel{name: inspect, status: retry_error}'
+      description: Number of inspect jobs which failed with a retriable error on their failure or cancelation process
+      y_axis_label: jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.protected_age_sec
+      exported_name: jobs_inspect_protected_age_sec
+      labeled_name: 'jobs.protected_age_sec{type: inspect}'
+      description: The age of the oldest PTS record protected by inspect jobs
+      y_axis_label: seconds
+      type: GAUGE
+      unit: SECONDS
+      aggregation: AVG
+      derivative: NONE
+    - name: jobs.inspect.protected_record_count
+      exported_name: jobs_inspect_protected_record_count
+      labeled_name: 'jobs.protected_record_count{type: inspect}'
+      description: Number of protected timestamp records held by inspect jobs
+      y_axis_label: records
+      type: GAUGE
+      unit: COUNT
+      aggregation: AVG
+      derivative: NONE
+    - name: jobs.inspect.resume_completed
+      exported_name: jobs_inspect_resume_completed
+      labeled_name: 'jobs.resume{name: inspect, status: completed}'
+      description: Number of inspect jobs which successfully resumed to completion
+      y_axis_label: jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.resume_failed
+      exported_name: jobs_inspect_resume_failed
+      labeled_name: 'jobs.resume{name: inspect, status: failed}'
+      description: Number of inspect jobs which failed with a non-retriable error
+      y_axis_label: jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.resume_retry_error
+      exported_name: jobs_inspect_resume_retry_error
+      labeled_name: 'jobs.resume{name: inspect, status: retry_error}'
+      description: Number of inspect jobs which failed with a retriable error
+      y_axis_label: jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: jobs.key_visualizer.currently_idle
       exported_name: jobs_key_visualizer_currently_idle
       labeled_name: 'jobs{type: key_visualizer, status: currently_idle}'

pkg/BUILD.bazel

@@ -496,6 +496,7 @@ ALL_TESTS = [
     "//pkg/sql/idxrecommendations:idxrecommendations_test",
     "//pkg/sql/idxusage:idxusage_test",
     "//pkg/sql/importer:importer_test",
+    "//pkg/sql/inspect:inspect_test",
     "//pkg/sql/inverted:inverted_disallowed_imports_test",
     "//pkg/sql/inverted:inverted_test",
     "//pkg/sql/lex:lex_disallowed_imports_test",
@@ -2020,6 +2021,8 @@ GO_TARGETS = [
     "//pkg/sql/idxusage:idxusage_test",
     "//pkg/sql/importer:importer",
     "//pkg/sql/importer:importer_test",
+    "//pkg/sql/inspect:inspect",
+    "//pkg/sql/inspect:inspect_test",
     "//pkg/sql/inverted:inverted",
     "//pkg/sql/inverted:inverted_test",
     "//pkg/sql/isql:isql",

pkg/base/testing_knobs.go

@@ -32,6 +32,7 @@ type TestingKnobs struct {
 	JobsTestingKnobs               ModuleTestingKnobs
 	BackupRestore                  ModuleTestingKnobs
 	TTL                            ModuleTestingKnobs
+	Inspect                        ModuleTestingKnobs
 	SchemaTelemetry                ModuleTestingKnobs
 	Streaming                      ModuleTestingKnobs
 	UpgradeManager                 ModuleTestingKnobs

pkg/ccl/changefeedccl/changefeed_test.go

@@ -6711,15 +6711,15 @@ func TestChangefeedErrors(t *testing.T) {
 		`CREATE CHANGEFEED FOR foo INTO $1`, `experimental-sql://d/?confluent_schema_registry=foo&weird=bar`,
 	)
 
-	badHostErrRE := "client has run out of available brokers"
+	badHostErrRE := "(no such host|connection refused|network is unreachable)"
 	if KafkaV2Enabled.Get(&s.ClusterSettings().SV) {
 		badHostErrRE = "(unable to dial|unable to open connection to broker|lookup .* on .*: server misbehaving|connection refused)"
 	}
 
 	// Check unavailable kafka - bad dns.
 	longTimeoutSQLDB.ExpectErrWithTimeout(
 		t, badHostErrRE,
-		`CREATE CHANGEFEED FOR foo INTO 'kafka://nope'`,
+		`CREATE CHANGEFEED FOR foo INTO 'kafka://nope:9999'`,
 	)
 
 	// Check unavailable kafka - not running.
@@ -6731,7 +6731,7 @@ func TestChangefeedErrors(t *testing.T) {
 	// Test that a well-formed URI gets as far as unavailable kafka error.
 	longTimeoutSQLDB.ExpectErrWithTimeout(
 		t, badHostErrRE,
-		`CREATE CHANGEFEED FOR foo INTO 'kafka://nope/?tls_enabled=true&insecure_tls_skip_verify=true&topic_name=foo'`,
+		`CREATE CHANGEFEED FOR foo INTO 'kafka://nope:9999/?tls_enabled=true&insecure_tls_skip_verify=true&topic_name=foo'`,
 	)
 
 	// kafka_topic_prefix was referenced by an old version of the RFC, it's
@@ -6846,12 +6846,12 @@ func TestChangefeedErrors(t *testing.T) {
 	)
 	sqlDB.ExpectErrWithTimeout(
 		t, badHostErrRE,
-		`CREATE CHANGEFEED FOR foo INTO 'kafka://nope/' WITH kafka_sink_config='{"Flush": {"Messages": 100, "Frequency": "1s"}}'`,
+		`CREATE CHANGEFEED FOR foo INTO 'kafka://nope:9999/' WITH kafka_sink_config='{"Flush": {"Messages": 100, "Frequency": "1s"}}'`,
 	)
 	sqlDB.ExpectErrWithTimeout(
 		t, `this sink is incompatible with option webhook_client_timeout`,
 		`CREATE CHANGEFEED FOR foo INTO $1 WITH webhook_client_timeout='1s'`,
-		`kafka://nope/`,
+		`kafka://nope:9999/`,
 	)
 	// The avro format doesn't support key_in_value or topic_in_value yet.
 	sqlDB.ExpectErrWithTimeout(
@@ -7279,6 +7279,21 @@ func TestChangefeedDescription(t *testing.T) {
 	}
 }
 
+func TestChangefeedKafkaV1ConnectionError(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		KafkaV2Enabled.Override(context.Background(), &s.Server.ClusterSettings().SV, false)
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+		sqlDB.Exec(t, `CREATE TABLE foo(id int primary key, s string)`)
+		sqlDB.Exec(t, `INSERT INTO foo(id, s) VALUES (0, 'hello'), (1, null)`)
+		_, err := f.Feed(`CREATE CHANGEFEED FOR foo`)
+		require.ErrorContains(t, err, "client has run out of available brokers")
+	}
+	cdcTest(t, testFn, feedTestForceSink("kafka"), feedTestForceKafkaV1ConnectionCheck)
+}
+
 func TestChangefeedPanicRecovery(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/ccl/changefeedccl/helpers_test.go

@@ -776,6 +776,7 @@ type feedTestOptions struct {
 	debugUseAfterFinish          bool
 	clusterName                  string
 	locality                     roachpb.Locality
+	forceKafkaV1ConnectionCheck  bool
 }
 
 type feedTestOption func(opts *feedTestOptions)
@@ -795,6 +796,12 @@ var feedTestNoExternalConnection = func(opts *feedTestOptions) { opts.forceNoExt
 // has privileges to create changefeeds on tables in the default database `d` only.
 var feedTestUseRootUserConnection = func(opts *feedTestOptions) { opts.forceRootUserConnection = true }
 
+// feedTestForceKafkaV1ConnectionCheck is a feedTestOption that will force the connection check
+// inside Dial() when using a Kafka v1 sink.
+var feedTestForceKafkaV1ConnectionCheck = func(opts *feedTestOptions) {
+	opts.forceKafkaV1ConnectionCheck = true
+}
+
 var feedTestForceSink = func(sinkType string) feedTestOption {
 	return feedTestRestrictSinks(sinkType)
 }
@@ -1339,7 +1346,7 @@ func makeFeedFactoryWithOptions(
 	}
 	switch sinkType {
 	case "kafka":
-		f := makeKafkaFeedFactory(t, srvOrCluster, db)
+		f := makeKafkaFeedFactoryWithConnectionCheck(t, srvOrCluster, db, options.forceKafkaV1ConnectionCheck)
 		userDB, cleanup := getInitialDBForEnterpriseFactory(t, s, db, options)
 		f.(*kafkaFeedFactory).configureUserDB(userDB)
 		return f, func() { cleanup() }

pkg/ccl/changefeedccl/sink_kafka.go

@@ -74,6 +74,7 @@ type kafkaSinkKnobs struct {
 	OverrideClientInit              func(config *sarama.Config) (kafkaClient, error)
 	OverrideAsyncProducerFromClient func(kafkaClient) (sarama.AsyncProducer, error)
 	OverrideSyncProducerFromClient  func(kafkaClient) (sarama.SyncProducer, error)
+	BypassConnectionCheck           bool
 }
 
 var _ sarama.StdLogger = (*kafkaLogAdapter)(nil)
@@ -114,6 +115,8 @@ type kafkaClient interface {
 	Config() *sarama.Config
 	// Close closes kafka connection.
 	Close() error
+	// LeastLoadedBroker retrieves broker that has the least responses pending.
+	LeastLoadedBroker() *sarama.Broker
 }
 
 // kafkaSink emits to Kafka asynchronously. It is not concurrency-safe; all
@@ -264,10 +267,16 @@ func (s *kafkaSink) Dial() error {
 		return err
 	}
 
-	if err = client.RefreshMetadata(s.Topics()...); err != nil {
-		// Now that we do not fetch metadata for all topics by default, we try
-		// RefreshMetadata manually to check for any connection error.
-		return errors.CombineErrors(err, client.Close())
+	// Make sure the broker can be reached.
+	if broker := client.LeastLoadedBroker(); broker != nil {
+		if err := broker.Open(s.kafkaCfg); err != nil && !errors.Is(err, sarama.ErrAlreadyConnected) {
+			return errors.CombineErrors(err, client.Close())
+		}
+		if ok, err := broker.Connected(); !ok || err != nil {
+			return errors.CombineErrors(err, client.Close())
+		}
+	} else if !s.knobs.BypassConnectionCheck {
+		return errors.CombineErrors(errors.New("client has run out of available brokers"), client.Close())
 	}
 
 	producer, err := s.newAsyncProducer(client)

pkg/ccl/changefeedccl/sink_test.go

@@ -278,6 +278,7 @@ func makeTestKafkaSink(
 				client := &fakeKafkaClient{config}
 				return client, nil
 			},
+			BypassConnectionCheck: true,
 		},
 	}
 	err = s.Dial()

pkg/ccl/changefeedccl/testfeed_test.go

@@ -1745,6 +1745,10 @@ func (c *fakeKafkaClient) Close() error {
 	return nil
 }
 
+func (c *fakeKafkaClient) LeastLoadedBroker() *sarama.Broker {
+	return nil
+}
+
 func (c *fakeKafkaClient) Config() *sarama.Config {
 	return c.config
 }
@@ -1775,6 +1779,8 @@ func (p *asyncIgnoreCloseProducer) Close() error {
 type sinkKnobs struct {
 	// kafkaInterceptor is only valid for the v1 kafka sink.
 	kafkaInterceptor func(m *sarama.ProducerMessage, client kafkaClient) error
+	// BypassConnectionCheck is used for v1 kafka sink.
+	bypassKafkaV1ConnectionCheck bool
 }
 
 // fakeKafkaSink is a sink that arranges for fake kafka client and producer
@@ -1795,6 +1801,7 @@ func (s *fakeKafkaSink) Dial() error {
 		client := &fakeKafkaClient{config}
 		return client, nil
 	}
+	kafka.knobs.BypassConnectionCheck = s.knobs.bypassKafkaV1ConnectionCheck
 
 	kafka.knobs.OverrideAsyncProducerFromClient = func(client kafkaClient) (sarama.AsyncProducer, error) {
 		// The producer we give to kafka sink ignores close call.
@@ -1946,13 +1953,14 @@ func mustBeKafkaFeedFactory(f cdctest.TestFeedFactory) *kafkaFeedFactory {
 	}
 }
 
-// makeKafkaFeedFactory returns a TestFeedFactory implementation using the `kafka` uri.
-func makeKafkaFeedFactory(
-	t *testing.T, srvOrCluster interface{}, rootDB *gosql.DB,
+func makeKafkaFeedFactoryWithConnectionCheck(
+	t *testing.T, srvOrCluster interface{}, rootDB *gosql.DB, forceKafkaV1ConnectionCheck bool,
 ) cdctest.TestFeedFactory {
 	s, injectables := getInjectables(srvOrCluster)
 	return &kafkaFeedFactory{
-		knobs: &sinkKnobs{},
+		knobs: &sinkKnobs{
+			bypassKafkaV1ConnectionCheck: !forceKafkaV1ConnectionCheck,
+		},
 		enterpriseFeedFactory: enterpriseFeedFactory{
 			s:      s,
 			db:     rootDB,
@@ -1963,6 +1971,13 @@ func makeKafkaFeedFactory(
 	}
 }
 
+// makeKafkaFeedFactory returns a TestFeedFactory implementation using the `kafka` uri.
+func makeKafkaFeedFactory(
+	t *testing.T, srvOrCluster interface{}, rootDB *gosql.DB,
+) cdctest.TestFeedFactory {
+	return makeKafkaFeedFactoryWithConnectionCheck(t, srvOrCluster, rootDB, false)
+}
+
 func exprAsString(expr tree.Expr) (string, error) {
 	evalCtx := eval.NewTestingEvalContext(cluster.MakeTestingClusterSettings())
 	semaCtx := tree.MakeSemaContext(nil /* resolver */)