Refresh assigned task's datastream list when a new dedup datastream is created - #1049
Refresh assigned task's datastream list when a new dedup datastream is created#1049mittalprince wants to merge 1 commit into
Conversation
| return; | ||
| } | ||
|
|
||
| boolean anyDatastreamReady = false; |
There was a problem hiding this comment.
i think anyNewDatastreamReady is a better name as initializing state only exists for new streams
There was a problem hiding this comment.
Good catch, thanks! While digging into a CI failure this surfaced, I actually went a step further: the broadcast condition itself was too broad (it fired on every new datastream, not just dedup ones), adding unnecessary onAssignmentChange() churn. I've replaced the anyDatastreamReady flag with anyDatastreamJoinedExistingTask, set true only when a newly-READY datastream's effective TASK_PREFIX matches an already-established datastream's — i.e. only for genuine dedup/reuse cases. Better name and a narrower/more correct condition. Pushed in the latest commit.
kanishkjaiswal2015
left a comment
kanishkjaiswal2015
left a comment
There was a problem hiding this comment.
a nit comment but LGTM overall
55bac32 to
18680e2
Compare
… an existing task When a new datastream dedupes into the task(s) of an already-established datastream (same effective TASK_PREFIX), the assignment (task names) does not change, so the normal assignment diff/rebalance path never touches the affected task's ZK node and the task's cached datastreams list on every instance goes stale. Narrow the fix to only broadcast a datastream update (which triggers every instance to refresh its locally-assigned tasks' datastreams lists) when a newly-READY datastream actually joins an existing task group, rather than on every datastream creation. This avoids extra onAssignmentChange() churn for the common case of a brand-new datastream getting a brand-new task. Also updates TestCoordinator to reflect that a task's representative datastream can now correctly be any datastream in its dedup group. Addresses review feedback from kanishkjaiswal2015.
18680e2 to
ecfddec
Compare
3 participants
Summary
When a new datastream is created that reuses the task(s) of an already-assigned datastream
(e.g. destination dedup — same
taskPrefix/partitions), the task's name/identity does notchange, so the normal assignment diff (
ZkAdapter#diffAssignmentNodes) never touches thattask's ZK node. As a result, every instance's cached
DatastreamTask#getDatastreams()list for that task goes stale and never picks up the newdatastream — even though
Coordinator#onDatastreamUpdate()already contains correct logicto refresh it. That refresh path is simply never triggered for this scenario.
Coordinatoralready has a working broadcast mechanism(
Coordinator#broadcastDatastreamUpdate->ZkAdapter#touchAllInstanceAssignments) thattouches every live instance's
/assignmentsnode, firing the existingonDatastreamUpdate()refresh on all instances. Today it's only invoked from thedatastream-update / pause-partitions REST paths. This change also invokes it from
Coordinator#handleDatastreamAddOrDelete()whenever any datastream transitionsINITIALIZING -> READY, so newly created datastreams (including dedup datastreams thatreuse an existing task) also trigger the refresh.
This is a small, low-risk change: it reuses an existing, already-in-production broadcast
mechanism from one additional trigger point rather than introducing new machinery or
touching core assignment-diffing logic.
Testing Done
Added
TestCoordinator#testTaskDatastreamsRefreshedAfterDedupeReusesExistingTask:creates a datastream, waits for its task assignment, then creates a second datastream that
dedupes into the same destination/task, and asserts the reused task's
getDatastreams()list is refreshed to include both streams.
Verified the test reproduces the bug: reverting only the
Coordinator.javachange causesthe new test to fail with the exact reported symptom (task's datastreams list still only
contains the original stream after the dedup stream becomes
READY). Restoring the fixmakes the test pass.