[
https://issues.apache.org/jira/browse/FLINK-41350?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ramin Gharib updated FLINK-41350:
---------------------------------
Priority: Minor (was: Critical)
> A materialized table left SUSPENDED by a failed table conversion cannot be
> resumed
> ----------------------------------------------------------------------------------
>
> Key: FLINK-41350
> URL: https://issues.apache.org/jira/browse/FLINK-41350
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Gateway
> Affects Versions: 2.4.0
> Reporter: Ramin Gharib
> Priority: Minor
> Fix For: 2.4.0
>
>
> When the refresh job of a converted table fails to start, the SQL Gateway
> keeps the materialized table and sets it to {{SUSPENDED}}, as documented. No
> job ever ran, so the table has no refresh handler. Every statement that reads
> the handler then fails with a {{NullPointerException}}: {{RESUME}}, {{ALTER
> MATERIALIZED TABLE ... SET}} and {{CREATE OR ALTER MATERIALIZED TABLE}}. Only
> {{DROP MATERIALIZED TABLE}} works, and it deletes the data.
> The docs say to fix the cause and {{RESUME}}. Suspend-on-failure was chosen
> over a rollback to keep the data, but here the only way out deletes it.
> h3. Repro
> Use a 2.4 distribution with
> {{table.materialized-table.conversion-from-table.enabled: true}} in
> {{config.yaml}}, through {{sql-gateway.sh}} and the SQL Client in gateway
> mode.
> {code:sql}
> CREATE CATALOG mt_cat WITH ('type' = 'test-filesystem', 'path' =
> '/tmp/flip578/catalog', 'default-database' = 'mydb');
> USE CATALOG mt_cat;
> CREATE TABLE orders (user_id BIGINT NOT NULL, amount BIGINT, order_time
> TIMESTAMP(3), ds STRING)
> WITH ('format' = 'json', 'source.monitor-interval' = '10s');
> CREATE TABLE spending_json (user_id BIGINT NOT NULL, total_amount BIGINT)
> WITH ('format' = 'json');
> -- 'json' cannot store the updates of a GROUP BY, so the refresh job fails to
> start
> CREATE OR ALTER MATERIALIZED TABLE spending_json
> WITH ('format' = 'json')
> FRESHNESS = INTERVAL '30' SECOND
> AS SELECT user_id, SUM(amount) AS total_amount FROM orders GROUP BY user_id;
> -- TableException: Table sink 'mt_cat.mydb.spending_json' doesn't support
> consuming update changes ...
> -- spending_json is now a materialized table with refresh-status SUSPENDED
> ALTER MATERIALIZED TABLE spending_json RESUME;
> -- java.lang.NullPointerException: Cannot read the array length because "buf"
> is null
> {code}
> ||Statement on the SUSPENDED table||Expected||Actual||
> |{{ALTER MATERIALIZED TABLE spending_json RESUME}}|Starts the refresh job, or
> fails with its error|{{NullPointerException}}|
> |{{ALTER MATERIALIZED TABLE spending_json SET ('format' =
> 'debezium-json')}}|Changes the option|{{NullPointerException}}|
> |{{CREATE OR ALTER ...}} with {{'format' = 'debezium-json'}}|Changes the
> option and starts the refresh job|{{NullPointerException}}|
> |{{DROP MATERIALIZED TABLE spending_json}}|Drops the table|Drops the table
> and deletes its data|
> Two other triggers reach the same state:
> * A conversion without a {{WITH}} clause. The old table's options are not
> carried over, so a test-filesystem table loses {{'format'}}: {{Missing
> required options are: format}}.
> * {{table.dml-sync = true}}. The statement hangs, because the refresh job
> inherits the setting. When the client disconnects, the job is cancelled with
> an {{InterruptedException}}, and the table is left SUSPENDED without a
> handler.
> Gateway log for {{ALTER ... SET}}:
> {noformat}
> java.lang.NullPointerException: Cannot read the array length because "buf" is
> null
> at
> java.base/java.io.ByteArrayInputStream.<init>(ByteArrayInputStream.java:109)
> at
> org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:467)
> at
> org.apache.flink.table.refresh.ContinuousRefreshHandlerSerializer.deserialize(ContinuousRefreshHandlerSerializer.java:42)
> at
> org.apache.flink.table.gateway.service.materializedtable.MaterializedTableManager.deserializeContinuousHandler(MaterializedTableManager.java:1258)
> at
> org.apache.flink.table.gateway.service.materializedtable.MaterializedTableManager.generateResetSavepointTableChange(MaterializedTableManager.java:1045)
> at
> org.apache.flink.table.gateway.service.materializedtable.MaterializedTableManager.callAlterMaterializedTableChangeOperation(MaterializedTableManager.java:1008)
> {noformat}
> {{RESUME}} fails in {{resumeContinuousRefreshJob}}
> ({{MaterializedTableManager.java:579}}) on the same call.
> h3. Cause
> On failure, {{convertTableToMaterializedTableInContinuousMode}} calls
> {{suspendMaterializedTable}}, which only changes the refresh status.
> {{getSerializedRefreshHandler()}} stays {{null}}. These paths deserialize it
> without a check:
> * {{resumeContinuousRefreshJob}}, line 579
> * {{callAlterMaterializedTableChangeOperation}}, SUSPENDED branch, through
> {{generateResetSavepointTableChange}}, line 1008
> * In full mode, {{resumeRefreshWorkflow}}, line 633, and
> {{deleteRefreshWorkflow}}, line 1149. By the code, a full-mode table left
> SUSPENDED by a failed workflow creation cannot even be dropped. We did not
> reproduce this. Freshness is validated at planning, and we found no other way
> to make the workflow creation fail.
> {{MaterializedTableConversionITCase#testConvertTableLeftSuspendedWhenRefreshJobFailsToStart}}
> checks the SUSPENDED status but never resumes. The code is the same on
> {{master}} and {{release-2.4}}.
> h3. Proposal
> Treat a SUSPENDED table without a refresh handler as never started:
> * {{RESUME}} in continuous mode starts the job without a restore path. In
> full mode it creates the workflow and stores the handler, as
> {{createPeriodicRefreshWorkflow}} already does.
> * {{ALTER}} on a SUSPENDED continuous table skips the savepoint reset when
> there is no handler.
> * {{DROP}} of a full-mode table skips {{deleteRefreshWorkflow}} when there is
> no handler.
> * Extend the ITCase with {{RESUME}}, {{ALTER ... SET}} and {{DROP}} after the
> failure.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
