[FLINK] CI failed due to nexmark q12 ut `NPE`

[KevinyhZou]

Backend

VL (Velox)

Bug description

Exception message:

10:49:34.510 [WindowAggregate[29] -> Writer -> nexmark_q12[30]: Writer (1/1)#0] WARN  org.apache.flink.runtime.taskmanager.Task  - WindowAggregate[29] -> Writer -> nexmark_q12[30]: Writer (1/1)#0 (8e336fa1f63b9e8650e42aedd249e5f7_337adade1e207453ed3502e01d75fd03_0_0) switched from RUNNING to FAILED with failure cause:
java.lang.NullPointerException: Cannot invoke "org.apache.gluten.table.runtime.operators.GlutenSessionResource.getSession()" because the return value of "java.util.Map.get(Object)" is null
	at org.apache.gluten.table.runtime.operators.GlutenSessionResources.getSession(GlutenSessionResources.java:104)
	at org.apache.gluten.table.runtime.typeutils.GlutenStatefulRecordSerializer.deserialize(GlutenStatefulRecordSerializer.java:71)
	at org.apache.gluten.table.runtime.typeutils.GlutenStatefulRecordSerializer.deserialize(GlutenStatefulRecordSerializer.java:36)
	at org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.deserialize(StreamElementSerializer.java:206)
	at org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.deserialize(StreamElementSerializer.java:44)
	at org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate.read(NonReusingDeserializationDelegate.java:53)
	at org.apache.flink.runtime.io.network.api.serialization.NonSpanningWrapper.readInto(NonSpanningWrapper.java:337)
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.readNonSpanningRecord(SpillingAdaptiveSpanningRecordDeserializer.java:128)
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.readNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:103)
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.getNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:93)
	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:102)
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:579)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:909)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:858)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:751)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)

Gluten version

No response

Spark version

None

Spark configurations

No response

System information

No response

Relevant logs

[KevinyhZou]

GlutenSessionResources hold the session of related opeator, but it doesn't consider the thread-safe problems under multiple parallelism