[Bug] seatunnel2.3.8使用restFulApi提交离线任务，服务部署成功运行一段时间后离线任务执行速度会越来越慢

[sunwl1993]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

采用分离集群部署，一个master(16C 32G) 两个worker(64C 64G)，使用restfulapi每分钟提交约80个任务，刚刚部署成功时离线任务约6-10秒执行成功，运行一段时间后就变成了1-2分钟左右
观察日志，在57分就已经打印task done日志但是在58分才打印complete with FINISHED日志，对应TaskExecutionService类内的void taskDone(Task task)方法内的两处日志打印，中间的某些处理似乎拖慢了整个流程

SeaTunnel Version

2.3.8

SeaTunnel Config

env {
"job.mode"=BATCH
"job.name"="SeaTunnel_Job"
"savemode.execute.location"=CLUSTER
}
source {
Jdbc {
    "connection_check_timeout_sec"="30"
    "fetch_size"="0"
    "use_select_count"="false"
    "skip_analyze"="false"
    "where_condition"="WHERE OccurTime >= DATEADD(minute, -10, GETDATE())"
    parallelism="1"
    "split.size"="8096"
    "split.even-distribution.factor.upper-bound"="100.0"
    "split.even-distribution.factor.lower-bound"="0.05"
    "split.sample-sharding.threshold"="1000"
    "split.inverse-sampling.rate"="1000"
    "result_table_name"=Table18562324779168
    query="SELECT [PID], [FlawID], [GlobalX], [OccurTime], [KernelLength], [KernelWidth], [DeformLength], [DeformWidth], [X], [Y], [GlassID], [FlawType], [FlawGrade] FROM [xx].[xx].[xxx]"
    "table_path"="xx.xx.xx"
    user=xx
    url="jdbc:sqlserver://xx:1446;database=xx"
    password=xx
    driver="com.microsoft.sqlserver.jdbc.SQLServerDriver"
}
}
transform {
Sql {
    query="select\n'9503A025' AS productlinecode,PID AS pid,OccurTime AS occurtime,FlawID AS flawid,GlobalX AS globalx,KernelLength AS kernellength,KernelWidth AS kernelwidth,DeformLength AS deformlength,DeformWidth AS deformwidth,X AS x,Y AS y,GlassID AS glassid,FlawType AS flawtype,FlawGrade AS flawgrade,CURRENT_TIMESTAMP AS etl_inserttime\nfrom FlawTable"
    "result_table_name"=Table18562324779169
    "source_table_name"=Table18562324779168
}
}
sink {
Doris {
    tableName="ods_pp_kld_glassflaw"
    "data_save_mode"="APPEND_DATA"
    "schema_save_mode"="CREATE_SCHEMA_WHEN_NOT_EXIST"
    "multi_table_sink_replica"="1"
    "save_mode_create_template"="CREATE TABLE IF NOT EXISTS `${database}`.`${table}` (${rowtype_fields}) ENGINE=OLAP PROPERTIES (\"replication_allocation\" = \"tag.location.default: 1\",\"in_memory\" = \"false\",\"storage_format\" = \"V2\",\"disable_auto_compaction\" = \"false\")"
    "needs_unsupported_type_casting"="false"
    "sink.label-prefix"=a05
    "doris.config" {
        "read_json_by_line"=true
        format=json
    }
    "doris.batch.size"="1024"
    "sink.enable-2pc"="false"
    "sink.enable-delete"="false"
    "source_table_name"=Table18562324779169
    database="xxx"
    table="xxx"
    password="xxx"
    username=xx
    fenodes="xxx:8030"
    query-port="9030"
}
}

Running Command

POST /hazelcast/rest/maps/submit-job
GET /hazelcast/rest/maps/job-info/{jobId}

Error Exception

暂无

Zeta or Flink or Spark Version

No response

Java or Scala Version

1.8

Screenshots

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[sunwl1993]

同时master和worker机器内存随着运行不断升高，master有过一次OOM，搜索Issues按照https://github.com/apache/seatunnel/pull/9696和https://github.com/apache/seatunnel/pull/9776/files修复了pendingJobMasterMap和metricsImap可能存在的内存泄漏，但是重新部署后master、worker内存仍不断升高,且任务逐渐变慢，观察oom时的dump日志发现hazelcast的HeapData对象占比极高

[sunwl1993]

这是master节点seatunnel.yaml的配置信息：

seatunnel:
  engine:
    classloader-cache-mode: true
    history-job-expire-minutes: 5
    backup-count: 0
    queue-type: blockingqueue
    print-execution-info-interval: 60
    print-job-metrics-info-interval: 60
    slot-service:
      dynamic-slot: false
      slot-num: 50
    checkpoint:
      interval: 10000
      timeout: 60000
      storage:
        type: localfile
        max-retained: 3
        plugin-config:
          storage.type: localfile
          namespace: /opt/seatunnel/checkpoint/
    telemetry:
      metric:
        enabled: false

[dybyte]

Hi @sunwl1993 , thank you for the detailed report!
@Hisoka-X I’m interested in investigating this issue and working on a fix. Could you please assign it to me?

[bugouyongxin]

我也遇到了这个问题，使用的版本是2.3.11，内存出现溢出，可能是因为imap 中存在数据未进行清理。
官方的描述:
**在SeaTunnel中，我们使用IMap(一种分布式的Map，可以实现数据跨节点跨进程的写入的读取 有关详细信息，请参阅 Hazelcast Map) 来存储每个任务及其task的状态，以便在任务所在节点宕机后，可以在其他节点上获取到任务之前的状态信息，从而恢复任务实现任务的容错。

默认情况下Imap的信息只是存储在内存中，我们可以设置Imap数据的复本数，具体可参考(4.1 Imap中数据的备份数设置)，如果复本数是2，代表每个数据会同时存储在2个不同的节点中。一旦节点宕机，Imap中的数据会重新在其它节点上自动补充到设置的复本数。但是当所有节点都被停止后，Imap中的数据会丢失。当集群节点再次启动后，所有之前正在运行的任务都会被标记为失败，需要用户手工通过seatunnel.sh -r 指令恢复运行。

为了解决这个问题，我们可以将Imap中的数据持久化到外部存储中，如HDFS、OSS等。这样即使所有节点都被停止，Imap中的数据也不会丢失，当集群节点再次启动后，所有之前正在运行的任务都会被自动恢复。**

但是将imap 数据持久化到本地后发现，调用后台 overview 接口、running-jobs 接口等经常返回{“status”: fail, "message": null}。

[LeonYoah]

这是master节点seatunnel.yaml的配置信息：

seatunnel:
engine:
classloader-cache-mode: true
history-job-expire-minutes: 5
backup-count: 0
queue-type: blockingqueue
print-execution-info-interval: 60
print-job-metrics-info-interval: 60
slot-service:
dynamic-slot: false
slot-num: 50
checkpoint:
interval: 10000
timeout: 60000
storage:
type: localfile
max-retained: 3
plugin-config:
storage.type: localfile
namespace: /opt/seatunnel/checkpoint/
telemetry:
metric:
enabled: false

When you encounter a task card, can you try to print the heap memory usage of the master and worker to see the space of the young generation and old generation as well as the GC frequency?

[Hisoka-X]

Please try upgrade to 2.3.12 when it released.

[bugouyongxin]

Please try upgrade to 2.3.12 when it released.

请问什么时候发布2.3.12版本？

[sunwl1993]

这是master节点seatunnel.yaml的配置信息：
seatunnel:
engine:
classloader-cache-mode: true
history-job-expire-minutes: 5
backup-count: 0
queue-type: blockingqueue
print-execution-info-interval: 60
print-job-metrics-info-interval: 60
slot-service:
dynamic-slot: false
slot-num: 50
checkpoint:
interval: 10000
timeout: 60000
storage:
type: localfile
max-retained: 3
plugin-config:
storage.type: localfile
namespace: /opt/seatunnel/checkpoint/
telemetry:
metric:
enabled: false

When you encounter a task card, can you try to print the heap memory usage of the master and worker to see the space of the young generation and old generation as well as the GC frequency?

在之前任务卡顿时Woker的gc日志截图，极为频繁，master相关截图缺失

不过任务卡顿定位到了一个可能性比较大的原因，咱们在TaskExecutionService.updateMetricsContextInImap、JobMaster.removeMetricsContext时在之前的版本用的Imap.tryLock这种方式、在最新的https://github.com/apache/seatunnel/pull内移除了锁使用了synchronized,由于我们的任务提交频率比较高(1分钟80+)，猜测在并发较高的场景下锁竞争较为消耗性能造成了任务卡顿，我现在把updateMetrics和removeMetrics这俩个方法改成了将相关信息放到LinkedBlockingQueue内，然后开启异步线程消费这个队列进行具体的业务处理，这样任务卡顿问题有所缓解，

代码修改及gc相关截图如下：
master和worker的gc频率仍较高,集中在每分钟前几秒（与离线任务提交时间吻合），master内存分配18G，worker 40G，每分钟执行80个任务且有两个worker，worker eden区会迅速占满young gc非常频繁，master 内存似乎有些对象无法回收老年代占用在缓慢上升

master_gc

worker_gc

[dybyte]

不过任务卡顿定位到了一个可能性比较大的原因，咱们在TaskExecutionService.updateMetricsContextInImap、JobMaster.removeMetricsContext时在之前的版本用的Imap.tryLock这种方式、

The approach you described, using a blocking queue and a separate thread for metrics updates, is interesting.

However, it could potentially introduce thread-safety issues, which may result in inconsistent or incorrect metrics being displayed.

If switching from distributed locks to synchronized still does not address the performance bottleneck, another option might be to revisit the single-key design of metricsImap. By partitioning the metrics into multiple keys and managing them separately, we could reduce contention.

@Hisoka-X What do you think about exploring this direction?

[dybyte]

Also, I think this issue may be related to this PR. I will continue to investigate and try to identify any additional memory leaks.

[Hisoka-X]

By partitioning the metrics into multiple keys and managing them separately, we could reduce contention.

It will created many keys in Imap. I'm not sure it's fine. We should do some test.

[dybyte]

I'll design some tests to compare the performance and memory usage between single-key and multi-key metricsImap.
If you have any specific scenarios you'd like me to cover, please let me know!