Hi all, Getting this error while using ForSt, any idea how to fix this?

StreamOperatorStateContext.
	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:361)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:284)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreStateAndGates(StreamTask.java:866)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$restoreInternal$5(StreamTask.java:820)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:820)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:779)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:973)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:945)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:760)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:569)
	at java.base/java.lang.Thread.run(Unknown Source)
Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for AsyncKeyedProcessOperator_f199ad3ac3e86e5fc9372ebb239a06cb_(27/64) 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:488)
	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:197)
	... 12 more
Caused by: org.apache.flink.runtime.state.BackendBuildingException: Caught unexpected exception.
	at org.apache.flink.state.forst.ForStKeyedStateBackendBuilder.build(ForStKeyedStateBackendBuilder.java:315)
	at org.apache.flink.state.forst.ForStStateBackend.createAsyncKeyedStateBackend(ForStStateBackend.java:473)
	at org.apache.flink.state.forst.ForStStateBackend.createAsyncKeyedStateBackend(ForStStateBackend.java:97)
	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$3(StreamTaskStateInitializerImpl.java:477)
	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 more
Caused by: java.io.InterruptedIOException: getFileStatus on <s3://y-cdp/flinkcdp/checkpoints/shared/op_AsyncKeyedProcessOperator_f199ad3ac3e86e5fc9372ebb239a06cb__27_64__attempt_7>: com.amazonaws.SdkClientException: Unable to execute HTTP request: Timeout waiting for connection from pool
	at org.apache.hadoop.fs.s3a.S3AUtils.translateInterruptedException(S3AUtils.java:395)
	at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:201)
	at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:175)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:3799)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:3688)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.lambda$exists$34(S3AFileSystem.java:4703)
	at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.lambda$trackDurationOfOperation$5(IOStatisticsBinding.java:499)
	at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDuration(IOStatisticsBinding.java:444)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.trackDurationAndSpan(S3AFileSystem.java:2337)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.trackDurationAndSpan(S3AFileSystem.java:2356)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.exists(S3AFileSystem.java:4701)
	at org.apache.flink.fs.s3hadoop.common.HadoopFileSystem.exists(HadoopFileSystem.java:168)
	at org.apache.flink.core.fs.PluginFileSystemFactory$ClassLoaderFixingFileSystem.exists(PluginFileSystemFactory.java:149)
	at org.apache.flink.core.fs.SafetyNetWrapperFileSystem.exists(SafetyNetWrapperFileSystem.java:117)
	at org.apache.flink.state.forst.ForStResourceContainer.prepareDirectories(ForStResourceContainer.java:373)
	at org.apache.flink.state.forst.ForStResourceContainer.prepareDirectories(ForStResourceContainer.java:336)
	at org.apache.flink.state.forst.ForStKeyedStateBackendBuilder.build(ForStKeyedStateBackendBuilder.java:254)
	... 19 more
Caused by: com.amazonaws.SdkClientException: Unable to execute HTTP request: Timeout waiting for connection from pool
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleRetryableException(AmazonHttpClient.java:1245)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeHelper(AmazonHttpClient.java:1191)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.doExecute(AmazonHttpClient.java:838)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeWithTimer(AmazonHttpClient.java:805)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.execute(AmazonHttpClient.java:779)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.access$500(AmazonHttpClient.java:735)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutionBuilderImpl.execute(AmazonHttpClient.java:717)
	at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:581)
	at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:559)
	at com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:5593)
	at com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:5540)
	at com.amazonaws.services.s3.AmazonS3Client.getObjectMetadata(AmazonS3Client.java:1422)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.lambda$getObjectMetadata$10(S3AFileSystem.java:2545)
	at org.apache.hadoop.fs.s3a.Invoker.retryUntranslated(Invoker.java:414)
	at org.apache.hadoop.fs.s3a.Invoker.retryUntranslated(Invoker.java:377)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.getObjectMetadata(S3AFileSystem.java:2533)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.getObjectMetadata(S3AFileSystem.java:2513)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:3776)
	... 32 more
Caused by: org.apache.http.conn.ConnectionPoolTimeoutException: Timeout waiting for connection from pool
	at org.apache.http.impl.conn.PoolingHttpClientConnectionManager.leaseConnection(PoolingHttpClientConnectionManager.java:316)
	at org.apache.http.impl.conn.PoolingHttpClientConnectionManager$1.get(PoolingHttpClientConnectionManager.java:282)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(Unknown Source)
	at java.base/java.lang.reflect.Method.invoke(Unknown Source)
	at com.amazonaws.http.conn.ClientConnectionRequestFactory$Handler.invoke(ClientConnectionRequestFactory.java:70)
	at com.amazonaws.http.conn.$Proxy50.get(Unknown Source)
	at org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:190)
	at org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:186)
	at org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:185)
	at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:83)
	at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:56)
	at com.amazonaws.http.apache.client.impl.SdkHttpClient.execute(SdkHttpClient.java:72)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeOneRequest(AmazonHttpClient.java:1378)
	at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeHelper(AmazonHttpClient.java:1183)
	... 48 more

It looks like this is the main issue:

org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:3776)
	... 32 more
Caused by: org.apache.http.conn.ConnectionPoolTimeoutException: Timeout waiting for connection from pool

Search results for this suggested that the issue is potentially with the (default) configuration being used for the Hadoop FS s3a settings. Here is a link to a similar issue, and the same suggestion(s) make sense to follow - updating the

fs.s3a.connection.maximum

. The Flink Hadoop/Presto S3 File Systems plugins section points to the Hadoop S3A properties, and a number of properties look worthwhile to try updating as needed. Here are some settings to consider updating: • fs.s3a.attempts.maximum • fs.s3a.connection.maximum • fs.s3a.threads.max • fs.s3a.retry.limit

thanks! tried to tweek all those configurations, but always hitting some sort of problem. Now after increasing the number of maximum connections to a point where I get a sort of OOM (maybe) that looks like that: RocksDBExceptionJni::ThrowNew/StatusJni - Error: unexpected exception! │ │ flink-main-container # │ │ flink-main-container # A fatal error has been detected by the Java Runtime Environment: │ │ flink-main-container # │ │ flink-main-container # SIGSEGV (0xb) at pc=0x00007f230964fd6a, pid=24, tid=1454 │ │ flink-main-container # │ │ flink-main-container # JRE version: OpenJDK Runtime Environment Temurin-21.0.10+7 (21.0.10+7) (build 21.0.10+7-LTS) │ │ flink-main-container # Java VM: OpenJDK 64-Bit Server VM Temurin-21.0.10+7 (21.0.10+7-LTS, mixed mode, sharing, tiered, compressed class ptrs, │ │ flink-main-container # Problematic frame: │ │ flink-main-container # C [libforstjni-linux64.so+0x44fd6a] forstdb:DBImpl:MaybeScheduleFlushOrCompaction()+0xaa │ │ flink-main-container # │ │ flink-main-container # Core dump will be written. Default location: /opt/flink/core.24 │ │ flink-main-container # │ │ flink-main-container # An error report file with more information is saved as: │ │ flink-main-container # /opt/flink/hs_err_pid24.log │ │ flink-main-container # │ │ flink-main-container # If you would like to submit a bug report, please visit: │ │ flink-main-container # https://github.com/adoptium/adoptium-support/issues │ │ flink-main-container # The crash happened outside the Java Virtual Machine in native code. │ │ flink-main-container # See problematic frame for where to report the bug. important to note that I increased the memory of the job a lot (256GB memory just for testing it out)

What is the base flink image that you're using? One search result suggested that you check the base image to make sure you're using the official flink base image, and verify you're not including any external RocksDB JNI related libraries, so you can avoid using a JNI library that is incompatible. > • Ensure you are using the Flink-provided library and not accidentally introducing an external, incompatible RocksDB JNI dependency (e.g., via your job's

pom.xml

with an incorrect scope or managing dependencies externally). I also see you're using the Temurin 21.0 version. Java 21 looks like it is still listed as experimental support. I'm curious if you're able to try out Java 17 instead to see if the issue goes away. One of the Google AI search results suggested the following: > OpenJDK 21 is generally supported, but ensure you are using the latest patch version of Temurin-21 to rule out a potential minor JRE bug. And one more thing to consider - which echoes the previous tip of making sure you have the latest patch version of your JRE to avoid a potential JRE bug. > • Memory Management: The crash occurred in a

MaybeScheduleFlushOrCompaction()

function, which is a native RocksDB operation. The

SIGSEGV

might be related to memory corruption originating from a concurrency bug in the native library itself, as has been seen in past RocksDB issues. The JRE crash log (

hs_err_pid24.log

) will contain more detailed memory information which could aid debugging.

The base image I'm using is the official one. Tried downgrading to java 17, still this issue occurs randomly. I think this is somehow related to insufficient memory, but I can't find which part of the flink memory distribution is restrained and needs more space.

Have you tried running the job using hashmap state backend first, and then switching to RocksDB (or ForSt)? If the issue is specifically due to memory settings, then you can focus on the memory settings until they are somewhat known. Once you have the memory locked down, then you can try switching to RocksDB/ForSt again - but with a better idea of what your memory needs are.

Found the issue, the task managers took long too initialize because of high load, and it reached a time out so the job manager scheduler killed them. The error message we saw was due to race condition between compaction and the task manager being killed.

I'm glad you found a likely root cause!