In a streams group heartbeat, the coordinator refines the target
assignment into an intermediate assignment with warm-up tasks. The
target assignment it refines already reflects the membership change of
the heartbeat: maybeUpdateStreamsTargetAssignment applies that change
through an UpdatedMembersAndTargetAssignmentView, which adds a joining
member and relabels a replaced static member to its new member ID. The
refiner, however, was handed group.members(), which does not contain
that member yet, because the records of the heartbeat are only replayed
after the handler returns. The refiner therefore treated the member as
one that had left the group.

The worst effect is on a static member that rejoins under a new member
ID, as it does on every restart. The refiner kept the member's stateful
active tasks on the old member ID, which the replacement removes, so the
rejoined member was not given them and no member ran them until the next
refinement step. In addition, a new member did not get a warm-up task
for a stateful task that the target assignment moves to it, and a new
member that is the target standby holder of a migrating task was not
held back, so the task ran one standby more than configured while the
migration lasted.

maybeUpdateStreamsTargetAssignment now returns the view's members
together with the target assignment, and the refiner is handed those
members whenever it refines that target assignment. The other call site,
which refines the current target assignment of a Stable group when the
heartbeat changes nothing, keeps using group.members(): a heartbeat that
adds or replaces a member never takes that path, because a new member
bumps the group epoch and a replaced static member has left with epoch
-2, so the group is not Stable.

See https://issues.apache.org/jira/browse/KAFKA-21321. The tests cover
two static member rejoins with the real refiner and warm-up tasks
enabled: one from a new process, where the target assignment is
recomputed, and one from the same process with no intermediate
assignment cached for the epoch, as after a coordinator failover. Both
fail without this change, because the rejoined member is given no active
task.

Reviewers: Sean Quah squah@confluent.io