KAFKA-10199: Add task updater metrics, part 1

[guozhangwang]

0. Moved pausing-tasks logic out of the commit-interval loop to be on the top-level loop, similar to resuming tasks.
1. Added thread-level restoration metrics.
2. Related unit tests.

Should only be reviewed after #13025

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[guozhangwang]

cc @lucasbru , would update after #13025 is done, and the pausing logic is extracted out of the checkAllUpdatingTaskStates function.

[guozhangwang]

ping @lucasbru @cadonna for reviews.

[guozhangwang]

This is to avoid checkstyle rules on func complexity, without much logical change (except letting restoreChangelog to return restored records count).

[guozhangwang]

This is related to 0) in PR description.

[guozhangwang]

Some minor logging fixes piggy-backed here.

[lucasbru]

lgtm, approving. Added minor comments, update as you prefer

[lucasbru]

Not sure, but is there any performance concern around running this loop in every single iteration of runOnce ?

[guozhangwang]

I think the perf impact should be small since pause/resume are not commonly used, and if the named topology are not paused, then checking the status is just a few cpu cycles.

Another motivation is that if we remove named topologies, than pausing / resuming would always be impact on all tasks, in which case we could have a simpler check (e.g. just check a single flag) which would be even cheaper. So it's probably better to maintain the code layout in this way.

[lucasbru]

Update the log message as well. This function isn't really checking task states anymore

[lucasbru]

As I understand it, this function will be called quite frequently to export metrics. We only need the size of the collection. It could make sense to avoid the allocations here and just implement a getNumberOfUpdaingActiveTasks as a non-essential but free optimization. Similar for getPausedStandbyTasks etc. pp.

[guozhangwang]

That's a good point, will update.