[Feature][Connector-V2][Kafka]Kafka source supports data deserializat…

[521daichen]

…ion failure skipping #4361

[Feature][Connector-V2][Kafka]Kafka source supports data deserialization failure skipping #4361

Purpose of this pull request

Check list

• Code changed are covered with tests, or it does not need tests for reason:
• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/incubator-seatunnel/tree/dev/docs
• If you are contributing the connector code, please check that the following files are updated:
  1. Update change log that in connector document. For more details you can refer to connector-v2
  2. Update plugin-mapping.properties and add new connector information in it
  3. Update the pom file of seatunnel-dist
• Update the release-note.

[TyrantLucifer]

Add e2e test case to verify this pull reqeust. Thanks

[521daichen]

Add e2e test case to verify this pull reqeust. Thanks

The function configuration has been added to the kafka-e2e project, and the local docker verification has been completed.

Thanks.

[TyrantLucifer]

Add e2e test case to verify this pull reqeust. Thanks

The function configuration has been added to the kafka-e2e project, and the local docker verification has been completed.

Thanks.

BTW, SeaTunnel use the spotless plugin to guarantee the style of code, so before you submit pull reqeuest you should execute mvn spotless:apply to format you code

[521daichen]

Add e2e test case to verify this pull reqeust. Thanks

Thanks a lot for the reminder.
I execute locally
mvn spotless: check
mvn spotless:apply returns successfully, but it is not clear why CI did not pass

[INFO] SeaTunnel : E2E : Connector V2 : DataHub ........... SUCCESS [ 0.015 s]
[INFO] SeaTunnel : E2E : Connector V2 : Mongodb ........... SUCCESS [ 0.015 s]
[INFO] SeaTunnel : E2E : Connector V2 : Hbase ............. SUCCESS [ 0.014 s]
[INFO] connector-maxcompute-e2e ........................... SUCCESS [ 0.014 s]
[INFO] SeaTunnel : E2E : Engine : ......................... SUCCESS [ 0.013 s]
[INFO] SeaTunnel : E2E : Engine : Base .................... SUCCESS [ 0.127 s]
[INFO] SeaTunnel : E2E : Engine : Console ................. SUCCESS [ 0.016 s]
[INFO] SeaTunnel : E2E : Transforms V2 .................... SUCCESS [ 0.015 s]
[INFO] SeaTunnel : E2E : Transforms V2 : Common ........... SUCCESS [ 0.015 s]
[INFO] SeaTunnel : E2E : Transforms V2 : Part 1 ........... SUCCESS [ 0.015 s]
[INFO] SeaTunnel : E2E : Transforms V2 : Part 2 ........... SUCCESS [ 0.016 s]
[INFO] SeaTunnel : Dist ................................... SUCCESS [ 0.013 s]
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 8.417 s
[INFO] Finished at: 2023-03-17T22:41:47+08:00
[INFO] ------------------------------------------------------------------------
➜ incubator-seatunnel git:(521daichen/kafka-source-support-error-handle) mvn spotless:apply

[TyrantLucifer]

Add an e2e case to do the following things:

1. Send different format messages to kafka and setting way to fail
2. Send different format messages to kafka and setting way to skip

You can refer to this case KafkaIT.

[laglangyue]

maybe we need to collect the dirty data about this by engine，it's very important featur about ETL sofeware， Actually, this pr can slove the serialization of kafka data，can we get the more detail reason about serialization failure，sometimes，we need such feature as null as empty，null,'default data' to solve the dirty illegal column.

[521daichen]

ok. I have added e2e

1、format_error_handle_way = fail
The data is invalid, an exception will be thrown
2、skip
The data is invalid and will be skipped

[TyrantLucifer]

LGTM, let's waiting CI/CD

[TyrantLucifer]

Need fix e2e test cases

[521daichen]

fixed

[EricJoy2048]

LGTM