Improve submitting/tracking job and fix

[elanv]

Changes

• Job submitting and tracking

  • Job submitter pod is finished after job submission.
  • Identify Flink job by extracting the job ID from termination-log of job submitter pod
  • Flink job is tracked by the operator itself (not by job submitter)
  • Shorten the bootstrap time of the job cluster by parallelizing cluster and job submit initialization

• Fix

  • Job recovery
    Fix bug that job recovery does not work if more than one job history remains in the job manager
  • Change latest savepoint to provided fromSavepoint when update.
    When updating with provided fromSavepoint, change latest savepoint (status.savepointLocation) to it also.

Resolves #294

[elanv]

I will add and fix some tests soon.

[functicons]

Thanks for the PR! I left a few comments.

[functicons]

A few more comments, thanks!

[functicons]

/gcbrun

[elanv]

Restored a CRD that was accidentally changed. Removed JobManager check start delay from submit script. This is because the time difference between the submitter and JM initialization is small and sometimes the submitter can take longer, such as downloading large file.

[functicons]

/gcbrun

[functicons]

I just did a test, the sample job finished successfully, but the status in the CR was not quite right, it was still in Pending status:

Status:
  Components:
    Job:
      Id:     af4f6808f9c78597623a626e300c73f5
      Name:   flinkjobcluster-sample-job
      State:  Pending
  ...
  Current Revision:  flinkjobcluster-sample-5d96cb58dd-1
  Last Update Time:  2020-12-06T04:55:33Z
  Next Revision:     flinkjobcluster-sample-5d96cb58dd-1
  State:             Stopped

[functicons]

I recommend we change the name suffix of the Job resource from job to job-submitter; otherwise, people might be confused when seeing it is in Completed status.

We also need to update the "Flink job" section of the user guide on how to monitor the job. Previously monitoring the job submitter makes sense, but with this change, it doesn't make much sense. I think we can briefly mention how the job is submitted.

[elanv]

Thanks for your review of something I missed.
Fixed finished job related issues.

Will proceed remaining works of renaming and docs.

[functicons]

Unit test is failing, could you fix it?

E1207 05:00:42.876848    3516 factory.go:35] Failed initializing volcano batch scheduler: invalid configuration: no configuration has been provided, try setting KUBERNETES_MASTER environment variable
--- FAIL: TestGetDesiredClusterState (0.01s)
    flinkcluster_converter_test.go:889: assertion failed: 
        --- desiredState.Job
        +++ expectedDesiredJob
          v1.Job{
                TypeMeta: v1.TypeMeta{},
                ObjectMeta: v1.ObjectMeta{
        -               Name:         "flinkjobcluster-sample-job-submitter",
        +               Name:         "flinkjobcluster-sample-job",
                        GenerateName: "",
                        Namespace:    "default",
                        ... // 13 identical fields
                },

[functicons]

I noticed another problem, in the current job status, it shows x-job-submitter in Succeeded status. It unclear whether the status is for the job submitter or the job. Maybe remove the Name field to make it less confusing?

Status:
    Components:
        Job:
            Id:     b2a57bdfdc5127ddbab05da9ec438168
            Name:   flinkjobcluster-sample-job-submitter
            State:  Succeeded

[elanv]

I noticed another problem, in the current job status, it shows x-job-submitter in Succeeded status. It unclear whether the status is for the job submitter or the job. Maybe remove the Name field to make it less confusing?

That's right. I think it would be better to remove it too.
Or it would be nice to keep the field and make it optional for later use, and not set that value now.

[functicons]

/gcbrun

[abroskin]

@elanv, could you please explain why you removed this functionality?
We use HA mode. So that, when the cluster is recreated (removed and and deployed again), the newly created jobmanager is able to restore cluster's state and the job. And in the same time the job submitter submitted the same job again. We ended up with two jobs.

[elanv]

@abroskin The job is now being tracked inside the operator (link) rather than the submitter script. And since the previous job tracking script had an issue like #294 with the flink operator's auto job recovery feature, this PR changed to track the job by ID unlike before. And is the spec.job.restartPolicy set to Never? If so, the flink operator should not automatically restart the job. In addition, it would be nice if you could explain more about the HA configuration and deployment.

[bains00]

Review plzz