Dennis-Mircea commented on code in PR #1202:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1202#discussion_r3977432775
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java:
##########
@@ -143,39 +144,55 @@ private void
recoverSession(FlinkResourceContext<FlinkDeployment> ctx) throws Ex
// Detects jobs which are not in globally terminated states
@VisibleForTesting
- Set<JobID> getNonTerminalJobs(FlinkResourceContext<FlinkDeployment> ctx) {
+ Optional<Set<JobID>>
getNonTerminalJobs(FlinkResourceContext<FlinkDeployment> ctx) {
LOG.debug("Starting nonTerminal jobs detection for session cluster");
- try {
- // Get all jobs running in the Flink cluster
- var flinkService = ctx.getFlinkService();
- var clusterClient =
flinkService.getClusterClient(ctx.getObserveConfig());
+ var flinkService = ctx.getFlinkService();
+ try (var clusterClient =
flinkService.getClusterClient(ctx.getObserveConfig())) {
var allJobs =
clusterClient
.sendRequest(
JobsOverviewHeaders.getInstance(),
EmptyMessageParameters.getInstance(),
EmptyRequestBody.getInstance())
- .get()
+ .get(
+
ctx.getOperatorConfig().getFlinkClientTimeout().toMillis(),
+ TimeUnit.MILLISECONDS)
.getJobs();
- // running job Ids
+ if (allJobs == null) {
+ return Optional.of(Set.of());
+ }
+
Set<JobID> nonTerminalJobIds =
allJobs.stream()
.filter(job ->
!job.getStatus().isGloballyTerminalState())
.map(JobDetails::getJobId)
.collect(Collectors.toSet());
- return nonTerminalJobIds;
+ return Optional.of(nonTerminalJobIds);
} catch (Exception e) {
LOG.warn("Failed to detect nonTerminal jobs in session cluster",
e);
- return Set.of();
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ return Optional.empty();
Review Comment:
I took a closer look on this one, and here by returning the
`Optional.empty()` we cannot differentiate between a broken session JobManager
and an unreachable session JobManager. Every failure lands in the same catch,
so further, `cleanupInternal` has no way to tell a JobManager that is
mid-restart from one that is crashlooping, in ImagePullBackOff, misconfigured,
or whose Deployment was removed by hand. If we have a broken session JobManager
(with or without unmanaged Flink session jobs), we end up in an endless loop
with no cleanup, where the finalizer is never removed, the CR stays in
Terminating, and the broken JM Deployment is never torn down.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java:
##########
@@ -143,39 +144,55 @@ private void
recoverSession(FlinkResourceContext<FlinkDeployment> ctx) throws Ex
// Detects jobs which are not in globally terminated states
@VisibleForTesting
- Set<JobID> getNonTerminalJobs(FlinkResourceContext<FlinkDeployment> ctx) {
+ Optional<Set<JobID>>
getNonTerminalJobs(FlinkResourceContext<FlinkDeployment> ctx) {
LOG.debug("Starting nonTerminal jobs detection for session cluster");
- try {
- // Get all jobs running in the Flink cluster
- var flinkService = ctx.getFlinkService();
- var clusterClient =
flinkService.getClusterClient(ctx.getObserveConfig());
+ var flinkService = ctx.getFlinkService();
+ try (var clusterClient =
flinkService.getClusterClient(ctx.getObserveConfig())) {
var allJobs =
clusterClient
.sendRequest(
JobsOverviewHeaders.getInstance(),
EmptyMessageParameters.getInstance(),
EmptyRequestBody.getInstance())
- .get()
+ .get(
+
ctx.getOperatorConfig().getFlinkClientTimeout().toMillis(),
+ TimeUnit.MILLISECONDS)
.getJobs();
- // running job Ids
+ if (allJobs == null) {
+ return Optional.of(Set.of());
+ }
+
Set<JobID> nonTerminalJobIds =
allJobs.stream()
.filter(job ->
!job.getStatus().isGloballyTerminalState())
.map(JobDetails::getJobId)
.collect(Collectors.toSet());
- return nonTerminalJobIds;
+ return Optional.of(nonTerminalJobIds);
} catch (Exception e) {
LOG.warn("Failed to detect nonTerminal jobs in session cluster",
e);
- return Set.of();
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ return Optional.empty();
}
}
@Override
public DeleteControl cleanupInternal(FlinkResourceContext<FlinkDeployment>
ctx) {
- var sessionJobs =
ctx.getJosdkContext().getSecondaryResources(FlinkSessionJob.class);
var deployment = ctx.getResource();
+ var status = deployment.getStatus();
+
+ if (status.getReconciliationStatus().isBeforeFirstDeployment()) {
Review Comment:
This guard here doesn't seem to me to help and to tackle a broken session
JobManager case. A broken session JobManager can most probably be deployed
already (aka `lastReconciledSpec != null`) and this guard be bypassed entirely.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
