Skip to content

[FSE-1855] Read Flink statement warnings from status.warnings #3419

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Ramin Gharib (raminqaf) wants to merge 2 commits into
base: main
Choose a base branch
from
Open

[FSE-1855] Read Flink statement warnings from status.warnings #3419

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b729b60
[FSE-1855] Read Flink statement warnings from status.warnings
raminqaf Jul 30, 2026
433a79b
[FSE-1855] Drop the warnings struct field comment
raminqaf Aug 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 25 additions & 8 deletions internal/flink/command_statement.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,17 +8,34 @@ import (
cmfsdk "github.com/confluentinc/cmf-sdk-go/v1"

"github.com/confluentinc/cli/v4/pkg/config"
"github.com/confluentinc/cli/v4/pkg/flink/types"
"github.com/confluentinc/cli/v4/pkg/output"
)

// printStatementWarnings renders warnings below the table, on stderr so that stdout stays the
// command's data. Serialized output already carries them in the warnings field.
func printStatementWarnings(cmd *cobra.Command, warnings []types.StatementWarning) {
if output.GetFormat(cmd) != output.Human {
return
}

if block := types.FormatStatementWarnings(warnings); block != "" {
output.ErrPrintln(false, "")
output.ErrPrintln(false, block)
output.ErrPrintln(false, "")
}
Comment on lines +18 to +26

@sgagniere Steven Gagniere (sgagniere) Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would it work for the ticket to log these to -v (warn), at least for the non-shell outputs?

It's only in the human readable output so it's not necessarily an issue, but it does deviate from the usual pattern for CLI outputs.

If we do want these to always be displayed, let's write it to stderr instead.

@raminqaf Ramin Gharib (raminqaf) Jul 31, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point on the convention, moved to stderr for the non-shell commands. stdout is now just the table, so redirecting or piping gives you clean data, and -o json/-o yaml are untouched since they carry the warnings in the payload itself.
I left the shell on stdout. Everything there is already conversational output on stdout, including the status messages and the old Details: banner, so splitting only the warnings onto stderr would be inconsistent with the surrounding lines.
On -v: I'd rather not hide them behind a verbosity flag. The requirement for this feature is that all warnings are always shown, which is the whole reason for moving them out of status.detail, so making them opt-in would undo that.
Goldens are unchanged since the integration harness captures combined output.

}

type statementOut struct {
CreationDate time.Time `human:"Creation Date" serialized:"creation_date"`
Name string `human:"Name" serialized:"name"`
Statement string `human:"Statement" serialized:"statement"`
ComputePool string `human:"Compute Pool,omitempty" serialized:"compute_pool,omitempty"`
Status string `human:"Status" serialized:"status"`
StatusDetail string `human:"Status Detail,omitempty" serialized:"status_detail,omitempty"`
LatestOffsets map[string]string `human:"Latest Offsets" serialized:"latest_offsets"`
LatestOffsetsTimestamp *time.Time `human:"Latest Offsets Timestamp" serialized:"latest_offsets_timestamp"`
CreationDate time.Time `human:"Creation Date" serialized:"creation_date"`
Name string `human:"Name" serialized:"name"`
Statement string `human:"Statement" serialized:"statement"`
ComputePool string `human:"Compute Pool,omitempty" serialized:"compute_pool,omitempty"`
Status string `human:"Status" serialized:"status"`
StatusDetail string `human:"Status Detail,omitempty" serialized:"status_detail,omitempty"`
Warnings []types.StatementWarning `human:"-" serialized:"warnings,omitempty"`
LatestOffsets map[string]string `human:"Latest Offsets" serialized:"latest_offsets"`
LatestOffsetsTimestamp *time.Time `human:"Latest Offsets Timestamp" serialized:"latest_offsets_timestamp"`
}

type statementOutOnPrem struct {
Expand Down
12 changes: 10 additions & 2 deletions internal/flink/command_statement_create.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@
return cmd
}

func (c *command) statementCreate(cmd *cobra.Command, args []string) error {

Check failure on line 57 in internal/flink/command_statement_create.go

View check run for this annotation

SonarQube-Confluent / SonarQube Code Analysis

Refactor this method to reduce its Cognitive Complexity from 38 to the 15 allowed. 

[S3776] Cognitive Complexity of functions should not be too high
 See more on https://sonarqube.confluent.io/project/issues?id=cli&pullRequest=3419&issues=3702533e-fcab-4dfd-89be-3da79a938b3f&open=3702533e-fcab-4dfd-89be-3da79a938b3f
environmentId, err := c.Context.EnvironmentId()
if err != nil {
return err
Expand Down Expand Up @@ -173,6 +173,8 @@
}
}

warnings := types.NewStatementWarnings(statement.Status.GetWarnings())
Comment thread
raminqaf marked this conversation as resolved.

table := output.NewTable(cmd)
table.Add(&statementOut{
CreationDate: statement.Metadata.GetCreatedAt(),
Expand All @@ -181,7 +183,13 @@
ComputePool: statement.Spec.GetComputePoolId(),
Status: statement.Status.GetPhase(),
StatusDetail: statement.Status.GetDetail(),
Warnings: warnings,
})
table.Filter([]string{"CreationDate", "Name", "Statement", "ComputePool", "Status", "StatusDetail"})
return table.Print()
table.Filter([]string{"CreationDate", "Name", "Statement", "ComputePool", "Status", "StatusDetail", "Warnings"})
if err := table.Print(); err != nil {
return err
}

printStatementWarnings(cmd, warnings)

@channingdong Channing Dong (channingdong) Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are the warnings immediately available after the create of the statement if we decide not to wait?

@raminqaf Ramin Gharib (raminqaf) Aug 6, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes. --wait defaults to false, and I verified against real devel that the warnings come back on the create response itself. A statement created without --wait printed its MISSING_WINDOW_START_END warning immediately. Validation-time warnings are attached before the statement leaves PENDING.

return nil
}
32 changes: 21 additions & 11 deletions internal/flink/command_statement_describe.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,20 +8,22 @@ import (
flinkgatewayv1 "github.com/confluentinc/ccloud-sdk-go-v2/flink-gateway/v1"

pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
"github.com/confluentinc/cli/v4/pkg/flink/types"
"github.com/confluentinc/cli/v4/pkg/output"
)

type describeStatementOut struct {
CreationDate time.Time `human:"Creation Date" serialized:"creation_date"`
Name string `human:"Name" serialized:"name"`
Statement string `human:"Statement" serialized:"statement"`
ComputePool string `human:"Compute Pool" serialized:"compute_pool"`
Status string `human:"Status" serialized:"status"`
StatusDetail string `human:"Status Detail,omitempty" serialized:"status_detail,omitempty"`
LatestOffsets map[string]string `human:"Latest Offsets" serialized:"latest_offsets"`
LatestOffsetsTimestamp *time.Time `human:"Latest Offsets Timestamp" serialized:"latest_offsets_timestamp"`
Properties map[string]string `human:"Properties" serialized:"properties"`
Principal string `human:"Principal" serialized:"principal"`
CreationDate time.Time `human:"Creation Date" serialized:"creation_date"`
Name string `human:"Name" serialized:"name"`
Statement string `human:"Statement" serialized:"statement"`
ComputePool string `human:"Compute Pool" serialized:"compute_pool"`
Status string `human:"Status" serialized:"status"`
StatusDetail string `human:"Status Detail,omitempty" serialized:"status_detail,omitempty"`
Warnings []types.StatementWarning `human:"-" serialized:"warnings,omitempty"`
LatestOffsets map[string]string `human:"Latest Offsets" serialized:"latest_offsets"`
LatestOffsetsTimestamp *time.Time `human:"Latest Offsets Timestamp" serialized:"latest_offsets_timestamp"`
Properties map[string]string `human:"Properties" serialized:"properties"`
Principal string `human:"Principal" serialized:"principal"`
}

func (c *command) newStatementDescribeCommand() *cobra.Command {
Expand Down Expand Up @@ -58,6 +60,8 @@ func (c *command) statementDescribe(cmd *cobra.Command, args []string) error {
return err
}

warnings := types.NewStatementWarnings(statement.Status.GetWarnings())

Comment thread
raminqaf marked this conversation as resolved.
table := output.NewTable(cmd)
table.Add(&describeStatementOut{
CreationDate: statement.Metadata.GetCreatedAt(),
Expand All @@ -66,10 +70,16 @@ func (c *command) statementDescribe(cmd *cobra.Command, args []string) error {
ComputePool: statement.Spec.GetComputePoolId(),
Status: statement.Status.GetPhase(),
StatusDetail: statement.Status.GetDetail(),
Warnings: warnings,

@channingdong Channing Dong (channingdong) Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This means having the warnings in both regular table printout, as well as an additional rendered warnings below the table, just want to make sure we want this duplicated information here.

@raminqaf Ramin Gharib (raminqaf) Aug 6, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These are never both shown. Warnings is tagged human:"-" so it is excluded from the human table, and printStatementWarnings returns early unless the format is human. So human output is table plus the stderr block, and -o json/-o yaml is the payload field only. Matches the summary you posted in the thread.

LatestOffsets: statement.Status.GetLatestOffsets(),
LatestOffsetsTimestamp: flinkgatewayv1.PtrTime(statement.Status.GetLatestOffsetsTimestamp()),
Properties: statement.Spec.GetProperties(),
Principal: statement.Spec.GetPrincipal(),
})
return table.Print()
if err := table.Print(); err != nil {
return err
}

printStatementWarnings(cmd, warnings)
return nil
}
2 changes: 2 additions & 0 deletions internal/flink/command_statement_list.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
"github.com/confluentinc/cli/v4/pkg/errors"
"github.com/confluentinc/cli/v4/pkg/examples"
"github.com/confluentinc/cli/v4/pkg/flink/types"
"github.com/confluentinc/cli/v4/pkg/log"
"github.com/confluentinc/cli/v4/pkg/output"
"github.com/confluentinc/cli/v4/pkg/utils"
Expand Down Expand Up @@ -100,6 +101,7 @@ func (c *command) statementList(cmd *cobra.Command, _ []string) error {
ComputePool: statement.Spec.GetComputePoolId(),
Status: statement.Status.GetPhase(),
StatusDetail: statement.Status.GetDetail(),
Warnings: types.NewStatementWarnings(statement.Status.GetWarnings()),

@channingdong Channing Dong (channingdong) Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For the list command, can you try and see if the warnings can fit into the CLI console output? I am just a bit concerned that the warnings can sometimes be long, and we may just limit it to the describe command.

@raminqaf Ramin Gharib (raminqaf) Aug 6, 2026 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree with limiting the full block to describe. That is what the PR does today. list does not call printStatementWarnings, and human:"-" keeps the field out of the table.

One thing worth confirming: list -o json does carry warnings, I verified that the list endpoint populates status.warnings. So serialized consumers already get them.

If you want list users to see that warnings exist, a scalar column would fit the table where prose does not. Either a count or the highest severity, so you know which statement to go describe. Happy to add that if you think it earns the column width as a follow-up

LatestOffsets: statement.Status.GetLatestOffsets(),
LatestOffsetsTimestamp: flinkgatewayv1.PtrTime(statement.Status.GetLatestOffsetsTimestamp()),
})
Expand Down
14 changes: 14 additions & 0 deletions ...er/.snapshots/TestStatementControllerTestSuite-TestExecuteStatementWithStructuredWarnings
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
Statement successfully submitted.
Waiting for statement to be ready. Statement phase: PENDING.
Warnings:

CRITICAL [UPSERT_PRIMARY_KEY_MISMATCH] (Logged: 2026-07-30T09:15:00Z)
The primary key does not match the upsert key derived from the query.

MODERATE [MISSING_WINDOW_START_END] (Logged: 2026-07-30T08:00:00Z)
The GROUP BY clause contains only `window_start` with no corresponding `window_end`.

Statement phase is RUNNING.
Listening for execution errors. Press Enter to detach.
Finished statement execution. Statement phase: COMPLETED.

37 changes: 37 additions & 0 deletions pkg/flink/internal/controller/statement_controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,6 +333,43 @@ func (s *StatementControllerTestSuite) TestExecuteStatementWithWarning() {
cupaloy.SnapshotT(s.T(), stdout)
}

func (s *StatementControllerTestSuite) TestExecuteStatementWithStructuredWarnings() {
statementToExecute := "insert into users values ('test');"
legacyDetail := "[Warning] The primary key does not match the upsert key derived from the query."
windowWarningTime := time.Date(2026, 7, 30, 8, 0, 0, 0, time.UTC)
upsertWarningTime := time.Date(2026, 7, 30, 9, 15, 0, 0, time.UTC)
warnings := []types.StatementWarning{
{
Severity: "MODERATE",
Reason: "MISSING_WINDOW_START_END",
Message: "The GROUP BY clause contains only `window_start` with no corresponding `window_end`.",
CreatedAt: &windowWarningTime,
},
{
Severity: "CRITICAL",
Reason: "UPSERT_PRIMARY_KEY_MISMATCH",
Message: "The primary key does not match the upsert key derived from the query.",
CreatedAt: &upsertWarningTime,
},
}
processedStatement := types.ProcessedStatement{Status: types.PENDING, Principal: "sa-123", StatusDetail: legacyDetail, Warnings: warnings}
runningStatement := types.ProcessedStatement{Status: types.RUNNING, StatusDetail: legacyDetail, Warnings: warnings}
completedStatement := types.ProcessedStatement{Status: types.COMPLETED}
s.store.EXPECT().ProcessStatement(statementToExecute).Return(&processedStatement, nil)
s.consoleParser.EXPECT().Read().Return(nil, nil).AnyTimes()
s.store.EXPECT().WaitPendingStatement(gomock.Any(), processedStatement).Return(&runningStatement, nil)
s.store.EXPECT().FetchStatementResults(runningStatement).Return(&runningStatement, nil)
s.store.EXPECT().WaitForTerminalStatementState(gomock.Any(), runningStatement).Return(&completedStatement, nil)

stdout := testUtils.RunAndCaptureSTDOUT(s.T(), func() {
returnedStatement, err := s.statementController.ExecuteStatement(statementToExecute)
require.Nil(s.T(), err)
require.Equal(s.T(), &completedStatement, returnedStatement)
})

cupaloy.SnapshotT(s.T(), stdout)
}

func (s *StatementControllerTestSuite) TestRenderMsgAndStatusLocalStatements() {
tests := []struct {
name string
Expand Down
2 changes: 2 additions & 0 deletions pkg/flink/internal/store/store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -351,6 +351,8 @@ func (s *Store) WaitForTerminalStatementState(ctx context.Context, statement typ

statement.Status = types.PHASE(statementObj.Status.GetPhase())
statement.StatusDetail = statusDetail
// Warnings can be added after submission, so refresh them on every poll.
statement.Warnings = types.NewStatementWarnings(statementObj.Status.GetWarnings())
if statement.IsTerminalState() {
break
}
Expand Down
31 changes: 23 additions & 8 deletions pkg/flink/types/processed_statement.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,13 +23,14 @@ const (

// ProcessedStatement Custom Internal type that shall be used internally by the client
type ProcessedStatement struct {
Statement string `json:"statement"`
StatementName string `json:"statement_name"`
Kind string `json:"kind"`
ComputePool string `json:"compute_pool"`
Principal string `json:"principal"` // Cloud only
Status PHASE `json:"status"`
StatusDetail string `json:"status_detail,omitempty"` // Shown at the top before the table
Statement string `json:"statement"`
StatementName string `json:"statement_name"`
Kind string `json:"kind"`
ComputePool string `json:"compute_pool"`
Principal string `json:"principal"` // Cloud only
Status PHASE `json:"status"`
StatusDetail string `json:"status_detail,omitempty"` // Shown at the top before the table
Warnings []StatementWarning `json:"warnings,omitempty"`
IsLocalStatement bool
IsSensitiveStatement bool
PageToken string
Expand All @@ -46,6 +47,7 @@ func NewProcessedStatement(statementObj flinkgatewayv1.SqlV1Statement) *Processe
ComputePool: statementObj.Spec.GetComputePoolId(),
Principal: statementObj.Spec.GetPrincipal(),
StatusDetail: statementObj.Status.GetDetail(),
Warnings: NewStatementWarnings(statementObj.Status.GetWarnings()),
Status: PHASE(statementObj.Status.GetPhase()),
Properties: statementObj.Spec.GetProperties(),
Traits: StatementTraits{FlinkGatewayV1StatementTraits: &traits},
Expand Down Expand Up @@ -92,10 +94,21 @@ func (s ProcessedStatement) printStatusMessageOfNonLocalStatement() {
}
}

if s.StatusDetail != "" {
// The status detail can repeat the warnings, so only print it when there are none. A failed
// statement is the exception: its detail holds the failure reason.
if s.StatusDetail != "" && (s.Status == FAILED || len(s.Warnings) == 0) {
utils.OutputInfof("Details: ")
utils.OutputWarn(s.StatusDetail)
}

s.printWarnings()
}

func (s ProcessedStatement) printWarnings() {
if warnings := FormatStatementWarnings(s.Warnings); warnings != "" {
utils.OutputWarn(warnings)
utils.OutputInfo("")
}
}

func (s ProcessedStatement) PrintOutputDryRunStatement() {
Expand All @@ -110,6 +123,8 @@ func (s ProcessedStatement) PrintOutputDryRunStatement() {
utils.OutputInfof("Details: ")
utils.OutputErr(s.StatusDetail)
}

s.printWarnings()
}

func (s ProcessedStatement) GetPageSize() int {
Expand Down
69 changes: 69 additions & 0 deletions pkg/flink/types/processed_statement_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
package types

import (
"testing"

"github.com/stretchr/testify/require"

testUtils "github.com/confluentinc/cli/v4/pkg/flink/test"
)

func TestPrintStatusMessagePrintsStructuredWarningsInsteadOfStatusDetail(t *testing.T) {
statement := ProcessedStatement{
Status: RUNNING,
StatusDetail: "[Warning] legacy inlined warning",
Warnings: []StatementWarning{{Severity: "CRITICAL", Reason: "SOME_REASON", Message: "Fix the query."}},
}

stdout := testUtils.RunAndCaptureSTDOUT(t, statement.PrintStatusMessage)

require.Contains(t, stdout, "CRITICAL [SOME_REASON]")
require.Contains(t, stdout, "Fix the query.")
require.NotContains(t, stdout, "legacy inlined warning")
require.NotContains(t, stdout, "Details: ")
}

func TestPrintStatusMessagePrintsStatusDetailOfFailedStatementAlongsideWarnings(t *testing.T) {
statement := ProcessedStatement{
Status: FAILED,
StatusDetail: "the failure reason",
Warnings: []StatementWarning{{Severity: "LOW", Reason: "SOME_REASON", Message: "Fix the query."}},
}

stdout := testUtils.RunAndCaptureSTDOUT(t, statement.PrintStatusMessage)

require.Contains(t, stdout, "the failure reason")
require.Contains(t, stdout, "LOW [SOME_REASON]")
}

func TestPrintStatusMessagePrintsStatusDetailWhenThereAreNoWarnings(t *testing.T) {
statement := ProcessedStatement{Status: RUNNING, StatusDetail: "something worth knowing"}

stdout := testUtils.RunAndCaptureSTDOUT(t, statement.PrintStatusMessage)

require.Contains(t, stdout, "Details: ")
require.Contains(t, stdout, "something worth knowing")
require.NotContains(t, stdout, "Warnings:")
}

func TestPrintStatusMessagePrintsNoWarningsBlockWhenThereAreNoWarnings(t *testing.T) {
statement := ProcessedStatement{Status: RUNNING}

stdout := testUtils.RunAndCaptureSTDOUT(t, statement.PrintStatusMessage)

require.Contains(t, stdout, "Statement successfully submitted.")
require.NotContains(t, stdout, "Warnings:")
require.NotContains(t, stdout, "Details: ")
}

func TestPrintOutputDryRunStatementPrintsWarnings(t *testing.T) {
statement := ProcessedStatement{
Status: COMPLETED,
Warnings: []StatementWarning{{Severity: "MODERATE", Reason: "SOME_REASON", Message: "Fix the query."}},
}

stdout := testUtils.RunAndCaptureSTDOUT(t, statement.PrintOutputDryRunStatement)

require.Contains(t, stdout, "MODERATE [SOME_REASON]")
require.Contains(t, stdout, "Fix the query.")
}
Loading