Hey Folks, I’m using Table API to consume proto messages from Kafka and save data to S3 object storage. I’m basically partitioning data for an event_id, event_name and a day. I’m getting

java.lang.OutOfMemoryError: Java heap space

. I have setup 2 TaskManagers of 6GB memory, I’m expecting ~50K records every hour. I set the JVM Heap Size to 4.5 GB. I suspect RocksDB not being used because Managed Memory never gets uitilized. But, I’m setting the same in my flink configurations -

state.backend: rocksdb

. How can I address this issue? Here is the complete error:

java.lang.OutOfMemoryError: Java heap space

at com.esotericsoftware.kryo.Serializer.copy(Serializer.java:86)

at com.esotericsoftware.kryo.Kryo.copy(Kryo.java:862)

at com.esotericsoftware.kryo.serializers.UnsafeCacheFields$UnsafeObjectField.copy(UnsafeCacheFields.java:297)

at com.esotericsoftware.kryo.serializers.FieldSerializer.copy(FieldSerializer.java:634)

at com.esotericsoftware.kryo.Kryo.copy(Kryo.java:862)

at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:306)

at org.apache.flink.api.common.typeutils.base.ListSerializer.copy(ListSerializer.java:102)

at org.apache.flink.api.common.typeutils.base.ListSerializer.copy(ListSerializer.java:42)

at org.apache.flink.api.common.typeutils.base.MapSerializer.copy(MapSerializer.java:111)

at org.apache.flink.api.common.typeutils.base.MapSerializer.copy(MapSerializer.java:43)

at org.apache.flink.runtime.state.ArrayListSerializer.copy(ArrayListSerializer.java:70)

at org.apache.flink.runtime.state.PartitionableListState.<init>(PartitionableListState.java:65)

at org.apache.flink.runtime.state.PartitionableListState.deepCopy(PartitionableListState.java:79)

at org.apache.flink.runtime.state.DefaultOperatorStateBackendSnapshotStrategy.syncPrepareResources(DefaultOperatorStateBackendSnapshotStrategy.java:77)

at org.apache.flink.runtime.state.DefaultOperatorStateBackendSnapshotStrategy.syncPrepareResources(DefaultOperatorStateBackendSnapshotStrategy.java:36)

at org.apache.flink.runtime.state.SnapshotStrategyRunner.snapshot(SnapshotStrategyRunner.java:77)

at org.apache.flink.runtime.state.DefaultOperatorStateBackend.snapshot(DefaultOperatorStateBackend.java:230)

at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:230)

at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:173)

at org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:336)

at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.checkpointStreamOperator(RegularOperatorChain.java:228)

at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.buildOperatorSnapshotFutures(RegularOperatorChain.java:213)

at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.snapshotState(RegularOperatorChain.java:192)

at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.takeSnapshotSync(SubtaskCheckpointCoordinatorImpl.java:715)

at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:350)

at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$14(StreamTask.java:1299)

at org.apache.flink.streaming.runtime.tasks.StreamTask$$Lambda$1819/0x0000000840cbac40.run(Unknown Source)

at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)

at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:1287)

at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:1244)

at <http://org.apache.flink.streaming.runtime.io|org.apache.flink.streaming.runtime.io>.checkpointing.CheckpointBarrierHandler.notifyCheckpoint(CheckpointBarrierHandler.java:147)

at <http://org.apache.flink.streaming.runtime.io|org.apache.flink.streaming.runtime.io>.checkpointing.SingleCheckpointBarrierHandler.triggerCheckpoint(SingleCheckpointBarrierHandler.java:287)