Making ThroughputViolationMap creation logic more robust

[harshOSS]

Fix NPE in onAssignmentChange and improve atomicity of populateThroughputViolatingTopicsMap

Problem

Two bugs in Coordinator.java:

1. NPE in onAssignmentChange: getDatastreamTask(task) can return null for tasks not yet
   visible to the instance (e.g., during a partition rebalance or when a newly assigned task hasn't
   been written to ZK yet). The return value was passed directly into new DatastreamGroup(...)
   without a null check, causing a NullPointerException.

2. Non-atomic map update in populateThroughputViolatingTopicsMap: The write lock was acquired
   after the map was already cleared but before iteration completed. If an exception was thrown
   mid-iteration, _throughputViolatingTopicsMap would be left empty — dropping all previously
   cached throughput violation data from concurrent readers holding the read lock.

Fix

NPE (onAssignmentChange): Added .filter(Objects::nonNull) after the getDatastreamTask
call to skip null tasks rather than propagating the NPE.

Atomicity (populateThroughputViolatingTopicsMap): Moved the iteration outside the lock into a
local next map. The write lock is now held only for the clear() + putAll(next) swap, making
the visible state transition atomic. A mid-iteration failure no longer corrupts the live map. Also
tightened topic string parsing: .map(String::trim) is now applied before
.filter(s -> !s.isEmpty()) to correctly handle whitespace-only entries.

Changes

• datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java

Testing

• Existing unit tests pass.
• The NPE scenario is covered by the onAssignmentChange path for instances receiving tasks not
  yet committed to ZK.

[khandelwal-ayush]

Looks good.