lnbest0707 opened a new pull request, #1153: URL: https://github.com/apache/flink-kubernetes-operator/pull/1153
## 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`](https://issues.apache.org/jira/browse/FLINK-39981) using the `DynamicKafkaSource.activeSplitCount` gauge added by [`FLINK-39990`](https://issues.apache.org/jira/browse/FLINK-39990). The metric is optional: disabled, missing, incomplete, malformed, or all-zero data keeps the existing behavior. ## 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`. - Immediately caps source parallelism when active splits are below current parallelism (`N < P`), bypassing the ordinary scale-down delay. - 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. - 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
