ksqldb on Confluent Cloud seems to inherit the upstream behavior for TS extraction (even if it should not) resulting in Null Pointer Exception

[geogmg]

Describe the bug
We are using ksqldb session windows for some use cases. We've noticed that any select * from table_name emit changes; query on cloud console was throwing a lot of null pointer exceptions in the processing log topic/stream.

{
  "LOGGER": "processing.transient_public_table_2_5927611079956313424.KsqlTopic.Source",
  "LEVEL": "ERROR",
  "TIME": 1700151893065,
  "MESSAGE": {
    "TYPE": 1,
    "DESERIALIZATIONERROR": null,
    "RECORDPROCESSINGERROR": {
      "ERRORMESSAGE": "Failed to extract timestamp from row",
      "RECORD": "key:[['6638392.2023-11-13T17:25:00.000+13:00']@1700151783904/1700151783904], value:null",
      "CAUSE": [
        "java.lang.NullPointerException"
      ]
    },
    "PRODUCTIONERROR": null,
    "SERIALIZATIONERROR": null,
    "KAFKASTREAMSTHREADERROR": null
  }
}

We've noticed that the mesage key corresponds to the null messages in the output topic, which are potentially populated by ksqldb to delete older/smaller sessions when two sessions are merged.

It seems to inherit the upstream behavior (even if it should not) to extract timestamp from a message attribute

To Reproduce
Steps to reproduce the behavior, include:

1. Use Confluent Cloud.
2. Create session windows which could create null messages in output topic to merge older/smaller sessions
3. Execute select * from table_name emit changes; and observe processing logs

Expected behavior
No error logs in processing log topic/stream

Actual behaviour
RecordProcessingError > Null pointer exception in processing log topic/stream