Gate leader task assignment behind a dynamic ZooKeeper flag

[manishGoyalCode]

Gate leader task assignment behind a dynamic ZooKeeper flag

Summary

Adds a dynamic ZooKeeper-backed kill-switch for datastream task assignment. The leader Coordinator now consults a /{cluster}/assignmentEnabled znode before performing assignment, allowing operators to pause/resume assignment at runtime without a redeploy or restart.

Motivation

During incidents or maintenance, we sometimes need to temporarily stop the leader from (re)assigning datastream tasks. Previously this required code/config changes and a restart. This change introduces a runtime toggle controlled purely via ZooKeeper.

Behavior

The flag at /{cluster}/assignmentEnabled is interpreted as:

Znode state	Assignment
Absent (default)	Enabled
Value "false" (case-insensitive)	Disabled
Empty / null value	Enabled
Any other value (e.g. "true", "yes")	Enabled

When assignment is disabled, the leader does not block the Coordinator event thread. Instead, handleLeaderDoAssignment skips the current cycle and reschedules a LEADER_DO_ASSIGNMENT event after the configured check period (default 5s). This keeps the Coordinator event loop responsive (heartbeats, partition operations, the leader's own assignment changes, and graceful shutdown) while assignment is paused. Assignment resumes automatically once the flag is re-enabled or the znode is deleted; the rescheduled event is a no-op if the instance is no longer the leader, so leadership loss and shutdown naturally stop the re-check loop.

Changes

Production code

• KeyBuilder.java — New assignmentEnabled(String cluster) helper returning the /{cluster}/assignmentEnabled znode path, with Javadoc describing the flag semantics.
• ZkAdapter.java — New isAssignmentEnabled() method that reads the flag from ZooKeeper. Defaults to enabled; treats only a case-insensitive "false" as disabled. Safely handles the znode being deleted between the exists() check and read (returnNullIfPathNotExists=true).
• Coordinator.java — handleLeaderDoAssignment checks isAssignmentEnabled() at the top; if assignment is disabled it logs, reschedules a LEADER_DO_ASSIGNMENT event via _scheduledExecutor after the configured check period (new scheduleLeaderDoAssignmentWhenReEnabled(...) helper), and returns immediately instead of blocking. The scheduler is guarded so it does not pile up redundant futures.
• CoordinatorConfig.java — New config brooklin.server.coordinator.assignmentEnabledCheckPeriodMs (default 5000 ms) controlling the re-check interval, with getter getAssignmentEnabledCheckPeriodMs().

Tests

• TestZkAdapter.java — testIsAssignmentEnabled() covering: absent znode (enabled), "false"/"FALSE" (disabled), "true" (enabled), unrecognized value (enabled), empty value (enabled), and znode deletion restoring the default.
• TestCoordinatorConfig.java — testAssignmentEnabledCheckPeriodConfig() validating the default (5s) and an override.

Operational usage

• Disable assignment: create/set /{cluster}/assignmentEnabled to false.
• Re-enable assignment: set the value to anything else (e.g. true) or delete the znode.

Testing Done

• Added unit tests for ZkAdapter.isAssignmentEnabled() (all flag states) and the new CoordinatorConfig property (default + override).
• mint build / mint test on datastream-server.

Risk / Compatibility

Low risk and backward compatible — the flag defaults to enabled when the znode is absent, so existing clusters behave exactly as before until an operator explicitly sets the flag.

[kanishkjaiswal2015]

since we are blocking this thread here, this will block the entire CoordinatorEventProcessor thread.
Are there any other important tasks that can be blocked due to this?

[manishGoyalCode]

it is intentional . we want to stop everything. as this will be for small time and we want to stop all operations of the cluster itself.

[kanishkjaiswal2015]

LGTM