[WIP-Improve][CDC] Upgrade debezium to 1.9.8.Final

[CheneyYin]

Purpose of this pull request

Upgrade debezium to 1.9.8.Final

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

• 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/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.

[CheneyYin]

// debezium 1.6.4.Final

    private ChangeEventQueue(Duration pollInterval, int maxQueueSize, int maxBatchSize, Supplier<LoggingContext.PreviousContext> loggingContextSupplier,
                             long maxQueueSizeInBytes, boolean buffering) {
        this.pollInterval = pollInterval;
        this.maxBatchSize = maxBatchSize;
        this.maxQueueSize = maxQueueSize;
        this.queue = new LinkedBlockingDeque<>(maxQueueSize);
        this.metronome = Metronome.sleeper(pollInterval, Clock.SYSTEM);
        this.loggingContextSupplier = loggingContextSupplier;
        this.maxQueueSizeInBytes = maxQueueSizeInBytes;
        this.buffering = buffering;
    }

https://github.com/debezium/debezium/blob/c6c40201d2a3cf70832525d77f87fc3270af145f/debezium-core/src/main/java/io/debezium/connector/base/ChangeEventQueue.java#L87-L98

// debezium 1.9.8.Final

    private ChangeEventQueue(Duration pollInterval, int maxQueueSize, int maxBatchSize, Supplier<LoggingContext.PreviousContext> loggingContextSupplier,
                             long maxQueueSizeInBytes, boolean buffering) {
        this.pollInterval = pollInterval;
        this.maxBatchSize = maxBatchSize;
        this.maxQueueSize = maxQueueSize;
        this.queue = new ArrayDeque<>(maxQueueSize);
        this.loggingContextSupplier = loggingContextSupplier;
        this.sizeInBytesQueue = new ArrayDeque<>(maxQueueSize);
        this.maxQueueSizeInBytes = maxQueueSizeInBytes;
        this.buffering = buffering;
    }

https://github.com/debezium/debezium/blob/393291bba54edc745f45a679ffd7b2aa7110f454/debezium-core/src/main/java/io/debezium/connector/base/ChangeEventQueue.java#L84-L94

Some JdbcSourceFetchTaskContext class (like MySqlSourceFetchTaskContext) set size of ChangeEventQueue on Integer.MAX_VALUE. It will lead OOM because ArrayQueue preallocated memory after upgrade debezium to 1.9.8.Final.

seatunnel/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/MySqlSourceFetchTaskContext.java

Lines 125 to 145 in 7b48a16

	// If in the snapshot read phase and enable exactly-once, the queue needs to be set to a
	// maximum size of `Integer.MAX_VALUE` (buffered a current snapshot all data). otherwise,
	// use the configuration queue size.
	final int queueSize =
	sourceSplitBase.isSnapshotSplit() && isExactlyOnce()
	? Integer.MAX_VALUE
	: getSourceConfig().getDbzConnectorConfig().getMaxQueueSize();
	this.queue =
	new ChangeEventQueue.Builder<DataChangeEvent>()
	.pollInterval(connectorConfig.getPollInterval())
	.maxBatchSize(connectorConfig.getMaxBatchSize())
	.maxQueueSize(queueSize)
	.maxQueueSizeInBytes(connectorConfig.getMaxQueueSizeInBytes())
	.loggingContextSupplier(
	() ->
	taskContext.configureLoggingContext(
	"mysql-cdc-connector-task"))
	// do not buffer any element, we use signal event
	// .buffering()
	.build();

[CheneyYin]

debezium/debezium@858557d