[SUPPORT] Flink KryoSerializer cannot be cast to class org.apache.flink.table.runtime.typeutils.AbstractRowDataSerializer

**Describe the problem you faced**

Hello.
I'm going to get the log data in json format from kafka and create an app that loads it into the hudi table using the hudi stream api.

Operation has been set to BULK_INSERT to load log data.

However, if you set it to BULK_INSERT, the casting problem will occur as follows.

`KryoSerializer cannot be cast to class org.apache.flink.table.runtime.typeutils.AbstractRowDataSerializer`

---

This occurs during the opening of the Sort Operator class.
- [SortOperator](https://github.com/apache/hudi/blob/25c2f5949384def4b22d1a77f3fe76cc0d493f06/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/SortOperator.java#L72)

Flink uses Kryo as the default Serializer.

**How can I use Sort Operator to perform BULK_INSERT?**

Below is my code.

```kotlin
@JvmStatic
fun main(args: Array) {
val env = StreamExecutionEnvironment.getExecutionEnvironment()
env.enableCheckpointing(5000)

// kafka source with json format data
val kafkaSource = KafkaSource.builder()
.setBootstrapServers(bootstrapServers)
.setTopics(topic)
.setGroupId(StatusV2CowApp::class.java.name)
.setClientIdPrefix(UUID.randomUUID().toString())
.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
.setDeserializer(CustomKafkaRecordDeserializationSchema())
.build()

// json data to GenericRowData
val stream: SingleOutputStreamOperator =
env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "hudi_bmt_cow_status_v2_source")
.map { DpObjectMapper.readValue(it.value!!, StatusV2VO::class.java) }
.map { it.toStatusV2WithPartitionVO()}
.map {
val dataSize = StatusV2Partition.values().size
val row = GenericRowData(dataSize)
HudiFieldUtils.rowSetAll(StatusV2Partition::class.java, it, row)
row
}

// sink Hudi Table
HudiFieldUtils.addColumnAllNonDelete(StatusV2Partition::class.java, HoodiePipeline.builder("hudi_cow_bmt_status_v2"))
.pk("key")
.partition("partition_day", "partition_hour")
.options(
mapOf(
FlinkOptions.PATH.key() to "hdfs:///user/geonyeong.kim/hudi_bmt/status_v2/cow",
// bulk_insert
FlinkOptions.OPERATION.key() to WriteOperationType.BULK_INSERT.value(),
FlinkOptions.TABLE_TYPE.key() to HoodieTableType.COPY_ON_WRITE.name,
FlinkOptions.INDEX_GLOBAL_ENABLED.key() to "false",
FlinkOptions.WRITE_BATCH_SIZE.key() to "3072D",
FlinkOptions.WRITE_TASK_MAX_SIZE.key() to "4096D",
FlinkOptions.WRITE_MERGE_MAX_MEMORY.key() to "3072"
)
)
.sink(stream as DataStream, true)

env.execute()
}
```

**Environment Description**

* Hudi version : 0.12.2

* Flink version : 1.15.1