I found my Flink job is strange. The task heap was being used, but it didn't consume any data since it started.
It's my code, a stateful processing with three state:
var listState: ListState[String] = _
var valueState: ValueState[TestModel] = _
var mapState: MapState[String, TestModel] = _
override def open(parameters: Configuration): Unit = {
val descriptor = new ListStateDescriptor[String](
"buffered-elements",
createTypeInformation[String]
)
listState = getRuntimeContext.getListState(descriptor)
val adStateDescriptor = new ValueStateDescriptor("LastEventState", createTypeInformation[TestModel])
valueState = getRuntimeContext.getState(adStateDescriptor)
val mapStateDescriptor = new MapStateDescriptor("SubsegmentState",
BasicTypeInfo.STRING_TYPE_INFO, createTypeInformation[TestModel])
mapState = getRuntimeContext.getMapState(mapStateDescriptor)
}
override def map(in: TestModel): TestModel = {
......
}
TaskManager memory is 6GB, and the Task heap size is 2.3GB