From 076edfd22ebf69740d3b30b073182c905a7ea94a Mon Sep 17 00:00:00 2001 From: Natea Eshetu Beshada Date: Wed, 8 Jul 2026 15:56:35 -0700 Subject: [PATCH] [FRT-1629] Dedupe retryable statement errors and always surface a failure reason --- pkg/flink/internal/store/store.go | 35 ++++++++++- pkg/flink/internal/store/store_onprem.go | 7 ++- pkg/flink/internal/store/store_test.go | 74 +++++++++++++++++++++++- 3 files changed, 110 insertions(+), 6 deletions(-) diff --git a/pkg/flink/internal/store/store.go b/pkg/flink/internal/store/store.go index 302414dfe1..79a5f476f7 100644 --- a/pkg/flink/internal/store/store.go +++ b/pkg/flink/internal/store/store.go @@ -128,9 +128,14 @@ func (s *Store) WaitPendingStatement(ctx context.Context, statement types.Proces // Check for failed or cancelled statements statementStatus = updatedStatement.Status if statementStatus != types.COMPLETED && statementStatus != types.RUNNING { + failureMessage := updatedStatement.StatusDetail + if failureMessage == "" { + failureMessage = fmt.Sprintf("the server did not report a failure reason; run `confluent flink statement describe %s` for the full status", statement.StatementName) + } return nil, &types.StatementError{ Message: fmt.Sprintf("can't fetch results. Statement phase is: %s", statementStatus), - FailureMessage: updatedStatement.StatusDetail, + FailureMessage: failureMessage, + Suggestion: "This statement reached a terminal state without producing results. Check the statement definition and run `confluent flink statement describe` for the full status.", StatusCode: types.StatusCode(err), } } @@ -249,7 +254,7 @@ func (s *Store) waitForPendingStatement(ctx context.Context, statementName strin return nil, &types.StatementError{ Message: fmt.Sprintf("the server can't process this statement right now, exiting after %d retries", len(capturedErrors)), - FailureMessage: fmt.Sprintf("captured retryable errors: %s", strings.Join(capturedErrors, "; ")), + FailureMessage: formatCapturedErrors(capturedErrors), } } @@ -277,7 +282,7 @@ func (s *Store) waitForPendingStatement(ctx context.Context, statementName strin var errorsMsg string if len(capturedErrors) > 0 { - errorsMsg = fmt.Sprintf("captured retryable errors: %s", strings.Join(capturedErrors, "; ")) + errorsMsg = formatCapturedErrors(capturedErrors) } return nil, &types.StatementError{ @@ -287,6 +292,30 @@ func (s *Store) waitForPendingStatement(ctx context.Context, statementName strin } } +// formatCapturedErrors collapses repeated retryable status details into a single, +// de-duplicated, insertion-ordered summary so an identical error is not printed once +// per retry. +func formatCapturedErrors(capturedErrors []string) string { + counts := make(map[string]int, len(capturedErrors)) + order := make([]string, 0, len(capturedErrors)) + for _, e := range capturedErrors { + if _, seen := counts[e]; !seen { + order = append(order, e) + } + counts[e]++ + } + + parts := make([]string, 0, len(order)) + for _, e := range order { + if counts[e] > 1 { + parts = append(parts, fmt.Sprintf("%s (repeated %d times)", e, counts[e])) + } else { + parts = append(parts, e) + } + } + return fmt.Sprintf("captured retryable errors: %s", strings.Join(parts, "; ")) +} + func (s *Store) getStatusDetail(statementObj flinkgatewayv1.SqlV1Statement) string { status := statementObj.GetStatus() if status.GetDetail() != "" { diff --git a/pkg/flink/internal/store/store_onprem.go b/pkg/flink/internal/store/store_onprem.go index f7c8185cda..d41dc2349f 100644 --- a/pkg/flink/internal/store/store_onprem.go +++ b/pkg/flink/internal/store/store_onprem.go @@ -117,9 +117,14 @@ func (s *StoreOnPrem) WaitPendingStatement(ctx context.Context, statement types. // Check for failed or cancelled statements statementStatus = updatedStatement.Status if statementStatus != types.COMPLETED && statementStatus != types.RUNNING { + failureMessage := updatedStatement.StatusDetail + if failureMessage == "" { + failureMessage = fmt.Sprintf("the server did not report a failure reason; run `confluent flink statement describe %s` for the full status", statement.StatementName) + } return nil, &types.StatementError{ Message: fmt.Sprintf("can't fetch results. Statement phase is: %s", statementStatus), - FailureMessage: updatedStatement.StatusDetail, + FailureMessage: failureMessage, + Suggestion: "This statement reached a terminal state without producing results. Check the statement definition and run `confluent flink statement describe` for the full status.", StatusCode: types.StatusCode(err), } } diff --git a/pkg/flink/internal/store/store_test.go b/pkg/flink/internal/store/store_test.go index a6d612f19f..9d059db1af 100644 --- a/pkg/flink/internal/store/store_test.go +++ b/pkg/flink/internal/store/store_test.go @@ -5,7 +5,6 @@ import ( "fmt" "net/http" "reflect" - "strings" "testing" "time" @@ -244,7 +243,7 @@ func TestWaitForPendingHitsErrorRetryLimit(t *testing.T) { } expectedError := &types.StatementError{ Message: "the server can't process this statement right now, exiting after 6 retries", - FailureMessage: fmt.Sprintf("captured retryable errors: %s", strings.Repeat(testStatusDetailMessage+"; ", 5)+testStatusDetailMessage), + FailureMessage: fmt.Sprintf("captured retryable errors: %s (repeated 6 times)", testStatusDetailMessage), } client.EXPECT().GetStatement("envId", testStatementName, "orgId").Return(statementObj, nil).AnyTimes() processedStatement, err := s.waitForPendingStatement(context.Background(), testStatementName, timeout) @@ -2077,6 +2076,7 @@ func (s *StoreTestSuite) TestWaitPendingStatementFailsOnNonCompletedOrRunningSta expectedError := &types.StatementError{ Message: "can't fetch results. Statement phase is: FAILED", FailureMessage: testStatusDetailMessage, + Suggestion: "This statement reached a terminal state without producing results. Check the statement definition and run `confluent flink statement describe` for the full status.", } { // Cloud store @@ -2144,6 +2144,7 @@ func (s *StoreTestSuite) TestWaitPendingStatementFetchesExceptionOnFailedStateme expectedError := &types.StatementError{ Message: "can't fetch results. Statement phase is: FAILED", FailureMessage: exception1, + Suggestion: "This statement reached a terminal state without producing results. Check the statement definition and run `confluent flink statement describe` for the full status.", } { // Cloud store @@ -2220,6 +2221,75 @@ func (s *StoreTestSuite) TestWaitPendingStatementFetchesExceptionOnFailedStateme } } +func (s *StoreTestSuite) TestWaitPendingStatementSetsPlaceholderWhenFailedWithNoReason() { + expectedError := &types.StatementError{ + Message: "can't fetch results. Statement phase is: FAILED", + FailureMessage: fmt.Sprintf("the server did not report a failure reason; run `confluent flink statement describe %s` for the full status", testStatementName), + Suggestion: "This statement reached a terminal state without producing results. Check the statement definition and run `confluent flink statement describe` for the full status.", + } + + { // Cloud store + client := mock.NewMockGatewayClientInterface(gomock.NewController(s.T())) + store := Store{ + Properties: NewUserPropertiesWithDefaults(map[string]string{"TestProp": "TestVal"}, map[string]string{}), + client: client, + appOptions: &types.ApplicationOptions{ + OrganizationId: "orgId", + EnvironmentId: "envId", + }, + tokenRefreshFunc: tokenRefreshFunc, + } + + statementObj := flinkgatewayv1.SqlV1Statement{ + Name: flinkgatewayv1.PtrString(testStatementName), + Status: &flinkgatewayv1.SqlV1StatementStatus{ + Phase: "FAILED", + }, + } + + client.EXPECT().GetStatement("envId", testStatementName, "orgId").Return(statementObj, nil) + client.EXPECT().GetExceptions("envId", testStatementName, "orgId").Return([]flinkgatewayv1.SqlV1StatementException{}, nil) + + processedStatement, err := store.WaitPendingStatement(context.Background(), types.ProcessedStatement{ + StatementName: testStatementName, + Status: types.PENDING, + }) + require.Nil(s.T(), processedStatement) + require.Equal(s.T(), expectedError, err) + } + { // On-prem store + client := mock.NewMockCmfClientInterface(gomock.NewController(s.T())) + store := StoreOnPrem{ + Properties: NewUserPropertiesWithDefaults(map[string]string{"TestProp": "TestVal"}, map[string]string{}), + client: client, + appOptions: &types.ApplicationOptions{ + EnvironmentId: "envId", + }, + tokenRefreshFunc: tokenRefreshFunc, + } + + statementObj := cmfsdk.Statement{ + Metadata: cmfsdk.StatementMetadata{ + Name: testStatementName, + }, + Status: &cmfsdk.StatementStatus{ + Phase: "FAILED", + }, + } + + client.EXPECT().CmfApiContext().Return(context.Background()).Times(2) + client.EXPECT().GetStatement(context.Background(), "envId", testStatementName).Return(statementObj, nil) + client.EXPECT().ListStatementExceptions(context.Background(), "envId", testStatementName).Return(cmfsdk.StatementExceptionList{}, nil) + + processedStatement, err := store.WaitPendingStatement(context.Background(), types.ProcessedStatement{ + StatementName: testStatementName, + Status: types.PENDING, + }) + require.Nil(s.T(), processedStatement) + require.Equal(s.T(), expectedError, err) + } +} + func (s *StoreTestSuite) TestGetStatusDetail() { exception1 := "Exception 1" exception2 := "Exception 2"