Make KubernetesWorkItem.shutdown idempotent

[kfaraz]

Description

This patch tries to address the following condition due to a bug in fabric8 client.

The callback executor in TaskQueue can get stuck in the following scenario.

• TaskQueue tries to shutdown a task via notifyStatus() and queues up a TaskRunner.shutdown() on the callback executor.
• TaskQueue then removes the entry for this task from its in-memory data-structures
• TaskRunner.shutdown() remains stuck due to the fabric8 bug
• TaskRunner still has the task in its memory
• When the TaskQueue periodically manages itself via startPendingTasksOnRunner(),
  it realizes that the TaskRunner has this unknown task and tries to shut it down again.

Fix

Subsequent calls to KubernetesWorkItem.shutdown() should not block and return immediately.

---

This PR has:

• been self-reviewed.
  • using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
• added integration tests.
• been tested in a test Druid cluster.

[capistrant]

Seems like a logical solution to me. I think this might be a time to take a second and consider the log watch in the shutdown and if it is even necessary. what do you think?

[capistrant]

do we even need to be calling startWatchingLogs() here? Kind of off topic for the PR. But part of the reason shutdown is slow and can even hang, is because of the LogWatch being unhealthy. #18444 was implemented to put a time limit on log persist for saving off task logs. In that code path we are actually writing the logs out somewhere. I guess I don't see how the call here is useful at all if we are trying to just shut the task down?

[kfaraz]

Yeah, I agree, the only apparent effect of calling startWatchingLogs() is to initialize KubernetesPeonLifecycle.logWatch field which would be initialized anyway via join() -> finally -> saveLogs() -> doSaveLogs().

We can explore that in a separate PR.

[capistrant]

#18579 opened this for proposing dropping the LogWatch init in shutdown