[FLINK-39981] [Autoscaler] Correct dynamic source topology from active split metrics

[lnbest0707]

What is the purpose of the change

Dynamic Kafka sources can remove clusters and splits while stale
KafkaSourceReader...currentOffset metric names remain visible to the autoscaler. The
legacy partition count can therefore stay above the source's actual active split count, leaving
empty source subtasks without a rescale or recovery action.

This change adds an opt-in topology correction path for
FLINK-39981 using the
DynamicKafkaSource.activeSplitCount gauge added by
FLINK-39990. The metric is optional:
disabled, missing, incomplete, or malformed data keeps the existing behavior. An exact all-zero
aggregate is retained only as a non-scaling signal that clears and re-arms a prior recovery latch.

The N >= P recovery branch depends on the connector recovery semantics merged in
apache/flink-connector-kafka#278
for FLINK-39980. That change adopts
SupportsSplitReassignmentOnRecovery: restored active splits are reported back to the enumerator
and requeued for redistribution. Without that connector support, a same-parallelism restart could
restore the same skewed reader ownership, so operators should not rely on this recovery branch.

Brief change log

• Adds job.autoscaler.dynamic-source.topology-correction.enabled, defaulting to false.
• Collects the optional active split gauge only when enabled and derives total active splits and
  minimum active splits from its aggregated value without replacing legacy
  NUM_SOURCE_PARTITIONS; an exact all-zero aggregate is retained only to clear a prior
  recovery latch and never becomes a scaling target.
• After active splits remain below current parallelism (N < P) for the full metrics window,
  caps source parallelism without waiting for the ordinary scale-down delay.
• When the active split gauge is valid, also uses it as the upper bound for later ordinary
  utilization scale-ups, so stale legacy partition metrics cannot scale the source back above
  its active split count.
• Detects a persistent empty-reader assignment hole when N >= P and the minimum active split
  count is zero for the full metrics window, then requests one durable same-parallelism recovery;
  apache/flink-connector-kafka#278
  turns that recovery into active-split redistribution.
• Adds a backward-compatible default realizer hook. The Kubernetes realizer uses an idempotent
  restart nonce for LAST_STATE jobs and reuses an already pending restart nonce.
• Persists the recovery latch in Kubernetes and JDBC state stores, including old-state
  compatibility coverage.
• Regenerates the autoscaler configuration documentation.

Verifying this change

This change added tests and can be verified as follows:

• mvn -q -pl flink-autoscaler -am test
• mvn -q -pl flink-autoscaler-plugin-jdbc -am -Dtest=JdbcAutoScalerStateStoreTest -Dsurefire.failIfNoSpecifiedTests=false test
• mvn -q -pl flink-kubernetes-operator -am -Dtest=KubernetesAutoScalerStateStoreTest,KubernetesScalingRealizerTest -Dsurefire.failIfNoSpecifiedTests=false test
• mvn -q -pl flink-kubernetes-docs -am -DskipTests -Pgenerate-docs package
• mvn -q spotless:check

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changes to the CustomResourceDescriptors: no
• Core observer or reconciler logic that is regularly executed: yes

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? docs / JavaDocs

AI assistance

This contribution was prepared with Codex (GPT-5); the author reviewed, adapted, and tested the
changes.

[bowenli86]

Could we keep activeSplitCount as the upper bound after the first correction too?

For example, P=10 and N=5 gets corrected to 5. But once P=N, this falls through to regular scaling, which still uses stale NUM_SOURCE_PARTITIONS

Under load it can scale above 5 again, then get capped back on the next pass. Maybe clamp later scale ups to N too when the metric is valid

[lnbest0707]

Good catch — addressed in 234725a. When the optional gauge is valid, ordinary utilization scaling now also clamps its upper bound to activeSplitCount, so after P=10/N=5 is corrected to P=5, stale legacy partition metrics cannot scale it back above 5. Missing or invalid gauges still use the legacy path. I also added tests for the steady-state cap and missing-gauge fallback.

[bowenli86]

Could we make N < P persist for a bit before bypassing the scale down delay?

During a DynamicKafkaSource metadata refresh, the enumerator sends MetadataUpdateEvent before the new enumerators start. Readers can briefly report a lower positive count while new splits are still in flight

One autoscaler poll in that window could cause an unnecessary downscale or restart. Maybe use the same stability idea as the N >= P path, or a shorter window

[lnbest0707]

Good point — the previous full-window check only covered the N >= P && min == 0 recovery path. Addressed in 9bc4e47: N < P is now only exposed to the scaler after every retained metrics-window sample reports valid 0 < N < P. A transient or missing sample falls back to legacy behavior.

[bowenli86]

@gyfora can you or anyone you know who has more expertise here can help review?