Emit estimated tasks per datastream as a metric and log

[harshOSS]

Summary

Add a per-datastream estimatedNumTasks gauge, plus an INFO log of the same value, reporting the estimated number of tasks calculated for a datastream.

• We already emit the numTasksCappedByMaxTasks meter and a WARN log, but only when the estimate exceeds the datastream's maxTasks. There is no visibility into the estimated task count for streams that are not being capped.
• This change logs and emits the estimated task count right before that maxTasks comparison, so the estimate is always visible and can be compared against maxTasks and against the actual numTasks.
• Purely an observability change: no assignment behavior is modified.

The emission lives in StickyPartitionAssignmentStrategy.validateNumTasksAgainstMaxTasks(), the choke point shared by StickyPartitionAssignmentStrategy (partition-count based estimate) and LoadBasedPartitionAssignmentStrategy (throughput/load based estimate), so both strategies are covered. The reported value is the estimate before it is capped at maxTasks.

• Metric: StickyPartitionAssignmentStrategy.<datastreamTaskPrefix>.estimatedNumTasks
• Log: Estimated number of tasks calculated for datastream <taskPrefix>: <estimate>

The gauge is declared in getMetricInfos(), registered alongside the existing per-datastream gauges (numTasks, actualPartitionsPerTask, elasticTaskParametersNeedAdjustment), and unregistered with them.

Known limitations (pre-existing behavior, intentionally not changed here)

• The estimate is recorded whenever the task count is calculated, i.e. during the partition assignment that follows a datastream becoming READY. If that assignment aborts to create more tasks, the value is logged and stored but the gauge is registered on the next successful partition assignment.
• LoadBasedPartitionAssignmentStrategy.unregisterMetrics does not chain to the base class, so per-datastream gauges of deleted streams can retain their last value there. This affects the existing gauges in the same way and is out of scope for this change.

Testing Done

• Rubber-duck review of the change — no correctness or safety issues found
• Unit tests updated:
  • TestStickyPartitionAssignmentStrategy.testElasticTaskPartitionAssignmentCreatesAtMostMaxTasks asserts the new gauge reports the uncapped estimate (13) while numTasks reports the capped actual (5)
  • TestLoadBasedPartitionAssignmentStrategy.testValidThroughputBasedPartitionAssignmentPath asserts the gauge is emitted on the load-based path as well
• TestStickyPartitionAssignmentStrategy (19 tests) and TestLoadBasedPartitionAssignmentStrategy (7 tests) pass under JDK 17
• TestCoordinator.testCoordinationWithPartitionAssignment and TestCoordinator.testCoordinationWithElasticTaskAssignmentPartitionAssignment pass
• checkstyleMain and checkstyleTest pass

🤖 Generated with GitHub Copilot CLI

[mittalprince]

LGTM

[dhananjay-sawner]

Just curious why do we need this metric?

[dhananjay-sawner]

LGTM