[FLINK-40445] Autoscaler: integration test for Kafka/Pulsar source partition-metric detection by spuru9 · Pull Request #1189 · apache/flink-kubernetes-operator

spuru9 · 2026-08-20T16:07:31Z

What is the purpose of the change

This pull request adds an integration test that verifies ScalingMetricCollector correctly derives numSourcePartitions for Kafka and Pulsar sources against real brokers. It is a follow-up to FLINK-40443 / #1187, added per maintainer consensus on that PR as a separate, non-blocking PR. It does not touch the hardening fix itself.

Brief change log

Adds KafkaPulsarPartitionMetricsITCase, covering: multi-topic summing, dotted and hyphenated topic names, source parallelism > 1, multiple source vertices in one job, and a non-Kafka/Pulsar source that must not report a partition count
Adds MiniClusterJobDriver, which runs the Flink job under test via a MiniCluster in its own JVM rather than in-process, because flink-autoscaler vendors a compatibility copy of org.apache.flink.runtime.rest.messages.job.JobDetailsInfo under src/main/java that shadows the real Flink class on this module's own classpath, which breaks a same-process JobManager's own REST handlers
Adds test-scoped dependencies to flink-autoscaler/pom.xml (flink-connector-kafka, flink-connector-pulsar, testcontainers-kafka, testcontainers-pulsar, commons-codec) and disables the failsafe module path (useModulePath=false), since the connector jars otherwise get placed on the module path as automatic modules and lose visibility into flink-runtime's unnamed-module classpath classes
Wires flink-autoscaler into CI (ci.yml) with the same per-module verify -Dit.skip=false step already used for flink-kubernetes-operator, flink-kubernetes-webhook, and flink-autoscaler-plugin-jdbc, so the new ITCase actually runs instead of being silently skipped by the whole-reactor build

Verifying this change

This change added tests and can be verified as follows:

Added KafkaPulsarPartitionMetricsITCase, an integration test that spins up real Kafka and Pulsar brokers via Testcontainers and asserts ScalingMetricCollector#getJobTopology reports the correct numSourcePartitions for: multi-topic summing, dotted/hyphenated topic names, source parallelism > 1, multiple source vertices in one job, and a negative case (non-Kafka/Pulsar source reports no partition count)
mvn -pl flink-autoscaler verify -Dit.skip=false passes locally (300 unit tests + the new IT)
mvn -pl flink-autoscaler checkstyle:check spotless:check is clean
CI run on this PR is green on both the JDK 17 and JDK 21 matrix legs, including the new Tests in flink-autoscaler step: https://github.com/apache/flink-kubernetes-operator/actions/runs/32391143762

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

Dependencies (does it add or upgrade a dependency): yes, but test-scope only, confined to flink-autoscaler/pom.xml (flink-connector-kafka, flink-connector-pulsar, testcontainers-kafka, testcontainers-pulsar, commons-codec)
The public API, i.e., is any changes to the CustomResourceDescriptors: no
Core observer or reconciler logic that is regularly executed: no

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? not applicable

Was generative AI tooling used to co-author this PR?

Yes (Claude Code)

spuru9 · 2026-08-27T09:01:27Z

@rmetzger Can you check this. Follow up to an ask on earlier PR.

spuru9 · 2026-09-01T06:25:20Z

@Dennis-Mircea Can you help me a review.

Copilot

Copilot review overview

🔵 Needs a closer look

The broker-backed, separate-JVM integration setup warrants final human review.

Review effort: Balanced
Findings: None

What changed in this PR

Adds broker-backed integration coverage for Kafka/Pulsar source partition-metric detection using Testcontainers and an isolated Flink MiniCluster JVM.

Changes:

Adds Kafka, Pulsar, and non-broker source integration scenarios.
Adds a separate-JVM MiniCluster test driver.
Adds test dependencies, Failsafe configuration, and CI execution.

File	Description
`flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MiniClusterJobDriver.java`	Runs test jobs in an isolated MiniCluster JVM.
`flink-autoscaler/src/test/java/org/apache/flink/autoscaler/KafkaPulsarPartitionMetricsITCase.java`	Verifies partition detection against real Kafka and Pulsar brokers.
`flink-autoscaler/pom.xml`	Adds test dependencies and Failsafe configuration.
`.github/workflows/ci.yml`	Runs autoscaler integration tests in CI.

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

rmetzger · 2026-09-24T14:12:31Z

+            Files.deleteIfExists(driverResultFile);
+        }
+        if (driverLogFile != null) {
+            Files.deleteIfExists(driverLogFile);


this will delete the log also in case of a failure -- keep it so that debugging is easier.

ideally when there's an error, print the entire log contents to the output?

Done in 338e8ce — the driver log is now kept when setup or the test fails, and its full contents are logged to the test output (CI doesn't upload artifacts, so the console is where it'll be seen). It's only deleted after a successful run.

rmetzger · 2026-09-24T14:14:27Z

+    /**
+     * Polls {@code resultFile} for the two lines {@link MiniClusterJobDriver} writes once ready.
+     */
+    private static void awaitDriverReady(Process process, Path resultFile) throws Exception {


use org.awaitility.Awaitility here as well?

Done in 338e8ce — switched awaitDriverReady to Awaitility; an early driver exit still fails immediately.

rmetzger · 2026-09-24T14:38:17Z

I agree with the approach and direction in this PR, happy to merge it once above items are addressed.

Copilot

Copilot review overview

🟢 Approval recommended

Changes are limited to test infrastructure and CI, no blocking issues remain, and the linked CI run succeeded.

Review effort: Balanced
Findings: None

…ion detection Follow-up to FLINK-40443/PR apache#1187 per maintainer consensus: verifies ScalingMetricCollector derives the correct numSourcePartitions against real Kafka and Pulsar brokers (multi-topic summing, dotted/hyphenated topic names, parallelism > 1, multiple source vertices, and a non-Kafka/Pulsar negative case), without touching the hardening fix itself. Generated-by: Claude Code

Adds the same per-module "verify -Dit.skip=false" step already used for flink-kubernetes-operator, flink-kubernetes-webhook, and flink-autoscaler-plugin-jdbc, so the new Kafka/Pulsar autoscaler ITCase actually runs instead of being silently skipped by the whole-reactor build. Generated-by: Claude Code

…y for driver readiness Generated-by: Claude Code (Claude Opus 5.5)

spuru9 changed the title ~~Add integration test for autoscaler Kafka/Pulsar partition detection~~ [FLINK-40445] Autoscaler: integration test for Kafka/Pulsar source partition-metric detection Aug 20, 2026

spuru9 force-pushed the autoscaler-kafka-pulsar-partition-it branch from 40e4441 to 94971d1 Compare August 20, 2026 16:17

spuru9 marked this pull request as ready for review August 27, 2026 08:57

rmetzger requested a balanced review from Copilot September 24, 2026 13:50

Copilot started reviewing on behalf of rmetzger September 24, 2026 13:57

View session

Copilot AI reviewed Sep 24, 2026

View reviewed changes

rmetzger reviewed Sep 24, 2026

View reviewed changes

rmetzger requested a balanced review from Copilot October 6, 2026 06:12

Copilot started reviewing on behalf of rmetzger October 6, 2026 06:13

View session

Copilot AI reviewed Oct 6, 2026

View reviewed changes

spuru9 requested a review from rmetzger October 30, 2026 19:16

spuru9 added 3 commits October 31, 2026 00:47

[FLINK-40445] Keep and print the driver log on failure, use Awaitilit…

7df0b66

…y for driver readiness Generated-by: Claude Code (Claude Opus 5.5)

spuru9 force-pushed the autoscaler-kafka-pulsar-partition-it branch from 338e8ce to 7df0b66 Compare October 30, 2026 19:49

Navigation Menu

Repository navigation

[FLINK-40445] Autoscaler: integration test for Kafka/Pulsar source partition-metric detection - #1189

[FLINK-40445] Autoscaler: integration test for Kafka/Pulsar source partition-metric detection#1189
spuru9 wants to merge 3 commits into
apache:mainfrom
spuru9:autoscaler-kafka-pulsar-partition-it

spuru9 commented Aug 20, 2026 •

edited

Loading

Uh oh!

spuru9 commented Aug 27, 2026

Uh oh!

spuru9 commented Sep 1, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

rmetzger Sep 24, 2026

Uh oh!

rmetzger Sep 24, 2026

Uh oh!

spuru9 Oct 30, 2026

Uh oh!

rmetzger Sep 24, 2026

Uh oh!

spuru9 Oct 30, 2026

Uh oh!

rmetzger commented Sep 24, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

spuru9 commented Aug 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

spuru9 commented Aug 27, 2026

Uh oh!

spuru9 commented Sep 1, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Copilot review overview

🔵 Needs a closer look

Uh oh!

rmetzger Sep 24, 2026

Choose a reason for hiding this comment

Uh oh!

rmetzger Sep 24, 2026

Choose a reason for hiding this comment

Uh oh!

spuru9 Oct 30, 2026

Choose a reason for hiding this comment

Uh oh!

rmetzger Sep 24, 2026

Choose a reason for hiding this comment

Uh oh!

spuru9 Oct 30, 2026

Choose a reason for hiding this comment

Uh oh!

rmetzger commented Sep 24, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Copilot review overview

🟢 Approval recommended

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

spuru9 commented Aug 20, 2026 •

edited

Loading