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]

/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

[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