Is the computeFactors() computation really heavy enough to require caching?
Especially since we are imposing a max SCALE_FACTOR_DISCRETE_DISTANCE.

How about we simplify computeFactors so that we compute only the required factors?

Example:

List<Integer> computeValidTaskCounts(int partitionCount, int currentTaskCount)
{
   final int currentPartitionsPerTask = partitionCount / currentTaskCount;
   final int minPartitionsPerTask = Math.max(1, currentPartitionsPerTask - 2);
   final int maxPartitionsPerTask = Math.min(partitionCount, currentPartitionsPerTask + 2);

   return IntStream.of(minPartitionPerTask, maxPartitionsPerTask + 1)
                   .map(partitionsPerTask -> (partitionCount / partitionsPerTask) + Math.min(partitionCount % partitionsPerTask, 1))
                   .collect(Collectors.toList();
}

As we discussed on voice chat, we leave current implementation as is, but slightly enhancing the way how it data is saved in cache.

Sorry, I meant that we should not have the start delay, period and collection interval as configs at all. Even with good defaults, unnecessary configs only complicate admin work and require code to handle all possible scenarios.

At most, maybe keep just one config scaleActionPeriod that can be specified as an ISO period (e.g. PT1M) or something (mostly since you would be using this in embedded tests). The other configs don't really add any value. They are legacy configs in lag-based auto-scaler which we might as well avoid adding in the new strategy.

This line can have behavioural side effects in the supervisor, since the taskCount should always reflect the current running task count and not the desired task count. Here we are updating the taskCount without actually changing the number of tasks, or suspending the supervisor.

Instead, we could do the following:

• Add an auto-scaler method isScaleDownOnRolloverOnly(). This will always return false for lag-based and always true for cost-based.
• CostBasedAutoScalerConfig.computeOptimalTaskCount() should return the optimal task count for scale down cases as well.
• SeekableStreamSupervisor can store this desired task count in an atomic boolean and retrieve it upon regular task rollover.

Another (perhaps cleaner) option is to simply invoke AutoScaler.computeOptimalTaskCount() whenever we do rollover and then just go with the optimal task count.

@Fly-Style , as you suggested on chat, we can take up the scale down behaviour in a follow up PR.
In the current PR, we can do scale down same as scale up.

Returning 1 (full idle) here would cause the auto-scaler to think that the tasks are doing nothing and cause a scale-down, when in fact the tasks failed to return the metrics and may be in a bad state. Scaling down might further worsen the problem.

Should we return 0 instead?

Scaling up might be an overkill in that scenario - nothing may happen and instead we will waste a resources.
0.5 looks optimal for me (during the implementation i thought between 0.5 and 1).

We can even return -1 to denote that we do not have metrics available, and just skip scaling rather than make a bad decision.

Nit: Is conversion to array still needed? Can we just return List or Set instead?

It's my personal belief: not a fan of boxed primitives A LOT :)

But we are still boxing the primitives while adding to the List. Might as well avoid the extra conversion.

Maybe just use a set to simplify this computation.

I like the idea of the WeightedCostFunction, but I think we need to make it much more intuitive.

Define requirements (already aligned with the current state of this PR):

1. A function that computes cost.
2. A task count with lower cost is better.
3. cost = lagCost * lagWeight + idlenessCost * idleWeight
4. Lower the task count, higher the predicted lag, higher the lag cost.
5. Higher the task count, higher the predicted idleness, higher the idleness cost.

Simplify computations

1. Use linear scaling only, logarithmic scaling makes the terms difficult to reason about and debug.
   The diminishing returns effect is already enforced by the window (discrete distance). If more terms are needed to account for say, task operational overhead, we will add them in the future.
2. Use only one mode i.e. do not invert scaling, even when lag is abnormally high.
3. Use actual metrics instead of normalized or adaptive bounds.
   If a supervisor once saw a lag of 100M, the adaptive ratio would make a lag of 1M seem very small (normalizedLag = 0.01 i.e. 1%). But in reality, a lag of 1M is bad too and needs to be given appropriate weight.
4. Always perform cost computation even if idleness is in the accepted range (0.2-0.6 in the PR).
   This would help us validate the correctness of the formula against real clusters by verifying that the current task count gives minimal cost.

We may re-introduce some of these enhancements in later patches once we have more data points using this auto-scaler, but it is best to start as simple as possible.

Use intuitive metric e.g. compute time

Connect the result of the cost function to an actual metric to make it more intuitive. The best metric I can think of is compute time or compute cycles, as it may be related to actual monetary cost of running tasks.

For example, what if we could model the cost as follows:

1. lagCost = expected time (in seconds) required to recover current lag
2. idlenessCost = total compute time (in seconds) wasted being idle in a single taskDuration
3. Intuitively, we can see that as task count increases, lagCost would increase and idlenessCost would decrease.

The formula for these costs may be something like:

lagCost
= expected time (in seconds) required to recover current lag
= currentAggregateLag / (proposedTaskCount * avgRateOfProcessing)

where,
currentAggregateLag = sum of current lag (in records) across all partitions
avgRateOfProcessing = average of task moving averages

idlenessCost
= total time (in seconds) wasted being idle in a single taskDuration
= total task run time * predicted idleness ratio

where,
total task run time = (proposedTaskCount * taskDuration)

predicted idleness ratio = (proposedTaskCount / currentTaskCount) - (1 - avgPollToIdleRatio)

e.g. if current poll-to-idle-ratio is 0.7, tasks are idle 70% of the time,
so reducing task count by 70% will make tasks busy all the time (idleness ratio = 0).

Assumptions

• Tasks are already at their peak processing rate and will remain at this rate.
• poll-to-idle ratio scales linearly with task count. We may use some reasonable clamps for min (say 0.05) and max (say 0.95).

Regarding:

Use actual metrics instead of normalized or adaptive bounds.
If a supervisor once saw a lag of 100M, the adaptive ratio would make a lag of 1M seem very small (normalizedLag = 0.01 i.e. 1%). But in reality, a lag of 1M is bad too and needs to be given appropriate weight.

Always perform cost computation even if idleness is in the accepted range (0.2-0.6 in the PR).
This would help us validate the correctness of the formula against real clusters by verifying that the current task count gives minimal cost.

Implemented in the latest commit. Other than that, we need to discuss each other point :)
Thanks in advance for your input, appreciate it A LOT!

On second though, we might want to return -1 from here if we don't have any metrics available.
This would cause the auto-scaler to skip scaling rather than make a bad decision.

We can even return -1 to denote that we do not have metrics available, and just skip scaling rather than make a bad decision.

But we are still boxing the primitives while adding to the List. Might as well avoid the extra conversion.

Based on the feedback from @jtuglu1 , I think we might need to reconsider this.

But rather than add a new config, I wonder if we can't improve this logic a bit.

Options:

1. startDelay = Math.min(taskDuration, 30 mins). So we consider scaling up only after the current tasks have run for a bit.
2. startDelay = 3 * config.getScaleActionPeriodMillis(). This could work too, and seems reasonable for most cases.
3. Add a separate config scaleActionStartDelayMillis (so that we remain aligned with the existing behaviour of lag-based auto-scaler), whose default value is 3 * config.getScaleActionPeriodMillis().

@jtuglu1 , @Fly-Style , which one do you guys prefer the most?

Please rename based on usage. IIUC, this constant represents the "maximum" amount by which we can "decrease" the current value of num partitions per task.

Moving averages over a longer time window (say 15 mins) might be more stable and thus more reliable.
If not available, then fallback to 5 minute, then 1 minute.

It might be weird to have this be fed from a config (even as a fallback mechanism).

In the future, we can consider computing this based on the stats of previously completed tasks of this supervisor.
OR just use the last known processing rate.

Use a set to simplify this logic.

We need to revisit the start delay.

We may not always want to skip scaling even if idleness is in the accepted range.

For example, if current idleness is 0.5 and there is no lag, a cluster admin might prefer to scale down the tasks, so that idleness is more like 0.2 or so. They should be allowed to control this via the idleWeight.

For the initial testing of this auto-scaler, let's remove this guardrail.

This is already emitted as a metric.

What would be more useful is the computed terms lagCost and idleCost.
Getting these out as metrics would enable users to choose better values for lagWeight and idleWeight.