AHeise commented on code in PR #29843:
URL: https://github.com/apache/flink/pull/29843#discussion_r4538941768
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java:
##########
@@ -591,18 +661,74 @@ private void executeContinuousRefreshJob(
final RefreshJobResult result =
jobSubmitter.submitRefreshJob(customConfig, insertStatement);
refreshJobController.register(result);
- ContinuousRefreshHandler continuousRefreshHandler =
- new ContinuousRefreshHandler(
- result.getTarget().getExecutionTarget(),
- result.getTarget().getClusterId(),
- result.getJobId());
+ return new ContinuousRefreshHandler(
+ result.getTarget().getExecutionTarget(),
+ result.getTarget().getClusterId(),
+ result.getJobId());
+ }
+ private void recordContinuousRefreshJob(
+ ObjectIdentifier materializedTableIdentifier,
+ CatalogMaterializedTable catalogMaterializedTable,
+ ContinuousRefreshHandler refreshHandler) {
updateRefreshHandler(
materializedTableIdentifier,
resolveCatalogMaterializedTable(catalogMaterializedTable),
RefreshStatus.ACTIVATED,
- continuousRefreshHandler.asSummaryString(),
- serializeContinuousHandler(continuousRefreshHandler));
+ refreshHandler.asSummaryString(),
+ serializeContinuousHandler(refreshHandler));
+ }
+
+ private TableException cancelUnrecordedRefreshJob(
+ ObjectIdentifier tableIdentifier,
+ ContinuousRefreshHandler refreshHandler,
+ Optional<String> restorePath,
+ Exception recordFailure) {
+ final String jobId = refreshHandler.getJobId();
+ LOG.error(
+ "Failed to record the continuous refresh job {} of
materialized table '{}' in the catalog, cancelling it.",
+ jobId,
+ tableIdentifier.asSummaryString(),
+ recordFailure);
+ try {
+ refreshJobController.cancel(refreshHandler);
+ } catch (Exception cancelFailure) {
+ LOG.error(
+ "Failed to cancel the continuous refresh job {} of
materialized table '{}' (execution.target={}, cluster id={}), which the catalog
does not record. It may still be running and must be cancelled manually.",
+ jobId,
+ tableIdentifier.asSummaryString(),
+ refreshHandler.getExecutionTarget(),
+ refreshHandler.getClusterId(),
+ cancelFailure);
+ final TableException failure =
+ new UncancelledRefreshJobException(
+ jobId,
+ String.format(
+ "Failed to record the continuous refresh
job %s of materialized table '%s' in the catalog, and cancelling it failed as
well. The job may still be running (execution.target=%s, cluster id=%s) and
must be cancelled manually.",
+ jobId,
+ tableIdentifier.asSummaryString(),
+ refreshHandler.getExecutionTarget(),
+ refreshHandler.getClusterId()),
+ cancelFailure);
+ failure.addSuppressed(recordFailure);
+ return failure;
+ }
+ final String cancelledMessage =
+ String.format(
+ "Failed to record the continuous refresh job %s of
materialized table '%s' in the catalog. The job was cancelled.",
+ jobId, tableIdentifier.asSummaryString());
+
+ // A catalog write can commit but a cancelled job can still be
reported as ACTIVATED, and
+ // RESUME would start without the savepoint this job restored from.
Review Comment:
Nit: hard to parse. Maybe "A catalog write can commit and still throw,
leaving the cancelled job recorded as ACTIVATED; RESUME would then start
without the savepoint this job restored from."
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java:
##########
@@ -846,40 +972,63 @@ 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) {
+ cleanUpAfterFailure(
+ () ->
+ generateRollbackAlterMaterializedTableOperation(
+ suspendMaterializedTable,
+
alterMaterializedTableChangeOperation)
+ .execute(operationCtx),
+ e,
+ withUncancelledRefreshJobHint(
+ restartFailureMessage
+ + " Rolling back to the previous query
failed as well, so the catalog holds the new query with the previous query's
savepoint, and RESUME would start the new query from that savepoint.",
+ e));
+ throw new TableException(
+ restartFailureMessage
+ + " The table was rolled back to its previous
query and left in SUSPENDED status, because the new refresh job could not be
cancelled. Cancel that job, then RESUME the table.",
+ e);
} catch (Exception e) {
final ContinuousRefreshHandler suspendedRefreshHandler =
deserializeContinuousHandler(
suspendMaterializedTable.getSerializedRefreshHandler());
LOG.warn(
- "Failed to start the continuous refresh job for
materialized table {} using new query {}, rollback to origin query {} with
savepoint {}.",
- tableIdentifier,
+ "Failed to start the continuous refresh job for
materialized table '{}' using new query {}, rollback to origin query {} with
savepoint {}.",
+ tableIdentifier.asSummaryString(),
op.getNewTable().getExpandedQuery(),
suspendMaterializedTable.getExpandedQuery(),
suspendedRefreshHandler.getRestorePath().orElse(null),
e);
// Roll back the changes to the materialized table and restore the
continuous
// refresh job
- generateRollbackAlterMaterializedTableOperation(
- suspendMaterializedTable,
alterMaterializedTableChangeOperation)
- .execute(operationCtx);
- executeContinuousRefreshJob(
- suspendMaterializedTable,
- tableIdentifier,
- 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);
+ cleanUpAfterFailure(
+ () -> {
+ generateRollbackAlterMaterializedTableOperation(
+ suspendMaterializedTable,
+ alterMaterializedTableChangeOperation)
+ .execute(operationCtx);
+ executeContinuousRefreshJob(
+ suspendMaterializedTable,
+ tableIdentifier,
+ Map.of(),
+ suspendedRefreshHandler.getRestorePath());
+ },
+ e,
+ restartFailureMessage
+ + " Rolling back to the previous query and
restarting its refresh job failed as well.");
Review Comment:
The fixed message hides which step failed:
- Rollback succeeded, restart failed: the table is SUSPENDED with the old
query and recoverable via RESUME, but the message doesn't say so.
- Restart threw `UncancelledRefreshJobException`: the "may still be running,
cancel it" hint only shows up deep in the cause chain.
Splitting the rollback and the restart into two `cleanUpAfterFailure` calls
with their own messages (and `withUncancelledRefreshJobHint` on the restart)
would cover both.
--
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]
