AHeise commented on code in PR #29843:
URL: https://github.com/apache/flink/pull/29843#discussion_r4536959165
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java:
##########
@@ -262,12 +280,17 @@ private void
convertTableToMaterializedTableInContinuousMode(
Map.of(),
Optional.empty());
} catch (Exception e) {
- suspendMaterializedTable(materializedTableIdentifier,
catalogMaterializedTable);
+ try {
+ suspendMaterializedTable(materializedTableIdentifier,
catalogMaterializedTable);
+ } catch (RuntimeException suspendFailure) {
+ suspendFailure.addSuppressed(e);
+ throw suspendFailure;
+ }
throw new TableException(
String.format(
- "Failed to start the continuous refresh job when
converting table %s to a materialized table. "
+ "Failed to start the continuous refresh job when
converting table '%s' to a materialized table. "
+ "The table was converted and left in
SUSPENDED status; resume it once the issue is resolved.",
- materializedTableIdentifier),
+ materializedTableIdentifier.asSummaryString()),
e);
Review Comment:
Pre-existing, but now reachable through the new compensation: after a failed
continuous conversion the table is SUSPENDED without a serialized refresh
handler, as far as I can tell. Does RESUME work from that state?
`resumeContinuousRefreshJob` deserializes the handler unconditionally
(:519-521), so I'd expect it to fail. If so, "resume it once the issue is
resolved" here and in `statements.md` doesn't hold. Extending
`convertCancelsARefreshJobItCouldNotRecord` with a RESUME would answer it.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java:
##########
@@ -191,12 +195,19 @@ private void createMaterializedTableInContinuousMode(
} catch (Exception e) {
// drop materialized table if submitting the Flink streaming job
encounters an
// exception. Thus, weak atomicity is guaranteed
- new DropMaterializedTableOperation(materializedTableIdentifier,
true)
- .execute(operationCtx);
+ cleanUpAfterFailure(
+ () ->
+ new
DropMaterializedTableOperation(materializedTableIdentifier, true)
+ .execute(operationCtx),
+ e,
+ String.format(
+ "Failed to submit continuous refresh job for
materialized table '%s', and dropping the table failed as well."
+ + LEFT_IN_INITIALIZING_STATUS,
+ materializedTableIdentifier.asSummaryString()));
throw new TableException(
String.format(
- "Failed to submit continuous refresh job for
materialized table %s.",
- materializedTableIdentifier),
+ "Failed to submit continuous refresh job for
materialized table '%s'.",
+ materializedTableIdentifier.asSummaryString()),
e);
Review Comment:
If recording fails and cancel fails as well, we still drop the table and
report "Failed to submit continuous refresh job", although the job was
submitted and may still be running. Only the cause says to cancel it manually.
A user re-running CREATE then gets a second writer. Can the top-level message
reflect the uncancelled case, like the ALTER branch does?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java:
##########
@@ -560,6 +601,25 @@ private void executeContinuousRefreshJob(
ObjectIdentifier materializedTableIdentifier,
Map<String, String> dynamicOptions,
Optional<String> restorePath) {
+ final ContinuousRefreshHandler refreshHandler =
+ submitContinuousRefreshJob(
+ catalogMaterializedTable,
+ materializedTableIdentifier,
+ dynamicOptions,
+ restorePath);
+ try {
+ recordContinuousRefreshJob(
+ materializedTableIdentifier, catalogMaterializedTable,
refreshHandler);
+ } catch (Exception e) {
+ throw cancelUnrecordedRefreshJob(materializedTableIdentifier,
refreshHandler, e);
+ }
Review Comment:
If `alterTable` committed but still threw (e.g. a timeout after commit), we
cancel a job the catalog now records as ACTIVATED without a savepoint, and the
next RESUME starts without state. Rare, but cheap to mitigate for RESUME: put
the restore path into the "The job was cancelled." message so the user can set
`execution.state-recovery.path`.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java:
##########
@@ -507,8 +548,8 @@ private void resumeContinuousRefreshJob(
} catch (Exception e) {
throw new TableException(
String.format(
- "Failed to resume the continuous refresh job for
materialized table %s.",
- tableIdentifier),
+ "Failed to resume the continuous refresh job for
materialized table '%s'.",
+ tableIdentifier.asSummaryString()),
e);
Review Comment:
nit: For `UncancelledRefreshJobException`, the top-level "Failed to resume…"
hides that a job may still be running. Retrying RESUME on SUSPENDED doesn't
check for it, so the user ends up with two writers. Same special-casing as in
ALTER?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java:
##########
@@ -846,19 +949,36 @@ private void alterActivatedContinuousMaterializedTable(
alterMaterializedTableChangeOperation.execute(operationCtx);
// 3. resume the materialized table
+ final String restartFailureMessage =
+ String.format(
+ "Failed to start the continuous refresh job using new
query %s when altering materialized table '%s' select query.",
+ op.getNewTable().getExpandedQuery(),
tableIdentifier.asSummaryString());
try {
executeContinuousRefreshJob(
alterMaterializedTableChangeOperation.getNewTable(),
tableIdentifier,
Map.of(),
Optional.empty());
+ } catch (UncancelledRefreshJobException e) {
+ try {
+ generateRollbackAlterMaterializedTableOperation(
+ suspendMaterializedTable,
alterMaterializedTableChangeOperation)
+ .execute(operationCtx);
+ } catch (RuntimeException rollbackFailure) {
+ rollbackFailure.addSuppressed(e);
+ throw rollbackFailure;
+ }
Review Comment:
If the rollback fails here, we rethrow the raw rollback exception. The
catalog then holds the new query, SUSPENDED status, and the old handler with
the old savepoint, while the new job keeps running. A RESUME would restore the
new query from the old query's savepoint. Wrap it with a message that names
both facts?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java:
##########
@@ -108,6 +108,10 @@ public class DefaultMaterializedTableExecutor implements
MaterializedTableExecut
private static final Logger LOG =
LoggerFactory.getLogger(DefaultMaterializedTableExecutor.class);
+ private static final String LEFT_IN_INITIALIZING_STATUS =
+ " The table remains in the catalog with its creation unfinished,
and no materialized"
+ + " table statement can drop or recover it. Remove it from
the catalog directly.";
Review Comment:
Dropping the status name makes this harder to act on: `DESCRIBE` still shows
the refresh status as `INITIALIZING`, and a later `DROP MATERIALIZED TABLE`
fails with "Current refresh status of materialized table '%s' is initializing"
(:1096). The user can no longer connect the two. How about keeping both, e.g.
"The table remains in the catalog with refresh status INITIALIZING (creation
unfinished), and no materialized table statement can drop or recover it."
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java:
##########
@@ -875,11 +995,7 @@ private void alterActivatedContinuousMaterializedTable(
Map.of(),
suspendedRefreshHandler.getRestorePath());
- throw new TableException(
- String.format(
- "Failed to start the continuous refresh job using
new query %s when altering materialized table %s select query.",
- op.getNewTable().getExpandedQuery(),
tableIdentifier),
- e);
+ throw new TableException(restartFailureMessage, e);
Review Comment:
Pre-existing: if the rollback or the restart of the old job above fails, `e`
(why the new query failed) is lost, because the exception escapes the catch
block. That now includes the new "job was cancelled" and uncancelled exceptions
from `executeContinuousRefreshJob`. Worth wrapping, but fine as a follow-up.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java:
##########
@@ -875,11 +995,7 @@ private void alterActivatedContinuousMaterializedTable(
Map.of(),
suspendedRefreshHandler.getRestorePath());
Review Comment:
`RefreshJobController#cancel` completes when the cancel request is
acknowledged, not when the job reaches CANCELED (`JobMaster#cancel` returns the
ack immediately). So the old job restarts from its savepoint while the new one
may still be cancelling. Much better than the permanent second writer before
this PR, but waiting for a globally terminal state (bounded by the client
timeout) would close the gap. Fine as a follow-up.
##########
docs/content/docs/sql/materialized-table/statements.md:
##########
@@ -81,6 +81,8 @@ AS <select_statement>
}
```
+In continuous mode, if a statement starts a refresh job but the catalog cannot
record it, the job is cancelled, and CREATE also drops the table it created.
Review Comment:
This applies to every statement but sits in the CREATE section, and it says
"the job is cancelled" without the "if that cancel fails…" caveat the ALTER
paragraph has.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
