You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Francis (Jira)" <ji...@apache.org> on 2025/10/29 08:33:00 UTC
[jira] [Commented] (FLINK-38567) ForSt state backend error with GROUP BY and async state execution
[ https://issues.apache.org/jira/browse/FLINK-38567?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18033765#comment-18033765 ]
Francis commented on FLINK-38567:
---------------------------------
I'm running into a potentially similar issue using the DataStream API. The full exception is:
{code:java}
java.io.IOException: Error while opening ForSt instance. at org.apache.flink.state.forst.ForStOperationUtils.openDB(ForStOperationUtils.java:97) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.restore.ForStHandle.loadDb(ForStHandle.java:140) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.restore.ForStHandle.openDB(ForStHandle.java:118) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.restore.ForStNoneRestoreOperation.restore(ForStNoneRestoreOperation.java:64) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.ForStKeyedStateBackendBuilder.build(ForStKeyedStateBackendBuilder.java:261) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.ForStStateBackend.createAsyncKeyedStateBackend(ForStStateBackend.java:474) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.ForStStateBackend.createAsyncKeyedStateBackend(ForStStateBackend.java:98) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$3(StreamTaskStateInitializerImpl.java:475) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:173) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:137) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:486) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:195) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:280) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreStateAndGates(StreamTask.java:858) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$restoreInternal$5(StreamTask.java:812) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:812) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:771) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:963) [flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932) [flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:756) [flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.runtime.taskmanager.Task.run(Task.java:568) [flink-dist-2.1.0.jar:2.1.0] at java.base/java.lang.Thread.run(Unknown Source) [?:?]Caused by: org.forstdb.RocksDBException: NotFound at org.forstdb.RocksDB.open(Native Method) ~[flink-dist-2.1.0.jar:2.1.0] at org.forstdb.RocksDB.open(RocksDB.java:318) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.ForStOperationUtils.openDB(ForStOperationUtils.java:85) ~[flink-dist-2.1.0.jar:2.1.0] ... 23 more2025-10-29 08:30:43,965 WARN org.apache.flink.streaming.api.operators.BackendRestorerProcedure [] - Exception while restoring keyed state backend for AsyncKeyedCoProcessOperator_84984c1fa2897a7a40e728e4e87e1da9_(1/1) from alternative (1/1), will retry while more alternatives are available.org.apache.flink.runtime.state.BackendBuildingException: Caught unexpected exception. at org.apache.flink.state.forst.ForStKeyedStateBackendBuilder.build(ForStKeyedStateBackendBuilder.java:319) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.ForStStateBackend.createAsyncKeyedStateBackend(ForStStateBackend.java:474) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.ForStStateBackend.createAsyncKeyedStateBackend(ForStStateBackend.java:98) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$3(StreamTaskStateInitializerImpl.java:475) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:173) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:137) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:486) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:195) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:280) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreStateAndGates(StreamTask.java:858) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$restoreInternal$5(StreamTask.java:812) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:812) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:771) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:963) [flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932) [flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:756) [flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.runtime.taskmanager.Task.run(Task.java:568) [flink-dist-2.1.0.jar:2.1.0] at java.base/java.lang.Thread.run(Unknown Source) [?:?]Caused by: java.io.IOException: Error while opening ForSt instance. at org.apache.flink.state.forst.ForStOperationUtils.openDB(ForStOperationUtils.java:97) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.restore.ForStHandle.loadDb(ForStHandle.java:140) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.restore.ForStHandle.openDB(ForStHandle.java:118) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.restore.ForStNoneRestoreOperation.restore(ForStNoneRestoreOperation.java:64) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.ForStKeyedStateBackendBuilder.build(ForStKeyedStateBackendBuilder.java:261) ~[flink-dist-2.1.0.jar:2.1.0] ... 19 moreCaused by: org.forstdb.RocksDBException: NotFound at org.forstdb.RocksDB.open(Native Method) ~[flink-dist-2.1.0.jar:2.1.0] at org.forstdb.RocksDB.open(RocksDB.java:318) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.ForStOperationUtils.openDB(ForStOperationUtils.java:85) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.restore.ForStHandle.loadDb(ForStHandle.java:140) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.restore.ForStHandle.openDB(ForStHandle.java:118) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.restore.ForStNoneRestoreOperation.restore(ForStNoneRestoreOperation.java:64) ~[flink-dist-2.1.0.jar:2.1.0] at org.apache.flink.state.forst.ForStKeyedStateBackendBuilder.build(ForStKeyedStateBackendBuilder.java:261) ~[flink-dist-2.1.0.jar:2.1.0] ... 19 more {code}
I'm seeing this when using the Presto S3 filesystem plugin and starting a job without an existing state store.
> ForSt state backend error with GROUP BY and async state execution
> -----------------------------------------------------------------
>
> Key: FLINK-38567
> URL: https://issues.apache.org/jira/browse/FLINK-38567
> Project: Flink
> Issue Type: Bug
> Components: Runtime / State Backends
> Affects Versions: 2.0.0, 2.1.0
> Reporter: Mohsen Rezaei
> Priority: Critical
>
> While testing the disaggregate state backend with ForSt DB I ran into the following exception:
> {code}
> java.lang.Exception: Exception while creating StreamOperatorStateContext.
> at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:359)
> at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:280)
> at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreStateAndGates(StreamTask.java:858)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$restoreInternal$5(StreamTask.java:812)
> at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:812)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:771)
> at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:963)
> at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932)
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:756)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:568)
> at java.base/java.lang.Thread.run(Unknown Source)
> Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for AsyncKeyedProcessOperator_dc09592d5320e3ce8d45932dc9094772_(1/1) from any of the 1 provided restore options.
> at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:165)
> at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:486)
> at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:195)
> ... 12 common frames omitted
> Caused by: org.apache.flink.runtime.state.BackendBuildingException: Caught unexpected exception.
> at org.apache.flink.state.forst.ForStKeyedStateBackendBuilder.build(ForStKeyedStateBackendBuilder.java:319)
> at org.apache.flink.state.forst.ForStStateBackend.createAsyncKeyedStateBackend(ForStStateBackend.java:474)
> at org.apache.flink.state.forst.ForStStateBackend.createAsyncKeyedStateBackend(ForStStateBackend.java:98)
> at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$3(StreamTaskStateInitializerImpl.java:475)
> at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:173)
> at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:137)
> ... 14 common frames omitted
> Caused by: java.io.IOException: Error while opening ForSt instance.
> at org.apache.flink.state.forst.ForStOperationUtils.openDB(ForStOperationUtils.java:97)
> at org.apache.flink.state.forst.restore.ForStHandle.loadDb(ForStHandle.java:140)
> at org.apache.flink.state.forst.restore.ForStHandle.openDB(ForStHandle.java:118)
> at org.apache.flink.state.forst.restore.ForStNoneRestoreOperation.restore(ForStNoneRestoreOperation.java:64)
> at org.apache.flink.state.forst.ForStKeyedStateBackendBuilder.build(ForStKeyedStateBackendBuilder.java:261)
> ... 19 common frames omitted
> Caused by: org.forstdb.RocksDBException: NotFound
> at org.forstdb.RocksDB.open(Native Method)
> at org.forstdb.RocksDB.open(RocksDB.java:318)
> at org.apache.flink.state.forst.ForStOperationUtils.openDB(ForStOperationUtils.java:85)
> ... 23 common frames omitted
> {code}
> The Flink SQL query I used is a simple {{GROUP BY}} operation:
> {code}
> SELECT COUNT(page) as page_count, userid FROM user_audit GROUP BY userid;
> {code}
> This is the configuration used to reproduce this:
> {code}
> execution.state-recovery.from-local, false
> jobmanager.adaptive-scheduler.submission.resource-stabilization-timeout, 30s
> taskmanager.memory.jvm-metaspace.size, 200m
> kubernetes.cluster-id, q-64dfbe96-e7e2-4991-8b95-a5906fd9b3c7
> high-availability.storageDir, s3p://...
> execution.checkpointing.savepoint-dir, s3p://...
> metrics.internal.query-service.port, 6127
> kubernetes.namespace, q-64dfbe96-e7e2-4991-8b95-a5906fd9b3c7
> state.backend.forst.writebuffer.size, 4mb
> metrics.reporters, prom
> state.backend.forst.metrics.estimate-num-keys, true
> state.backend.type, forst
> table.exec.async-state.enabled, true
> execution.checkpointing.mode, EXACTLY_ONCE
> metrics.reporter.prom.port, 9999
> taskmanager.memory.process.size, 1024m
> execution.checkpointing.incremental, true
> taskmanager.memory.network.fraction, 0.08
> jobmanager.rpc.port, 6123
> web.upload.dir, /var/run/flink-app-jar
> execution.checkpointing.interval, 30000
> execution.checkpointing.timeout, 60000
> rest.port, 8081
> s3.upload.max.concurrent.uploads, 10
> restart-strategy.fixed-delay.delay, 3s
> state.backend.forst.compaction.level.max-size-level-base, 8mb
> taskmanager.memory.managed.fraction, 0.35
> blob.server.port, 6124
> state.backend.forst.compaction.level.target-file-size-base, 4mb
> metrics.system-resource, true
> high-availability.type, KUBERNETES
> state.backend.forst.local-dir, /var/run/flink-forst-local
> taskmanager.cpu.cores, 1
> metrics.reporter.prom.factory.class, org.apache.flink.metrics.prometheus.PrometheusReporterFactory
> state.backend.forst.writebuffer.count, 16
> query.server.port, 6125
> taskmanager.memory.jvm-overhead.min, 128m
> s3.endpoint, s3.us-west-2.amazonaws.com
> state.backend.forst.metrics.compaction-pending, true
> taskmanager.data.port, 6121
> taskmanager.numberOfTaskSlots, 1
> execution.checkpointing.dir, s3p://...
> taskmanager.memory.network.min, 0m
> restart-strategy.fixed-delay.attempts, 10
> jobmanager.memory.process.size, 1024m
> taskmanager.rpc.port, 6122
> table.exec.mini-batch.enabled, false
> jobmanager.scheduler, Adaptive
> high-availability, org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
> table.optimizer.agg-phase-strategy, ONE_PHASE
> restart-strategy, fixed-delay
> execution.checkpointing.snapshot-compression, true
> state.backend.forst.metrics.num-running-compactions, true
> s3.path.style.access, true
> {code}
> Based on [documentation|https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/disaggregated_state/#for-sql-jobs], a non-distinct aggregation should already be supported for asynchronous state access, and it seems like I've ran into a bug with the async disaggregated state. I haven't tested any of the other operators, e.g. windowing, but I confirmed that a simple projection works just fine with the same configuration.
> If the {{table.exec.async-state.enabled}} config is set to {{false}}, the jobs are able to make progress.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)