[FLINK-41390][table-planner] Cancel refresh jobs the catalog could not record

docs/content.zh/docs/sql/materialized-table/statements.md

@@ -81,6 +81,8 @@ AS <select_statement>
 }
 ```
 
+In continuous mode, if the refresh job starts but the catalog cannot record it, CREATE cancels the job and drops the table. If the job cannot be cancelled either, the error names it, so you can cancel it before you run CREATE again.
+
 ## PRIMARY KEY
 
 `PRIMARY KEY` 定义了一组可选的列，用于唯一标识表中的每一行。主键列必须非空。
@@ -673,6 +675,8 @@ ALTER MATERIALIZED TABLE my_materialized_table RESUME;
 ALTER MATERIALIZED TABLE my_materialized_table RESUME WITH ('sink.parallelism'='10');
 ```
 
+In continuous mode, if the refresh job starts but the catalog cannot record it, the job is cancelled. If the job cannot be cancelled either, the error names it, so you can cancel it before you retry.
+
 ## SET
 
 ```
@@ -781,6 +785,8 @@ ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name AS <select_statemen
    - 新的流式作业会从头开始，而不会从之前的流式作业状态恢复。
    - 数据源的起始位点会由到连接器的默认实现或查询中设置的 [dynamic hint]({{< ref "docs/sql/reference/queries/hints" >}}#dynamic-table-options) 决定。
 
+If the new refresh job starts but the catalog cannot record it, the job is cancelled and the old job restarts from its savepoint. If that cancel fails, the table is rolled back and left SUSPENDED instead of restarting the old job.
+
 **示例：**
 
 ```sql

docs/content/docs/sql/materialized-table/statements.md

@@ -81,6 +81,8 @@ AS <select_statement>
 }
 ```
 
+In continuous mode, if the refresh job starts but the catalog cannot record it, CREATE cancels the job and drops the table. If the job cannot be cancelled either, the error names it, so you can cancel it before you run CREATE again.
+
 ## PRIMARY KEY
 
 `PRIMARY KEY` defines an optional list of columns that uniquely identifies each row within the table. The column as the primary key must be non-null.
@@ -673,6 +675,8 @@ ALTER MATERIALIZED TABLE my_materialized_table RESUME;
 ALTER MATERIALIZED TABLE my_materialized_table RESUME WITH ('sink.parallelism'='10');
 ```
 
+In continuous mode, if the refresh job starts but the catalog cannot record it, the job is cancelled. If the job cannot be cancelled either, the error names it, so you can cancel it before you retry.
+
 ## SET
 
 ```
@@ -782,6 +786,8 @@ The modification process depends on the refresh mode of the materialized table:
    - The new refresh job starts from the beginning and does not restore from the previous state.
    - The starting offset of the data source is determined by the connector’s default implementation or the [dynamic hint]({{< ref "docs/sql/reference/queries/hints" >}}#dynamic-table-options) specified in the query.
 
+If the new refresh job starts but the catalog cannot record it, the job is cancelled and the old job restarts from its savepoint. If that cancel fails, the table is rolled back and left SUSPENDED instead of restarting the old job.
+
 **Example:**
 
 ```sql

flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java

@@ -494,8 +494,8 @@ void testCreateMaterializedTableFailedInInContinuousMode() {
                 .cause()
                 .hasMessageContaining(
                         String.format(
-                                "Failed to submit continuous refresh job for materialized table %s.",
-                                userShopsIdentifier.asSerializableString()));
+                                "Failed to submit continuous refresh job for materialized table '%s'.",
+                                userShopsIdentifier.asSummaryString()));
 
         // verify the materialized table is not created
         assertThatThrownBy(() -> getTable(userShopsIdentifier))
@@ -848,8 +848,8 @@ void testAlterMaterializedTableWithRepeatedSuspendAndResumeInContinuousMode(
                 .isInstanceOf(TableException.class)
                 .hasMessageContaining(
                         String.format(
-                                "Materialized table %s continuous refresh job has been suspended",
-                                userShopsIdentifier));
+                                "Materialized table '%s' continuous refresh job has been suspended",
+                                userShopsIdentifier.asSummaryString()));
 
         // resume materialized table
         String alterMaterializedTableResumeDDL = "ALTER MATERIALIZED TABLE users_shops RESUME";
@@ -870,8 +870,8 @@ void testAlterMaterializedTableWithRepeatedSuspendAndResumeInContinuousMode(
                 .isInstanceOf(TableException.class)
                 .hasMessageContaining(
                         String.format(
-                                "Materialized table %s continuous refresh job has been resumed",
-                                userShopsIdentifier));
+                                "Materialized table '%s' continuous refresh job has been resumed",
+                                userShopsIdentifier.asSummaryString()));
     }
 
     @Test
@@ -979,8 +979,8 @@ void testAlterMaterializedTableWithRepeatedSuspendAndResumeInFullMode() throws E
                 .isInstanceOf(TableException.class)
                 .hasMessageContaining(
                         String.format(
-                                "Materialized table %s refresh workflow has been suspended.",
-                                userShopsIdentifier));
+                                "Materialized table '%s' refresh workflow has been suspended.",
+                                userShopsIdentifier.asSummaryString()));
 
         // resume materialized table
         String alterMaterializedTableResumeDDL =
@@ -1002,8 +1002,8 @@ void testAlterMaterializedTableWithRepeatedSuspendAndResumeInFullMode() throws E
                 .isInstanceOf(TableException.class)
                 .hasMessageContaining(
                         String.format(
-                                "Materialized table %s refresh workflow has been resumed.",
-                                userShopsIdentifier));
+                                "Materialized table '%s' refresh workflow has been resumed.",
+                                userShopsIdentifier.asSummaryString()));
     }
 
     @Test
@@ -1577,8 +1577,8 @@ void testDropMaterializedTableInContinuousMode() throws Exception {
                 .isInstanceOf(ValidationException.class)
                 .hasMessage(
                         String.format(
-                                "Materialized table with identifier %s does not exist.",
-                                userShopsIdentifier.asSerializableString()));
+                                "Materialized table with identifier '%s' does not exist.",
+                                userShopsIdentifier.asSummaryString()));
 
         String dropNonExistMaterializedTableDDL2 = "DROP MATERIALIZED TABLE IF EXISTS users_shops";
         OperationHandle dropNonExistMaterializedTableHandle2 =
@@ -1593,8 +1593,8 @@ void testDropMaterializedTableInContinuousMode() throws Exception {
                 .isInstanceOf(ValidationException.class)
                 .hasMessage(
                         String.format(
-                                "Table %s is not a materialized table, does not support materialized table related operation.",
-                                getObjectIdentifier("datagenSource").asSerializableString()));
+                                "Table '%s' is not a materialized table, does not support materialized table related operation.",
+                                getObjectIdentifier("datagenSource").asSummaryString()));
     }
 
     @Test
@@ -1876,8 +1876,8 @@ void testRefreshMaterializedTableWithInvalidParameterInContinuousMode() throws E
                 .isInstanceOf(ValidationException.class)
                 .hasMessage(
                         String.format(
-                                "The scheduler time must not be null during the periodic refresh of the materialized table %s.",
-                                objectIdentifier.asSerializableString()));
+                                "The scheduler time must not be null during the periodic refresh of the materialized table '%s'.",
+                                objectIdentifier.asSummaryString()));
 
         // refresh the materialized table with invalid schedule time
         String invalidTime = "20240103 00:00:00.000";
@@ -1899,8 +1899,8 @@ void testRefreshMaterializedTableWithInvalidParameterInContinuousMode() throws E
                 .isInstanceOf(TableException.class)
                 .hasMessage(
                         String.format(
-                                "Failed to parse a valid partition value for the field 'ds' in materialized table %s using the scheduler time '20240103 00:00:00.000' based on the date format 'yyyy-MM-dd HH:mm:ss'.",
-                                objectIdentifier.asSerializableString()));
+                                "Failed to parse a valid partition value for the field 'ds' in materialized table '%s' using the scheduler time '20240103 00:00:00.000' based on the date format 'yyyy-MM-dd HH:mm:ss'.",
+                                objectIdentifier.asSummaryString()));
     }
 
     @Test