[APIE-720,APIE-724] Flink Shell: don't fetch statement results if the statement phase is failed (#3229)

Author: sgagniere

pkg/cmd/authenticated_cli_command.go

@@ -120,7 +120,7 @@ func (c *AuthenticatedCLICommand) getGatewayUrlForComputePool(access, id string)
 		return privateURL, nil
 	}
 	if access == "" {
-		output.Printf(c.Config.EnableColor, "No Flink endpoint is specified, defaulting to public endpoint: `%s`\n", publicURL)
+		output.ErrPrintf(c.Config.EnableColor, "No Flink endpoint is specified, defaulting to public endpoint: `%s`\n", publicURL)
 	}
 	return publicURL, nil
 }
@@ -146,7 +146,7 @@ func (c *AuthenticatedCLICommand) getGatewayUrlForRegion(accessType, provider, r
 		return "", errors.NewErrorWithSuggestions("invalid region", "Please select a valid region - use `confluent flink region list` to see available regions")
 	}
 	if accessType == "" {
-		output.Printf(c.Config.EnableColor, "No Flink endpoint is specified, defaulting to public endpoint: `%s`\n", hostUrl)
+		output.ErrPrintf(c.Config.EnableColor, "No Flink endpoint is specified, defaulting to public endpoint: `%s`\n", hostUrl)
 	}
 
 	u, err := purl.Parse(hostUrl)

pkg/flink/internal/results/result_fetcher.go

@@ -120,7 +120,7 @@ func (t *ResultFetcher) updateState(newResults *types.ProcessedStatement, err *t
 		return
 	}
 
-	if err != nil {
+	if err != nil || (newResults != nil && newResults.Status == types.FAILED) {
 		t.refreshState.setState(types.Failed)
 		return
 	}

pkg/flink/internal/store/store.go

@@ -146,6 +146,12 @@ func (s *Store) FetchStatementResults(statement types.ProcessedStatement) (*type
 		return &statement, nil
 	}
 
+	statementObj, err := s.authenticatedGatewayClient().GetStatement(s.appOptions.GetEnvironmentId(), statement.StatementName, s.appOptions.GetOrganizationId())
+	if err != nil {
+		return nil, &types.StatementError{Message: err.Error()}
+	}
+	statement.Status = types.PHASE(statementObj.Status.GetPhase())
+
 	// Process remote statements that are now running or completed
 	statementResultObj, err := s.authenticatedGatewayClient().GetStatementResults(s.appOptions.GetEnvironmentId(), statement.StatementName, s.appOptions.GetOrganizationId(), statement.PageToken)
 	if err != nil {

pkg/flink/internal/store/store_test.go

@@ -1268,6 +1268,14 @@ func (s *StoreTestSuite) TestFetchResultsNoRetryWithCompletedStatement() {
 		userProperties := NewUserProperties(&appOptions)
 		store := NewStore(client, mockAppController.ExitApplication, userProperties, &appOptions, tokenRefreshFunc)
 
+		statementObj := flinkgatewayv1.SqlV1Statement{
+			Name: flinkgatewayv1.PtrString(testStatementName),
+			Status: &flinkgatewayv1.SqlV1StatementStatus{
+				Phase: "COMPLETED",
+			},
+		}
+		client.EXPECT().GetStatement("envId", statement.StatementName, "orgId").Return(statementObj, nil)
+
 		statementResultObj := flinkgatewayv1.SqlV1StatementResult{
 			Metadata: flinkgatewayv1.ResultListMeta{},
 			Results:  &flinkgatewayv1.SqlV1StatementResultResults{},
@@ -1319,6 +1327,14 @@ func (s *StoreTestSuite) TestFetchResultsWithRunningStatement() {
 		userProperties := NewUserProperties(&appOptions)
 		store := NewStore(client, mockAppController.ExitApplication, userProperties, &appOptions, tokenRefreshFunc)
 
+		statementObj := flinkgatewayv1.SqlV1Statement{
+			Name: flinkgatewayv1.PtrString(testStatementName),
+			Status: &flinkgatewayv1.SqlV1StatementStatus{
+				Phase: "RUNNING",
+			},
+		}
+		client.EXPECT().GetStatement("envId", statement.StatementName, "orgId").Return(statementObj, nil)
+
 		statementResultObj := flinkgatewayv1.SqlV1StatementResult{
 			Metadata: flinkgatewayv1.ResultListMeta{},
 			Results:  &flinkgatewayv1.SqlV1StatementResultResults{},
@@ -1374,6 +1390,14 @@ func (s *StoreTestSuite) TestFetchResultsNoRetryWhenPageTokenExists() {
 		userProperties := NewUserProperties(&appOptions)
 		store := NewStore(client, mockAppController.ExitApplication, userProperties, &appOptions, tokenRefreshFunc)
 
+		statementObj := flinkgatewayv1.SqlV1Statement{
+			Name: flinkgatewayv1.PtrString(testStatementName),
+			Status: &flinkgatewayv1.SqlV1StatementStatus{
+				Phase: "RUNNING",
+			},
+		}
+		client.EXPECT().GetStatement("envId", statement.StatementName, "orgId").Return(statementObj, nil)
+
 		nextPage := "https://devel.cpdev.cloud/some/results?page_token=eyJWZX"
 		statementResultObj := flinkgatewayv1.SqlV1StatementResult{
 			Metadata: flinkgatewayv1.ResultListMeta{Next: &nextPage},
@@ -1441,6 +1465,14 @@ func (s *StoreTestSuite) TestFetchResultsNoRetryWhenResultsExist() {
 		userProperties := NewUserProperties(&appOptions)
 		store := NewStore(client, mockAppController.ExitApplication, userProperties, &appOptions, tokenRefreshFunc)
 
+		statementObj := flinkgatewayv1.SqlV1Statement{
+			Name: flinkgatewayv1.PtrString(testStatementName),
+			Status: &flinkgatewayv1.SqlV1StatementStatus{
+				Phase: "RUNNING",
+			},
+		}
+		client.EXPECT().GetStatement("envId", statement.StatementName, "orgId").Return(statementObj, nil)
+
 		statementResultObj := flinkgatewayv1.SqlV1StatementResult{
 			Metadata: flinkgatewayv1.ResultListMeta{},
 			Results:  &flinkgatewayv1.SqlV1StatementResultResults{Data: &[]any{map[string]any{"op": 0}}},