[Bugfix] Fix fastjson classloader leaked

[banmoy]

What type of PR is this：

• BugFix
• Feature
• Enhancement
• Refactor
• UT
• Doc
• Tool

Which issues of this PR fixes ：

Fixes #

Problem Summary(Required) ：

Problem

If a job fails many times, flink can warn there is a classloader leak when fastjson serializes/deserializes data
Caused by: java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration 'classloader.check-leaked-classloader'.

Reason

Fastjson uses two global/static configs by default to serialize and deserialize data (SerializeConfig#globalInstance and ParserConfig#global). Flink's classloader mechanism will treat them as classloader leak if a JVM instance run a job in multiple times, such as failover or in session mode. See flink classloader.check-leaked-classloader for details.

Solution

Don't use fastjson's global configs. Instead each sink instance creates and uses it's own configs, and do not shared with other instance

Checklist:

• I have added test cases for my bug fix or my new feature
• This pr will affect users' behaviors
• This pr needs user documentation (for new or modified features or behaviors)
• I have added documentation for my new feature or new function