[Improve][Zeta] Improve job metrics handling with partitioning support

[dybyte]

Fixes #9817

Purpose of this pull request

This PR introduces a new configuration option JOB_METRICS_PARTITION_COUNT which controls the number of partitions used to store running job metrics in Hazelcast IMap. This allows distributing metrics across multiple keys to reduce contention and improve performance when multiple tasks report metrics concurrently.

If you prefer to use a single key for storing metrics, set JOB_METRICS_PARTITION_COUNT to 1 (this is the default, so no configuration is required).
For better parallelism under high load, you can increase this value.
However, setting it too high may introduce overhead when aggregating metrics across multiple partitions, which can lead to reduced overall performance.

Does this PR introduce any user-facing change?

Yes. Users can tune JOB_METRICS_PARTITION_COUNT to balance write contention and aggregation overhead. Default is 1 for backward compatibility.

How was this patch tested?

• Current tests verify that metricsImap is updated and cleared with a custom partition configuration.

• Performance comparison of updateMetrics:

partitionCount: 1
Average completion time per op: 33.461999 seconds
Distributed metrics performance:
- ops: 100, seconds: 119.766, ops/s: 1
----
partitionCount: 10
Average completion time per op: 30.051322 seconds
Distributed metrics performance:
- ops: 100, seconds: 113.089, ops/s: 1
----
partitionCount: 50
Average completion time per op: 24.181994 seconds
Distributed metrics performance:
- ops: 100, seconds: 88.389, ops/s: 1
----
partitionCount: 1000
Average completion time per op: 20.029176 seconds
Distributed metrics performance:
- ops: 100, seconds: 71.765, ops/s: 1
----
partitionCount: 2000
Average completion time per op: 26.631462 seconds
Distributed metrics performance:
- ops: 100, seconds: 94.996, ops/s: 1
----
Without using partition key(i.e., the previous synchronous approach):
- Average completion time per op: 48.34 seconds
- Total time for 100 ops: 176.83 seconds
- Throughput: 1 ops/sec

Using multiple partitions significantly reduces lock contention, improving the performance of the updateMetrics method. This demonstrates that the multi-key approach performs better than the single-key approach under high concurrency (in a local test).

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 plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[dybyte]

I was planning to document this new configuration option, but I couldn't find the relevant section for master server config options. Could you please point me to where this should be documented? @Hisoka-X

[dybyte]

I am not yet very experienced with performance metrics, so I did not include specific measurements in this PR. However, I can provide benchmarks or metrics if needed—please let me know if you’d like to see them.

[Hisoka-X]

where this should be documented

https://github.com/apache/seatunnel/blob/dev/docs/en/seatunnel-engine/hybrid-cluster-deployment.md#4-configure-the-seatunnel-engine and https://github.com/apache/seatunnel/blob/dev/docs/en/seatunnel-engine/separated-cluster-deployment.md#4-configure-seatunnel-engine

[qiaoborui]

By the way, is there a solution for the current issue of runningJobMetrics growing indefinitely? When synchronizing tasks involving several thousand tables, the WAL persisted in OSS frequently reaches hundreds of GB.
@Hisoka-X

[dybyte]

Hi @qiaoborui , Thanks for raising this important point!

My understanding is that the root cause is removeMetrics not executing properly due to heavy contention on the single key.
As a result, finished job metrics were not cleared in time, which caused runningJobMetrics to grow indefinitely.

With this PR, since metrics are partitioned, lock contention should be significantly reduced, so removeMetrics is expected to work more efficiently.
This may help prevent runningJobMetrics from growing indefinitely, even under heavy load.

I’ll keep monitoring this behavior, and if you have further feedback or see this problem persist, please let me know!

[zhangshenghang]

Can we provide a best practice to guide customers in configuration, such as recommended calculation formulas

[Hisoka-X]

By the way, is there a solution for the current issue of runningJobMetrics growing indefinitely? When synchronizing tasks involving several thousand tables, the WAL persisted in OSS frequently reaches hundreds of GB. @Hisoka-X

@qiaoborui We almost fix it in #9696 and #9776. This PR is only to address potential risks in extreme scenarios.

[dybyte]

Can we provide a best practice to guide customers in configuration, such as recommended calculation formulas

@zhangshenghang
Based on local performance tests, increasing the partition count provides significant benefits when the number of tasks exceeds approximately 20,000.
As a practical guideline, a partition count of around 1,000-2,000 tends to offer the best balance between reducing lock contention and minimizing overhead.
It is recommended to start with this value and then adjust based on your cluster size and workload characteristics.
Please note that these recommendations are based on my test results, and the optimal values may vary depending on your environment.

[zhangshenghang]

Can we provide a best practice to guide customers in configuration, such as recommended calculation formulas

@zhangshenghang Based on local performance tests, increasing the partition count provides significant benefits when the number of tasks exceeds approximately 20,000. As a practical guideline, a partition count of around 2,000 tends to offer the best balance between reducing lock contention and minimizing overhead. It is recommended to start with this value and then adjust based on your cluster size and workload characteristics. Please note that these recommendations are based on my test results, and the optimal values may vary depending on your environment.

@dybyte Thanks, you can include it in the document for reference by others. This makes it more convenient for users to use.

[dybyte]

@zhangshenghang Thanks, I've added it to the documentation for everyone’s reference.
Let me know if there’s anything else to improve!

[qiaoborui]

Can we provide a best practice to guide customers in configuration, such as recommended calculation formulas

@zhangshenghang Based on local performance tests, increasing the partition count provides significant benefits when the number of tasks exceeds approximately 20,000. As a practical guideline, a partition count of around 1,000-2,000 tends to offer the best balance between reducing lock contention and minimizing overhead. It is recommended to start with this value and then adjust based on your cluster size and workload characteristics. Please note that these recommendations are based on my test results, and the optimal values may vary depending on your environment.

Thank you for pointing this out. I've been keeping an eye on this issue recently as well. However, I previously reviewed these two PRs and noticed that the memory leaks addressed were related to RunningJobStateIMap and pendingJobMasterMap. I didn't observe any changes concerning RunningJobMetrics. Nevertheless, I'll give the latest PRs modifying the IMaps a try. Thank you very much for your contributions.

[Hisoka-X]

Why foreach metricsImap twice? Can we use only once?

[dybyte]

I intentionally separated the logic into two iterations.
The first pass filters tasks outside of metricsImap to minimize the time holding its internal resources,
and the second pass performs only the minimal compute operations inside the metricsImap.
Since write operations are frequent, I believe this approach can help reduce contention.
I’d be happy to adjust the implementation if you feel a single iteration would be more appropriate.

[Hisoka-X]

We are a bit limited by imap, maybe we should switch to rocksdb in the future. #9851