[SPARK-53631][CORE] Optimize memory and perf on SHS bootstrap

[pan3793]

What changes were proposed in this pull request?

Core ideas:

1. Change the log replay thread pool to have a bounded queue, and block task submission when the queue is full.

   Currently, the log replay thread pool uses an unbounded queue, when there are a large number (e.g., millions) of event logs under spark.history.fs.logDirectory, all tasks will be queued at the thread pool queue without blocking the scanning thread, and in the next schedule, enqueue again ...

   https://stackoverflow.com/questions/4521983/java-executorservice-that-blocks-on-submission-after-a-certain-queue-size

2. Move log compaction to a dedicated thread pool.

   Replaying and compaction are different types of workloads, isolating them from each other could improve the resource utilization.

Why are the changes needed?

Improve performance and reduce memory usage on the SHS bootstrap with empty KV cache, when there are tons of event logs.

Does this PR introduce any user-facing change?

No functionality changes, but brings a new config spark.history.fs.numCompactThreads

How was this patch tested?

Tested on an internal cluster, starting SHS with an empty spark.history.store.path and ~650k event logs under spark.history.fs.logDirectory, the related configs are

spark.history.fs.cleaner.maxNum 650000
spark.history.fs.logDirectory hdfs://foo/spark2-history
spark.history.fs.update.interval 5s
spark.history.provider org.apache.spark.deploy.history.FsHistoryProvider
spark.history.store.maxDiskUsage 100GB
spark.history.store.path /foo/bar/historyStore
spark.history.fs.numReplayThreads 64
spark.history.fs.numCompactThreads 4
spark.history.store.hybridStore.enabled true
spark.history.store.hybridStore.maxMemoryUsage 16g
spark.history.store.hybridStore.diskBackend ROCKSDB

• spark.history.store.path is configured to an HDD path
• we disable spark.eventLog.rolling.enabled so numCompactThreads has no heavy work

It's much faster than before, and metrics show better CPU utilization and lower memory usage.

(before vs. after, the 3rd figure is "memory used")

Was this patch authored or co-authored using generative AI tooling?

No.

[pan3793]

IDEA can not highlight without this change.

[pan3793]

I escalated this log level to info, it's not noisy and is relatively important for the user to understand the scanning frequency

[pan3793]

It's actually a MUST change. Otherwise, submitting a task to the self pool will cause a deadlock when the queue is full.

[pan3793]

1024 - queue size, since listing logDir may take some time, it's better to allow some tasks to be queued before starting the next scanning round
60s - core thread idle time

I don't think it's necessary to expose these configs to users, but I'm open to different ideas.

[pan3793]

the ${updated.size} is useful, but the ${updated.map(_.rootPath)} is dangerous, it produces a huge string that may cause OOM.

[pan3793]

cc @dongjoon-hyun @LuciferYang @mridulm @wangyum

[pan3793]

this class is marked as @Private, but the API has been stable since 2016, so it should be low risk to use directly.

[dongjoon-hyun]

Well, I'm not sure this is a good decision.

[peter-toth]

The idea and implementation looks good to me, but I share the concerns of @dongjoon-hyun .

[pan3793]

@dongjoon-hyun @peter-toth thanks for the review, I understand your concerns, let me explain my thoughts.

my initial idea is to implement such a policy

val BLOCK_PLOCY = new RejectedExecutionHandler() {
  override def rejectedExecution(r: Runnable, executor: ThreadPoolExecutor): Unit = {
    executor.getQueue.put(r)
  }
}

but after searching on Stack Overflow, this seems to be trickier than I thought

https://stackoverflow.com/questions/3446011/threadpoolexecutor-block-when-its-queue-is-full/3518588#3518588

https://stackoverflow.com/questions/4521983/java-executorservice-that-blocks-on-submission-after-a-certain-queue-size

Given that Hadoop already has a solid (I suppose so) and stable implementation, I eventually chose to use it directly. In the worst cases, for example, Hadoop breaks the API, or Spark decides to cut off the Hadoop deps in the future (both should be rare to happen), we can fork the code from Hadoop.

[dongjoon-hyun]

I understand the reasoning. However, I disagree with the point of cutting off the Hadoop deps. If that's tricky now, it will be the same in the end.

Without moving forward, we need to check the following.

1. Can we borrow Private-scope code in Spark side stably without increasing further maintenance burden?
2. Can we ask Apache Hadoop community to expose that class officially? This will remove our forked code at that time.

Historically, we spent lots of efforts to migrate to Hadoop Client to avoid this kind of dependencies issue. I don't think both the Apache Spark and Apache Hadoop community wants to have a kind of hard dependencies again.

[pan3793]

@dongjoon-hyun @peter-toth The Hadoop code comes from Apache S4, I wrote a simple "BlockingThreadPoolExecutorService" based on it, so we don't have to worry about using Hadoop private API.

[pan3793]

BTW, I also verified the updated code on the prod env, and it has the same effect as the first version.

[dongjoon-hyun]

We need to avoid introducing the usage of private-scoped method.

[dongjoon-hyun]

+1, LGTM. Thank you, @pan3793 and @peter-toth .

Merged to master for Apache Spark 4.1.0-preview2 (next week).