# Resolve Statement Issues in Confluent Cloud for Apache Flink Identify and resolve Flink SQL query inefficiencies in Confluent Cloud for Apache Flink® by acting on Statement Advisor warnings at submission time and by diagnosing statements that enter a `DEGRADED` state at runtime. The sections that follow map each common warning and failure mode to its root cause and a concrete remediation, including primary key mismatches, non-deterministic updates, and resource-bound execution. ## Statement enters DEGRADED state When a Flink statement is unable to make consistent progress, it can enter a DEGRADED state. This typically occurs due to performance bottlenecks or resource constraints. You might see the following error message: ```none Your Flink statement has entered a Degraded state because it is unable to make consistent progress. This can be caused by inefficient query logic or insufficient compute resources. Please review your statement for performance bottlenecks. If the issue persists, consider scaling your compute pool or contacting Confluent support for assistance. ``` To resolve a DEGRADED state, follow these steps: 1. **Check for Statement Advisor warnings**: Review and resolve any warnings the statement advisor returns during query submission. If you’re unsure whether Flink produced warnings, run your query with the [EXPLAIN](../reference/statements/explain.md#flink-sql-explain) statement to see whether Flink generates any warnings. 2. **Profile your query**: Use the [Query Profiler](profile-query.md#flink-sql-profile-query) to identify performance bottlenecks and data flow issues in your statement. 3. **Review compute resources**: Check whether your compute pool has reached its maximum CFU limit. If so, consider: - Increasing the maximum CFU limit for your compute pool. - Moving the statement to a dedicated compute pool with more CFU capacity. - Optimizing your query to reduce resource consumption. 4. **Optimize query logic**: Based on the warnings and profiling results, implement the specific optimizations described in the following warning sections. ## Primary key differs from derived upsert key ```none [Warning] The primary key "" does not match the upsert key "" that is derived from the query. If the primary key and upsert key don't match, the system needs to add a state-intensive operation for correction, which can result in a DEGRADED statement and higher CFU consumption. If possible, revisit the table declaration with the primary key or change your query. For more information, see https://cnfl.io/primary_vs_upsert_key. ``` This warning occurs when you insert data into a table where the table’s defined `PRIMARY KEY` doesn’t align with the key columns derived from the `INSERT INTO ... SELECT` or `CREATE TABLE ... AS SELECT` query’s grouping or source. When the keys mismatch, Flink must introduce an expensive internal operator (`UpsertMaterialize`) to ensure correctness, which consumes more state and resources. The following example illustrates a query that triggers this warning: ```sql -- Create a table to store customer total orders CREATE TABLE customer_orders ( total_orders INT PRIMARY KEY NOT ENFORCED, -- Primary Key is total_orders customer_name STRING ); -- Insert aggregated order counts per customer INSERT INTO customer_orders SELECT SUM(order_count), customer_name -- Upsert key derived from GROUP BY is customer_name FROM ( VALUES ('Bob', 2), -- Bob placed 2 orders ('Alice', 1), -- Alice placed 1 order ('Bob', 2) -- Bob placed 2 more orders ) AS OrderData(customer_name, order_count) GROUP BY customer_name; ``` To resolve this warning: Align primary key : Modify the `PRIMARY KEY` definition in your `CREATE TABLE` statement to match the columns used to uniquely identify rows in your `INSERT` query (often the `GROUP BY` columns). In the previous example, changing the primary key to `customer_name` resolves the warning. Modify query : Adjust your `INSERT INTO ... SELECT` query so the selected columns or grouping aligns with the existing primary key definition. This might involve changing the `GROUP BY` clause or the columns you select. Check for warnings : If you’re unsure whether your query produces this warning, run it with the [EXPLAIN](../reference/statements/explain.md#flink-sql-explain) statement to see whether Flink generates any warnings. ## High state operator without state TTL ```none [Warning] Your query includes one or more highly state-intensive operators but does not set a time-to-live (TTL) value, which means that the system potentially needs to store an infinite amount of state. This can result in a DEGRADED statement and higher CFU consumption. If possible, change your query to use a different operator, or set a time-to-live (TTL) value. For more information, see https://cnfl.io/high_state_intensive_operators. ``` Certain SQL operations, such as joins on unbounded streams or aggregations without windowing, require Flink to maintain internal state. If you don’t configure this state to expire (using a Time-To-Live or TTL setting), it can grow indefinitely, leading to excessive memory usage, performance degradation, and higher costs. The following example illustrates a query that triggers this warning: ```sql -- Joining two unbounded streams without TTL SELECT c.*, o.* FROM `examples`.`marketplace`.`clicks` c INNER JOIN `examples`.`marketplace`.`orders` o ON c.user_id = o.customer_id; ``` To resolve this warning: Set state TTL : Configure a state time-to-live (TTL) for the table(s) involved in the stateful operation. This ensures that Flink automatically clears state older than the specified duration. You can apply this to the full statement by using the [SET ‘sql.state-ttl’](../reference/statements/set.md#flink-sql-set-statement) option, or to individual tables by using [State TTL Hints](../reference/statements/hints.md#flink-sql-hints). Use windowed operations : If applicable, rewrite your query to use windowed operations, such as windowed joins or windowed aggregations, instead of unbounded operations. Windows limit the amount of state required inherently. Refactor query : Analyze whether the stateful operation is necessary or whether you can change the query logic to avoid large state requirements. Check for warnings : If you’re unsure whether your query produces this warning, run it with the [EXPLAIN](../reference/statements/explain.md#flink-sql-explain) statement to see whether Flink generates any warnings. ## Missing `window_start` or `window_end` in GROUP BY for window aggregation ```none [Warning] Your query contains only "window_end" in the GROUP BY clause, with no corresponding "window_start". This means that the query is considered a regular aggregation query and not a windowed aggregation, which can result in unexpected, continuously updating output and higher CFU consumption. if you want a windowed aggregation in your query, ensure that you include both "window_start" and "window_end" in the GROUP BY clause. For more information, see https://cnfl.io/regular_vs_window_aggregation. ``` A similar warning appears if only `window_start` is included without `window_end`. When performing windowed aggregations using functions such as `TUMBLE`, `HOP`, `CUMULATE`, `SESSION`, you typically group by the window boundaries (`window_start` and `window_end`) along with any other grouping keys. If you include only one of the window boundary columns, either `window_start` or `window_end`, in the `GROUP BY` clause, Flink interprets this as a regular, non-windowed aggregation. This leads to continuously updating results for each input row rather than a single result per window, which is usually not the intended behavior and can consume more resources. The following example illustrates a query that triggers this warning: ```mysql -- Incorrect GROUP BY for TUMBLE window SELECT window_end, SUM(price) as `sum` FROM TUMBLE(TABLE `examples`.`marketplace`.`orders`, DESCRIPTOR($rowtime), INTERVAL '10' MINUTES) GROUP BY window_end; -- Missing window_start ``` To resolve this warning when it occurs in a query: Include both window boundaries : When performing windowed aggregations, ensure that your `GROUP BY` clause includes *both* `window_start` and `window_end`. Check for warnings : If you’re unsure whether your query produces this warning, run it with the [EXPLAIN](../reference/statements/explain.md#flink-sql-explain) statement to see whether Flink generates any warnings. The following example shows the revised query that resolves this warning: ```mysql -- Correct GROUP BY for TUMBLE window SELECT window_start, window_end, SUM(price) as `sum` FROM TUMBLE(TABLE `examples`.`marketplace`.`orders`, DESCRIPTOR($rowtime), INTERVAL '10' MINUTES) GROUP BY window_start, window_end; -- Includes both window boundaries ``` ## Session window without a PARTITION BY key ```none [Warning] Your query uses a SESSION window without a PARTITION BY clause. This results in all data being processed by a single, non-parallel task, which can create a significant bottleneck, leading to poor performance and high resource consumption. To improve performance and enable parallel execution, specify a PARTITION BY key in your SESSION window. For more information, see https://cnfl.io/session_without_partioning. ``` When you use a SESSION window, Flink groups data into sessions based on periods of activity, separated by a specified inactivity gap. If you don’t include a PARTITION BY clause, Flink sends all data to a single, non-parallel task to correctly identify these sessions. This creates a significant performance bottleneck and prevents the query from scaling. The following example shows a query that triggers this warning: ```mysql -- This query uses a SESSION window without a PARTITION BY key SELECT * FROM SESSION( TABLE `examples`.`marketplace`.`orders`, DESCRIPTOR($rowtime), INTERVAL '5' MINUTES ); ``` To resolve this warning: Add a PARTITION BY key : Modify your SESSION window definition to include a PARTITION BY clause. This partitions the data by the specified key(s), allowing Flink to perform sessionization independently and in parallel for each partition. This is important for performance and scalability. Check for warnings : If you’re unsure whether your query produces this warning, run it with the [EXPLAIN](../reference/statements/explain.md#flink-sql-explain) statement to see whether Flink generates any warnings. The following example shows the revised query that resolves the warning: ```mysql -- Corrected query with PARTITION BY to enable parallel execution SELECT * FROM SESSION( TABLE `examples`.`marketplace`.`orders` PARTITION BY customer_id, DESCRIPTOR($rowtime), INTERVAL '5' MINUTES ); ``` ## Related content - [Profile a Query](profile-query.md#flink-sql-profile-query) - [EXPLAIN Statement](../reference/statements/explain.md#flink-sql-explain) - [SET](../reference/statements/set.md#flink-sql-set-statement) - [HINTS](../reference/statements/hints.md#flink-sql-hints) - [Window Aggregation](../reference/queries/window-aggregation.md#flink-sql-window-aggregation) - [Window TopN](../reference/queries/window-topn.md#flink-sql-window-top-n) - [Window Join](../reference/queries/window-join.md#flink-sql-window-join) - [Window Deduplication](../reference/queries/window-deduplication.md#flink-sql-window-deduplication) - [Interval join](../reference/queries/joins.md#flink-sql-interval-joins) - [Temporal join](../reference/queries/joins.md#flink-sql-temporal-joins) #### NOTE This website includes content developed at the [Apache Software Foundation](https://www.apache.org/) under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).