Ramin Gharib created FLINK-41350:
------------------------------------
Summary: 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
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)
