有大佬遇到过这个问题吗？用flink cdc采集数据并行度开高一点就报错了

org.apache.flink.util.FlinkException: Global failure triggered by OperatorCoordinator for 'Source: TableSourceScan(table=[[default_catalog, default_database, orderline_source_4]], fields=[id, order_id, extra_info, promotion_info, item_id, item_name, item_type, barcode, sku_id, sku_name, goods_id, category_id, img_url, quantity, unit_type, unit_name, unit_price, discount_price, total_price, create_time, update_time, deleted, tag, biz_line_id]) -> Calc(select=[CAST(id) AS id, CAST(order_id) AS order_id, CAST(extra_info) AS extra_info, promotion_info, CAST(item_id) AS item_id, CAST(item_name) AS item_name, CAST(item_type) AS item_type, CAST(barcode) AS barcode, CAST(sku_id) AS sku_id, CAST(sku_name) AS sku_name, CAST(goods_id) AS goods_id, CAST(category_id) AS category_id, CAST(img_url) AS img_url, CAST(quantity) AS quantity, unit_type, CAST(unit_name) AS unit_name, CAST(unit_price) AS unit_price, CAST(discount_price) AS discount_price, CAST(total_price) AS total_price, create_time, CAST(update_time) AS update_time, CAST(deleted) AS deleted, tag, biz_line_id, CAST(DATE_FORMAT(create_time, _UTF-16LE'yyyy-MM-dd')) AS dt]) -> NotNullEnforcer(fields=[id, order_id, extra_info, item_id, item_name, item_type, barcode, sku_id, sku_name, goods_id, category_id, img_url, quantity, unit_name, unit_price, discount_price, total_price, update_time, deleted, dt])' (operator cbc357ccb763df2852fee8c4fc7d55f2). at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.failJob(OperatorCoordinatorHolder.java:545) at org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator$QuiesceableContext.failJob(RecreateOnResetOperatorCoordinator.java:231) at org.apache.flink.runtime.source.coordinator.SourceCoordinatorProvider$CoordinatorExecutorThreadFactory.lambda$new$0(SourceCoordinatorProvider.java:101) at org.apache.flink.runtime.source.coordinator.SourceCoordinatorProvider$CoordinatorExecutorThreadFactory.uncaughtException(SourceCoordinatorProvider.java:124) at org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext.lambda$runInCoordinatorThread$6(SourceCoordinatorContext.java:255) at org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:42) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:750) Caused by: java.lang.OutOfMemoryError: Java heap space at java.util.Arrays.copyOf(Arrays.java:3236) at java.io.ByteArrayOutputStream.toByteArray(ByteArrayOutputStream.java:191) at org.apache.flink.runtime.source.coordinator.SourceCoordinator.writeCheckpointBytes(SourceCoordinator.java:389) at org.apache.flink.runtime.source.coordinator.SourceCoordinator.toBytes(SourceCoordinator.java:370) at org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$checkpointCoordinator$5(SourceCoordinator.java:244) at org.apache.flink.runtime.source.coordinator.SourceCoordinator$$Lambda$1247/2109093479.run(Unknown Source) at org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$runInEventLoop$8(SourceCoordinator.java:329) at org.apache.flink.runtime.source.coordinator.SourceCoordinator$$Lambda$992/1412453376.run(Unknown Source) at org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:40)