GlobalWindowAggregate(groupBy=[stime, log_type, station, unixtime, method, scheme, domain, uri, uri_param, return_code, request_time, request_size, response_size, hit_info, remote_addr, server_addr, server_port, http_refer, content_type, user_agent, remote_port, last_modified, uuid, via_info, response_body_first_byte_time, body_bytes_sent, sent_http_content_length, http_range, sent_http_content_range, user_info, http_x_forwarded_for, country, http_version, quic], window=[TUMBLE(slice_end=[$slice_end], size=[30 s])], select=[stime, log_type, station, unixtime, method, scheme, domain, uri, uri_param, return_code, request_time, request_size, response_size, hit_info, remote_addr, server_addr, server_port, http_refer, content_type, user_agent, remote_port, last_modified, uuid, via_info, response_body_first_byte_time, body_bytes_sent, sent_http_content_length, http_range, sent_http_content_range, user_info, http_x_forwarded_for, country, http_version, quic, start('w$) AS window_start, end('w$) AS window_end]) -> Calc(select=[stime, log_type, station, unixtime, method, scheme, domain, uri, uri_param, return_code, request_time, request_size, response_size, hit_info, remote_addr, server_addr, server_port, http_refer, content_type, user_agent, remote_port, last_modified, uuid, via_info, response_body_first_byte_time, body_bytes_sent, sent_http_content_length, http_range, sent_http_content_range, user_info, http_x_forwarded_for, country, http_version, quic, CAST(LOCALTIMESTAMP()) AS push_time, server_addr AS shostname]) -> NotNullEnforcer(fields=[stime]) -> Sink: Sink(table=[default_catalog.default_database.dwm_i1_scdn_vod_ali_access_log], fields=[stime, log_type, station, unixtime, method, scheme, domain, uri, uri_param, return_code, request_time, request_size, response_size, hit_info, remote_addr, server_addr, server_port, http_refer, content_type, user_agent, remote_port, last_modified, uuid, via_info, response_body_first_byte_time, body_bytes_sent, sent_http_content_length, http_range, sent_http_content_range, user_info, http_x_forwarded_for, country, http_version, quic, push_time, shostname]) (2/4)#0 (56539a92c72703c5d93b94cc399ef30b) switched from RUNNING to FAILED with failure cause: java.lang.NullPointerException
	at org.apache.flink.table.data.GenericRowData.getLong(GenericRowData.java:154)
	at com.onething.beta.engine.formats.event.AvroEventSerializationSchema.getValue(AvroEventSerializationSchema.java:96)
	at com.onething.beta.engine.formats.event.AvroEventSerializationSchema.serialize(AvroEventSerializationSchema.java:71)
	at com.onething.beta.engine.formats.event.AvroEventSerializationSchema.serialize(AvroEventSerializationSchema.java:29)
	at org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaSerializationSchema.serialize(DynamicKafkaSerializationSchema.java:137)
	at org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaSerializationSchema.serialize(DynamicKafkaSerializationSchema.java:37)
	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.invoke(FlinkKafkaProducer.java:907)
	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.invoke(FlinkKafkaProducer.java:99)
	at org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.invoke(TwoPhaseCommitSinkFunction.java:223)
	at org.apache.flink.table.runtime.operators.sink.SinkOperator.processElement(SinkOperator.java:65)
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71