Hi Everyone, I tried running my stream join job on Kinesis and I think I may be having some memory issues. Locally, my job runs fine with

memory.size

set to 4096. However, I can't set config properties on kinesis because kinesis does not allow

flink-conf.yaml

. This is the error stack trace below. Does any have suggestions on how to resolve this?

2023-03-09 12:40:24
java.lang.Exception: Exception while creating StreamOperatorStateContext.
	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:255)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:268)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:700)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:676)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:643)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:922)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:568)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for WindowOperator_7e145440e6a4bf9d4540b5d7293494cc_(1/1) from any of the 1 provided restore options.
	at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160)
	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:346)
	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:164)
	... 11 more
Caused by: java.io.IOException: Failed to acquire shared cache resource for RocksDB
	at org.apache.flink.contrib.streaming.state.RocksDBOperationUtils.allocateSharedCachesIfConfigured(RocksDBOperationUtils.java:296)
	at org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:444)
	at org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:97)
	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:329)
	at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)
	at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
	... 13 more
Caused by: java.lang.IllegalArgumentException: The fraction of memory to allocate should not be 0. Please make sure that all types of managed memory consumers contained in the job are configured with a non-negative weight via `taskmanager.memory.managed.consumer-weights`.
	at org.apache.flink.util.Preconditions.checkArgument(Preconditions.java:160)
	at org.apache.flink.runtime.memory.MemoryManager.validateFraction(MemoryManager.java:672)
	at org.apache.flink.runtime.memory.MemoryManager.computeMemorySize(MemoryManager.java:653)
	at org.apache.flink.runtime.memory.MemoryManager.getSharedMemoryResourceForManagedMemory(MemoryManager.java:521)
	at org.apache.flink.contrib.streaming.state.RocksDBOperationUtils.allocateSharedCachesIfConfigured(RocksDBOperationUtils.java:292)
	... 18 more

when you say

kinesis

are you referring to kinesis data analytics?

The fraction of memory to allocate should not be 0.

Was this set in your flink-conf.yaml?

yes kinesis data analytics

hi @Reme Ajayi could you create a support ticket on this? KDA Engineering could have a look as to why you’d be seeing this

I've done that

but out of curiosity what is the job doing that warrants custom memory settings?

It's a complex join.

got it--what is your expected throughput and how many KPUs?

I'm actually relatively new to Flink so I don't know if I should be using more memory, however I am able to run my job locally without any issues when I update the memory size.

we generally recommend that for each 1MB of data ingested (per second) you should allocate 1 KPU. Since you have a few GB in your topics that will likely all get pulled in to KDA all at once this is likely underprovisioned

Oh Wow. Thank you for this explanation

definitely don’t want to blow out your bill--perhaps we can understand your architecture better

No it's a streaming application

ideally flink wouldn’t pull in several GBs all at once into a streaming app

This is what the application architecture looks like in my local environment

Ah okay so after that 6 hours it threw the error right

No, it ran for 6 hours locally. But it threw the error as soon as it started on KDA

Ah okay is this screenshot from local or KDA

The screenshot is from local, I was trying to show you the architecture.

Got it got it. I'm not sure about the error but it could be underprovsioned. Take a look at the containerCPUUtilization and containedMemoryUtilization of your application in cloud watch this may give us more info

I can't get a CPUUtilization values cause it's not running at all

Hmm okay. Are you restoring from save points when you start the app or just starting fresh

starting fresh

Ok I'll have service team review ticket

thank you

ah okay--do you have autoscaling turned on?

Yes I do

so this app may be autoscaling given that. Can you show containerCPUUtilization as well?

ahh--let’s try setting the PPKPU down to 1, we generally only recommend this going higher with I/O bound applications

Locally, the sources are receiving about 3MB per second if this provides any more insight

you can leave parallelism at 32 for now but i think that’s likely too much for 3MB/s--is that across all 4 topics?

Oh no, this is a smaller job with just two topics. It's giving the same error though, the other job probably has 4x as much

sounds good--will follow up here once the ticket is resolved 🙂

Hey @Jeremy Ber, AWS reached out with a solution, the problem was that I set the stateBackend which Kinesis does not allow.

Excellent!

The default is fine thank you