[GH] (flink-kubernetes-operator/fix/session-cluster-deletion-fail-open): Workflow run "Flink Kubernetes Operator CI" failed!

[GitBox <gi...@apache.org>]

The GitHub Actions job "Flink Kubernetes Operator CI" on flink-kubernetes-operator.git/fix/session-cluster-deletion-fail-open has failed.
Run started by GitHub user spuru9 (triggered by gyfora).

Head commit for run:
94bc18cc03262f57fa125056c4ccd6c4b177b2db / Purushottam Sinha <ps...@confluent.io>
[FLINK-40542] Block session cluster deletion when running jobs cannot be determined

SessionReconciler.getNonTerminalJobs returned Set.of() on any failure to
query the cluster. cleanupInternal treats an empty set as "no running jobs,
safe to delete", so a transient JobManager outage during deletion of a
session cluster with block-on-unmanaged-jobs enabled (the default) silently
deleted the cluster while unmanaged jobs were still running - stopping them
without a checkpoint, with only a WARN.

Return Optional<Set<JobID>> so an indeterminate result (empty Optional) is
distinguishable from a determined "no jobs" (present, empty set).
cleanupInternal now blocks deletion and emits a CleanupFailed event on the
indeterminate case, mirroring the block-on-session-jobs branch. Also close
the RestClusterClient via try-with-resources and bound the jobs request with
the Flink client timeout instead of an unbounded get().

The fail-closed change unmasked a deletion deadlock: a session cluster that
never successfully deployed also has an unreachable JobManager, so it would
block deletion forever instead of being torn down. cleanupInternal now
checks isBeforeFirstDeployment() first and deletes immediately in that case,
mirroring ApplicationReconciler and SessionJobReconciler. Also guard against
a null jobs list, restore the interrupt flag on InterruptedException, and
log CleanupFailed warnings consistently with the existing session-jobs
branch.

Per review feedback, apply the same bounded-timeout treatment to the other
blocking REST calls that shared the unbounded get() pattern:
AbstractFlinkService#fetchSavepointInfo, #fetchCheckpointInfo,
#fetchCheckpointStats, #populateStateSize (bounded with the operator's
kubernetes.operator.flink.client.timeout), and the autoscaler's
ScalingMetricCollector#queryAggregatedMetricNames,
RestApiMetricsCollector#queryAggregatedVertexMetrics, #queryJmMetrics,
#queryAggregatedTmMetrics (bounded with job.autoscaler.flink.rest-client.timeout).

Generated-by: Claude Code (Claude Opus 4.8)

Report URL: https://github.com/apache/flink-kubernetes-operator/actions/runs/34254625675

With regards,
GitHub Actions via GitBox