Job tracking routine changes for the blocking mode

[elanv]

I recently found that the job tracking routine has changed. It may be very rare, but I think there may be exceptional situations in which multiple jobs or duplicate jobs are submitted to the cluster due to user mistakes or errors in the associated system. Because it is difficult to predict or cope with all the various side effects from that, in order to prevent such a situation, I have previously implemented to identify and track the job by the ID obtained from the submitted job.

As for blocking mode, job tracking routine seems to be changed as it is now because the job ID cannot be obtained immediately, but I am a little concerned about taking these issues to support blocking mode.

flink-on-k8s-operator/controllers/flinkcluster_observer.go

Lines 301 to 304 in 35d5413

	flinkJobStatus.flinkJob = &flinkJobList.Jobs[0]
	for _, job := range flinkJobList.Jobs[1:] {
	flinkJobStatus.flinkJobsUnexpected = append(flinkJobStatus.flinkJobsUnexpected, job.Id)
	}

In addition to selecting and tracking an arbitrary job when there are multiple jobs as above, when handling unexpected jobs, they are currently arbitrarily cleared collectively as follows. Rather, what do you think about to make users to check and handle unexpected jobs themselves like PR #107?

flink-on-k8s-operator/controllers/flinkcluster_reconciler.go

Lines 497 to 502 in 35d5413

	// Cancel unexpected jobs
	if len(observed.flinkJobStatus.flinkJobsUnexpected) > 0 {
	log.Info("Cancelling unexpected running job(s)")
	err = reconciler.cancelUnexpectedJobs(false /* takeSavepoint */)
	return requeueResult, err
	}

<PR #107>
https://github.com/elanv/flink-on-k8s-operator/blob/2ad3293cf85ea92a9b6d020b707a878a0ec59ed2/controllers/flinkcluster_reconciler.go#L509-L520

[regadas]

Hi @elanv sorry for the late reply.

The blocking mode was added to support the use cases like collecting metrics if users are relying on Apache Beam; It's not possible to do it in detached mode; This, however, can potentially be deprecated once we add support for Application mode but we are not there yet.

Right now I want us to keep supporting this mode and I do like the idea of if there's already running jobs to fail the submission and let the user address it.

[elanv]

Thank you for explaining @regadas.
I'm also looking for support from upstream to resolve this issue.
https://issues.apache.org/jira/browse/FLINK-24365

[regadas]

Hi @elanv! I was revisiting this issue and from a snd look the JobId does get printed out in both modes; So guess to accurately track it we can inspect the pod logs get it from there instead

[elanv]

Looks good! Thanks for the work @regadas.
When that PR is merged, I would proceed my work.

When I looked more closely, the ID was not printed out in version <= 1.9 with blocking mode. (This could be guided by document)

For now, your work seems the best way. And if application mode is supported later, it would be good if a more stable method is introduced because it doesn't seem reliable getting the ID from a pod, the transient, either via termination log or via log stream. If Beam is supported in application mode later, it may be worth considering introducing Flink REST API that can submit a job with the ID created by the operator. I tried to utilize REST API in the past, but it was not applied due to several issues. (GoogleCloudPlatform/flink-on-k8s-operator#372)

[regadas]

Hi @elanv!

Are you sure it's not being logged? Looking at the release 1.9.3 it should https://github.com/apache/flink/blob/6d23b2c38c7a8fd8855063238c744923e1985a63/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java#L764

[elanv]

@regadas I tested again a while ago with Flink versions 1.9.0 ~ 1.9.3 and the result is the same.
I canceled the job after 8 minunute in Flink web console, additional logs are printed out after "The program finished with the following exception", but there was no ID until the end.

---------- Submitting job ----------
/opt/flink/bin/flink run --jobmanager t13-jobmanager:8081 --class org.apache.flink.streaming.examples.statemachine.StateMachineExample --parallelism 1 ./examples/streaming/StateMachineExample.jar --brokers sm-cp-kafka.kafka.svc.cluster.local:9092 --kafka-topic flink-app
Starting execution of program
Usage with built-in data generator: StateMachineExample [--error-rate <probability-of-invalid-transition>] [--sleep <sleep-per-record-in-ms>]
Usage with Kafka: StateMachineExample --kafka-topic <topic> [--brokers <brokers>]
Options for both the above setups:
	[--backend <file|rocks>]
	[--checkpoint-dir <filepath>]
	[--async-checkpoints <true|false>]
	[--incremental-checkpoints <true|false>]
	[--output <filepath> OR null for stdout]

Reading from kafka topic flink-app @ sm-cp-kafka.kafka.svc.cluster.local:9092


------------------------------------------------------------
 The program finished with the following exception:

org.apache.flink.client.program.ProgramInvocationException: Job failed. (JobID: 7a95df7eb581c2f98bf7a1f26eb24af1)
	at org.apache.flink.client.program.rest.RestClusterClient.submitJob(RestClusterClient.java:262)
	at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:338)
	at org.apache.flink.streaming.api.environment.StreamContextEnvironment.execute(StreamContextEnvironment.java:60)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1507)
	at org.apache.flink.streaming.examples.statemachine.StateMachineExample.main(StateMachineExample.java:142)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:604)
	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:466)
	at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:274)
	at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:746)
	at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:273)
	at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:205)
	at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1008)
	at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1081)
	at org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30)
	at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1081)
Caused by: org.apache.flink.runtime.client.JobCancellationException: Job was cancelled.
	at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:149)
	at org.apache.flink.client.program.rest.RestClusterClient.submitJob(RestClusterClient.java:259)
	... 18 more

[regadas]

Ahah I see; I also initially thought that could be do to logging properties but according to the line I shared it should have been logging to stdout as well ...

So this means that neither of the solutions will work for older versions.

[elanv]

In detatched mode, the ID was printed out in the version <= 1.9. It looks like Flink client bug.