Merge #144161

144161: changefeedccl: deflake TestChangefeedContinuousTelemetryOnTermination r=andyyang890 a=KeithCh

There was an attempt to deflake this test in #135961
but we were still seeing this occasionally.
This change expands on that attempt.

The main cause is from a transient KV server error
that occurs when we cancel a changefeed. The
error causes a new changefeed aggregator to
be started, which in turn creates a new
telemetryMetricsRecorder instance. Since the
lastEmitTime field in telemetryMetricsRecorder is
initialized to 0, it emits a
changefeed_emitted_bytes event at the very next
maybeFlushLogs() call. A counter is added to
detect when this transient error occurs, and 
prevents the second logger from emitting any 
logs.

Resolves #143700

Release note: None

Co-authored-by: Keith Chow <[email protected]>

Author: craig[bot]

pkg/ccl/changefeedccl/changefeed_test.go

@@ -7553,6 +7553,7 @@ func TestChangefeedContinuousTelemetry(t *testing.T) {
 
 type testTelemetryLogger struct {
 	telemetryLogger
+	id                      int32
 	afterIncEmittedCounters func(numMessages int, numBytes int)
 }
 
@@ -7563,6 +7564,18 @@ func (t *testTelemetryLogger) incEmittedCounters(numMessages int, numBytes int)
 	t.afterIncEmittedCounters(numMessages, numBytes)
 }
 
+func (t *testTelemetryLogger) maybeFlushLogs() {
+	if t.id == 1 {
+		t.telemetryLogger.maybeFlushLogs()
+	}
+}
+
+func (t *testTelemetryLogger) close() {
+	if t.id == 1 {
+		t.telemetryLogger.close()
+	}
+}
+
 func TestChangefeedContinuousTelemetryOnTermination(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -7583,6 +7596,7 @@ func TestChangefeedContinuousTelemetryOnTermination(t *testing.T) {
 			}
 			return nil
 		}
+		var numPeriodicTelemetryLogger atomic.Int32
 		// Synchronization to prevent a race between the changefeed closing
 		// and the telemetry logger getting emitted counts after messages
 		// have been emitted to the sink.
@@ -7595,6 +7609,7 @@ func TestChangefeedContinuousTelemetryOnTermination(t *testing.T) {
 						seen.Store(true)
 					}
 				},
+				id: numPeriodicTelemetryLogger.Add(1),
 			}
 		}
 
@@ -7623,7 +7638,14 @@ func TestChangefeedContinuousTelemetryOnTermination(t *testing.T) {
 
 		// Close the changefeed and ensure logs were created after closing.
 		require.NoError(t, foo.Close())
-		verifyLogsWithEmittedBytesAndMessages(t, jobID, afterFirstLog.UnixNano(), interval.Nanoseconds(), true /* closing */)
+
+		if numPeriodicTelemetryLogger.Load() > 1 {
+			t.Log("transient error")
+		}
+
+		verifyLogsWithEmittedBytesAndMessages(
+			t, jobID, afterFirstLog.UnixNano(), interval.Nanoseconds(), true, /* closing */
+		)
 	}
 
 	cdcTest(t, testFn)