Standardize executor limit enforcement for dynamic and static allocation

[MehulBatra]

There's a bug in how it enforces the maxExecutor boundary.
In the dynamic allocation section, it only checks:
restartedExecutors

if (restartedExecutors.size >= maxExecutor) {
  return
}

But this only counts the restartedExecutors, not all executors. Meanwhile, properly we should check the total:
This correctly checks total executors

if((appInfo.executors.size + restartedExecutors.size) >= executorInstances) {
  return
}

The fix would be to modify the dynamic allocation condition to match the static allocation approach:

if (dynamicAllocationEnabled) {
  val maxExecutor = conf.getInt("spark.dynamicAllocation.maxExecutors", 0)
  if ((appInfo.executors.size + restartedExecutors.size) >= maxExecutor) {
    return
  }
}

This bug explains exactly why we are getting maxExecutor set to 3 but 4 pods/machines available, it keeps creating executors until it hits the physical limit rather than respecting your configured maximum.

Static Allocation:
This handles executor scaling when Spark's dynamic allocation is disabled:

      else {
        val executorInstances = conf.getInt("spark.executor.instances", 0)
        if (executorInstances != 0) {
          if((appInfo.executors.size + restartedExecutors.size) >=  executorInstances) {
            return
          }
        }
      }

[pang-wu]

@MehulBatra the build failed due to lint, do you mind fix that?

[MehulBatra]

@MehulBatra the build failed due to lint, do you mind fix that?

Linting is fixed, This is a port conflict issue. Despite setting include_dashboard=False in the ray.init() call, Ray is still trying to allocate ports for dashboard-related components, and it's assigning the same port (52365) to both dashboard_agent_grpc and dashboard_agent_http, could you help me with this @pang-wu

[pang-wu]

@carsonwang can you help here -- I don't have access to the github action jobs.

[carsonwang]

Restarted the test.

[pang-wu]

lgtm

[carsonwang]

Thank you all!