From cd4a0336417cbf48051c113fe57c787a1ba3b19a Mon Sep 17 00:00:00 2001 From: Wang Qi Date: Wed, 16 Sep 2026 12:00:17 +0800 Subject: [PATCH] Fix: use component name instead of ID (#19700) --- internal/ingestion/pipeline/pipeline.go | 12 ++++++++---- internal/ingestion/task/debug_log_sink_test.go | 10 +++++----- 2 files changed, 13 insertions(+), 9 deletions(-) diff --git a/internal/ingestion/pipeline/pipeline.go b/internal/ingestion/pipeline/pipeline.go index 4a6740e8fa..db04962863 100644 --- a/internal/ingestion/pipeline/pipeline.go +++ b/internal/ingestion/pipeline/pipeline.go @@ -720,17 +720,21 @@ func (p *Pipeline) componentProgressCallback(ctx context.Context) runtime.Progre return nil } return func(ev runtime.ProgressEvent) { + componentName := ev.Component + if comp, ok := p.canvas.Components[ev.Component]; ok && comp.Obj.ComponentName != "" { + componentName = comp.Obj.ComponentName + } var msg string switch ev.Phase { case runtime.PhaseEnter: - msg = ev.Component + " Started" + msg = componentName + " Started" case runtime.PhaseExit: - msg = ev.Component + " Done" + msg = componentName + " Done" case runtime.PhaseError: if ev.Err != nil { - msg = ev.Component + ": " + ev.Err.Error() + msg = componentName + ": " + ev.Err.Error() } else { - msg = ev.Component + " Error" + msg = componentName + " Error" } } // Surface every component lifecycle event as a structured log line so diff --git a/internal/ingestion/task/debug_log_sink_test.go b/internal/ingestion/task/debug_log_sink_test.go index 9e6c820ebe..8b09733e8a 100644 --- a/internal/ingestion/task/debug_log_sink_test.go +++ b/internal/ingestion/task/debug_log_sink_test.go @@ -277,8 +277,8 @@ func (traceChunkComponent) Invoke(_ context.Context, _ *gorm.DB, _ map[string]an // wiring a dataflow debug run uses: NewPipelineFromDSL + WithProgressSink // (DebugLogSink) + Run. The canvas framework wraps each component in // realComponentBody → runtime.TrackProgress, which MUST emit BOTH PhaseEnter -// (" Started", progress 0) and PhaseExit (" Done", progress 1) on -// success. +// (" Started", progress 0) and PhaseExit +// (" Done", progress 1) on success. // // This locks the contract the front-end depends on: every component ends with a // Done line, not just a Started line. It is the regression guard for the @@ -333,7 +333,7 @@ func TestDebugLogSink_RealPipeline_EachComponentTraceHasStartedAndDone(t *testin // Every real component (a, b) must show BOTH Started (progress 0) and Done // (progress 1) within its own trace — proving TrackProgress emitted the full // enter→exit lifecycle, not just enter. - for _, comp := range []string{"a", "b"} { + for comp, componentName := range map[string]string{"a": compA, "b": compB} { var gotStart, gotDone bool for _, el := range arr { if el["component_id"] != comp { @@ -348,9 +348,9 @@ func TestDebugLogSink_RealPipeline_EachComponentTraceHasStartedAndDone(t *testin prog, _ := tm["progress"].(float64) msg, _ := tm["message"].(string) switch { - case prog == 0 && msg == comp+" Started": + case prog == 0 && msg == componentName+" Started": gotStart = true - case prog == 1 && msg == comp+" Done": + case prog == 1 && msg == componentName+" Done": gotDone = true } }