Add configurable periodic refresh of throughput-violating topics cache

[harshOSS]

Summary

The coordinator's host-level throughput-violating topics cache was only rebuilt on assignment and datastream-update events, so between rebalances it could remain stale for a long time. This change adds a periodic rebuild from the current assignment so cache freshness is bounded by a configurable interval regardless of rebalance timing, and makes the behavior fully config-driven:

• brooklin.server.coordinator.throughputViolatingTopicsRefreshPeriodMs (default 5 minutes) — the refresh interval.
• brooklin.server.coordinator.enableThroughputViolatingTopicsPeriodicRefresh (default true) — a dedicated on/off toggle, additionally gated by the existing enableThroughputViolatingTopicsHandling feature flag.

Additional hardening in the same code path:

• onAssignmentChange now rebuilds the cache before queuing the async task-start event (matching onDatastreamUpdate), so a newly started producer cannot emit before its topic is present in the map.
• The shared rebuild path logs a WARN when an assigned task cannot be resolved and is dropped, or when a non-empty assignment yields an empty/partial rebuild, so silent partial rebuilds become detectable.

Testing Done

• Local code review completed
• Unit tests added/updated
• Integration tests pass
• Manual testing performed

Added and ran (JDK 17, Gradle 7.6.4):

• TestCoordinatorConfig#testThroughputViolatingTopicsRefreshPeriodConfig, #testEnableThroughputViolatingTopicsPeriodicRefreshConfig
• TestCoordinator#testThroughputViolatingTopicsPeriodicRefreshRebuildsStaleMap, #testThroughputViolatingTopicsPeriodicRefreshEnablementGating
• Existing throughput-violating TestCoordinator tests still pass; checkstyleMain/checkstyleTest are clean for datastream-server and datastream-server-restli.

🤖 Generated with GitHub Copilot CLI

[kanishkjaiswal2015]

will this be updated with the latest assignment or older assignment?

Can we just decouple the violating topics map from the assignments itself? Basically all throughout violating topics updated in the cache irrespective of whether or not they are assigned to the node?

[harshOSS]

will this be updated with the latest assignment or older assignment?

• this will be always the latest assignment present in Zookeeper .

Can we just decouple the violating topics map from the assignments itself? Basically all throughout violating topics updated in the cache irrespective of whether or not they are assigned to the node?

• IMO , we should not do this as on small cluster it will not be an issue but on large cluster it will be unncessary parsing of all assignments across the cluster (assignment can go upto 1k or even more) , so it will result in additional network call without any benefits.