lsyldliu commented on code in PR #24888:
URL: https://github.com/apache/flink/pull/24888#discussion_r1628807845
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java:
##########
@@ -780,37 +752,18 @@ private ResultFetcher callDropMaterializedTableOperation(
CatalogMaterializedTable materializedTable =
getCatalogMaterializedTable(operationExecutor,
tableIdentifier);
- if (CatalogMaterializedTable.RefreshStatus.ACTIVATED
- == materializedTable.getRefreshStatus()) {
- ContinuousRefreshHandler refreshHandler =
-
deserializeContinuousHandler(materializedTable.getSerializedRefreshHandler());
- // get job running status
- JobStatus jobStatus = getJobStatus(operationExecutor, handle,
refreshHandler);
- if (!jobStatus.isTerminalState()) {
- try {
- cancelJob(operationExecutor, handle,
refreshHandler.getJobId());
- } catch (Exception e) {
- jobStatus = getJobStatus(operationExecutor, handle,
refreshHandler);
- if (!jobStatus.isTerminalState()) {
- throw new SqlExecutionException(
- String.format(
- "Failed to drop the materialized table
%s because the continuous refresh job %s could not be canceled."
- + " The current status of the
continuous refresh job is %s.",
- tableIdentifier,
refreshHandler.getJobId(), jobStatus),
- e);
- } else {
- LOG.warn(
- "An exception occurred while canceling the
continuous refresh job {} for materialized table {},"
- + " but since the job is in a terminal
state, skip the cancel operation.",
- refreshHandler.getJobId(),
- tableIdentifier);
- }
+ if (CatalogMaterializedTable.RefreshStatus.ACTIVATED ==
materializedTable.getRefreshStatus()
Review Comment:
We should minimize the nesting depth of `if else` in code snippet, so can
optimize the code:
```
CatalogMaterializedTable.RefreshMode refreshMode =
materializedTable.getRefreshMode();
CatalogMaterializedTable.RefreshStatus refreshStatus =
materializedTable.getRefreshStatus();
if (CatalogMaterializedTable.RefreshStatus.ACTIVATED == refreshStatus
|| CatalogMaterializedTable.RefreshStatus.SUSPENDED ==
refreshStatus) {
if (CatalogMaterializedTable.RefreshMode.FULL == refreshMode) {
deleteRefreshWorkflow(tableIdentifier, materializedTable);
} else if (CatalogMaterializedTable.RefreshMode.CONTINUOUS ==
refreshMode
&& CatalogMaterializedTable.RefreshStatus.ACTIVATED ==
refreshStatus) {
cancelContinuousRefreshJob(
operationExecutor, handle, tableIdentifier,
materializedTable);
}
} else if (CatalogMaterializedTable.RefreshStatus.INITIALIZING ==
refreshStatus) {
throw new ValidationException(
String.format(
"Current refresh status of materialized table %s
is initializing, skip the drop operation.",
tableIdentifier.asSerializableString()));
}
```
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java:
##########
@@ -268,13 +268,10 @@ void testCreateMaterializedTableInFullMode() throws
Exception {
.containsKey("table.catalog-store.file.path")
.doesNotContainKey(WORKFLOW_SCHEDULER_TYPE.key())
.doesNotContainKey(RESOURCES_DOWNLOAD_DIR.key());
-
- // delete the workflow
- embeddedWorkflowScheduler.deleteScheduleWorkflow(jobKey.getName(),
jobKey.getGroup());
}
@Test
- void testCreateMaterializedTableFailed() throws Exception {
+ void testCreateMaterializedTableFailedInInContinuousMode() throws
Exception {
Review Comment:
`throws Exception` is not needed in the method signature.
--
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]
