[Bugfix][Core] Fix Handover using linked blocking queue cause the oom

[harveyyue]

Fix following OOM with spark engine.

22/11/18 10:04:51 INFO AsyncEventQueue: Process of event SparkListenerExecutorMetricsUpdate(driver,WrappedArray((0,0,0,Vector(AccumulableInfo(7,None,Some(35908),None,false,true,None), AccumulableInfo(24,None,Some(2000),None,false,true,None), AccumulableInfo(1,Some(number of output rows),Some(2000),None,true,true,Some(sql))))),Map((0,0) -> org.apache.spark.executor.ExecutorMetrics@20f22a4a)) by listener AppStatusListener took 1.304755574s.

Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "RemoteBlock-temp-file-clean-thread"

Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "mysql-cj-abandoned-connection-cleanup"

Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "parallel-split-enumerator-executor-0-1"

Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "netty-rpc-env-timeout"
^C^C^COpenJDK 64-Bit Server VM warning: Exception java.lang.OutOfMemoryError occurred dispatching signal SIGINT to handler- the VM may need to be forcibly terminated

Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "dispatcher-event-loop-0"

Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "driver-heartbeater"

Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "SparkUI-38"

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

[TyrantLucifer]

@ashulin PTAL.

[hailin0]

Why not use ArrayBlockingQueue?

[ashulin]

IMO, Add a default capacity to LinkedBlockingQueue

[hailin0]

IMO, Add a default capacity to LinkedBlockingQueue

agree with you

@harveyyue please use new LinkedBlockingQueue<>(capacity) or new ArrayBlockingQueue<>(capacity)

revert

[CalvinKirs]

hi, how is the progress?

[dijiekstra]

LGTM

[wuchunfu]

LGTM