KAFKA-20372: Fix flaky test in NamedTopologyIntegrationTest

[chickenchickenlove]

This PR fixes a flaky test in NamedTopologyIntegrationTest:
shouldAddToEmptyInitialTopologyRemoveResetOffsetsThenAddSameNamedTopologyWithRepartitioning.

The failure is caused by a race in the task lifecycle during named
topology removal.

In some runs, local tasks for the topology move to active/running
quickly. In other runs, some tasks can remain in a transitional state
for a short time, such as state update / restore, before becoming fully
active.

The test currently removes and cleans up the topology without waiting
for this lifecycle to settle. Because of that,
removeNamedTopology(...) may run while some local tasks are still
transitioning, and cleanUpNamedTopology(...) may follow before those
tasks are fully gone. This makes the subsequent re-add flow
non-deterministic and leads to intermittent failures.

To make the test deterministic, this PR adds explicit waits around the
remove / cleanup sequence:

• wait until all local tasks for the topology are running before calling
  removeNamedTopology(...)
• wait until no local tasks remain for the topology before calling
  cleanUpNamedTopology(...)

This keeps the test semantics unchanged and only stabilizes task
lifecycle timing in the test.

Test result in my local

• Before: 12/17
• After: 89/89

Considered Alternatives

1. Send records to partition 1 as well
   This would make it more likely that both partition 0 and partition 1
   tasks are initialized before removal.

However, this changes the test input and can also introduce additional
rebalance and task-transition timing effects, since partition 1
processing becomes part of the exercised path. In other words, it may
reduce one symptom while introducing a different concurrency surface
related to task initialization and reassignment.

More importantly, the root cause of the flake is that the test does not
wait for task lifecycle transitions to settle before calling
removeNamedTopology(...) and cleanUpNamedTopology(...). Adding more
input does not address that directly.

2. Reduce the input topic partition count from 2 to 1
   This would avoid the multi-partition timing issue by construction.
   However, this also weakens the coverage of the test by removing the
   multi-partition setup entirely. Since the production code supports
   multiple partitions, reducing the topic to a single partition would hide
   the race instead of synchronizing with the intended lifecycle of the
   test.

Reviewers: Lianet Magrans 98415067+lianetm@users.noreply.github.com

[chickenchickenlove]

@lianetm Hi!
When you get a chance, could you take a look?

[chickenchickenlove]

@lianetm
Gently ping!
When you get a chance, please take a look. 🙇‍♂️

[lianetm]

Thanks for looking into this!

[lianetm]

if there are no tasks, this allTasksRunning would still return true, which is ok in the current test usage, but I wonder if it may be confusing. Should we add a java doc to call out the behaviour?

[lianetm]

nit: seems clearer if we excplititly mention the final state that we didn't get to (~ tasks did not transition to running state as expected)

[lianetm]

nit

Suggested change

	"topology tasks still exist internally after remove"
	"tasks still exist internally after topology removed"