Merge #125721

125721: changefeedccl: add more logging around checkpointing r=rharding6373 a=andyyang890 This patch adds a log message that will log checkpoints that are persisted to a changefeed job's progress during normal operation. It also adds a log message to log the checkpoint that is read during planning, which will be sent to aggregators to restore their progress. These messages are useful for debugging checkpointing-related problems. Epic: CRDB-37337 Release note: None Co-authored-by: Andy Yang <[email protected]>
cockroachdb · Jun 17, 2024 · efc894b · efc894b
2 parents efed5b6 + 8e2531d
commit efc894b
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 1 deletion.
diff --git a/pkg/ccl/changefeedccl/changefeed_dist.go b/pkg/ccl/changefeedccl/changefeed_dist.go
@@ -456,6 +456,9 @@ func makePlan(
 			aggregatorCheckpoint.Spans = checkpoint.Spans
 			aggregatorCheckpoint.Timestamp = checkpoint.Timestamp
 		}
+		if log.V(2) {
+			log.Infof(ctx, "aggregator checkpoint: %s", aggregatorCheckpoint)
+		}
 
 		aggregatorSpecs := make([]*execinfrapb.ChangeAggregatorSpec, len(spanPartitions))
 		for i, sp := range spanPartitions {

diff --git a/pkg/ccl/changefeedccl/changefeed_processors.go b/pkg/ccl/changefeedccl/changefeed_processors.go
@@ -1682,7 +1682,7 @@ func (cf *changeFrontier) checkpointJobProgress(
 			}
 
 			if updateRunStatus {
-				md.Progress.RunningStatus = fmt.Sprintf("running: resolved=%s", frontier)
+				progress.RunningStatus = fmt.Sprintf("running: resolved=%s", frontier)
 			}
 
 			ju.UpdateProgress(progress)
@@ -1699,6 +1699,9 @@ func (cf *changeFrontier) checkpointJobProgress(
 		}); err != nil {
 			return false, err
 		}
+		if log.V(2) {
+			log.Infof(cf.Ctx(), "change frontier persisted highwater=%s and checkpoint=%s", frontier, checkpoint)
+		}
 	}
 
 	cf.localState.SetHighwater(frontier)