func canTakeSavepoint(cluster v1beta1.FlinkCluster) bool {
	var jobSpec = cluster.Spec.Job
	var savepointStatus = cluster.Status.Savepoint
	var jobStatus = cluster.Status.Components.Job
	return jobSpec != nil && jobSpec.SavepointsDir != nil &&
		!isJobStopped(jobStatus) &&
		(savepointStatus == nil || savepointStatus.State != v1beta1.SavepointStateInProgress)
}
Checks whether it is possible to take savepoint.
package controllers
import (
	"bytes"
	"encoding/json"
	"fmt"
	v1beta1 "github.com/googlecloudplatform/flink-operator/api/v1beta1"
	"github.com/googlecloudplatform/flink-operator/controllers/history"
	"gopkg.in/yaml.v2"
	appsv1 "k8s.io/api/apps/v1"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	extensionsv1beta1 "k8s.io/api/extensions/v1beta1"
	"k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"os"
	"strconv"
	"strings"
	"time"
)
const (
	ControlSavepointTriggerID = "SavepointTriggerID"
	ControlJobID              = "jobID"
	ControlRetries            = "retries"
	ControlMaxRetries         = "3"
	SavepointTimeoutSec = 900  
	RevisionNameLabel = "flinkoperator.k8s.io/revision-name"
	SavepointAgeForJobUpdateSec      = 300
	SavepointRequestRetryIntervalSec = 10
)
type UpdateState string
const (
	UpdateStatePreparing  UpdateState = "Preparing"
	UpdateStateInProgress UpdateState = "InProgress"
	UpdateStateFinished   UpdateState = "Finished"
)
type objectForPatch struct {
	Metadata objectMetaForPatch `json:"metadata"`
}
type objectMetaForPatch struct {
	Annotations map[string]interface{} `json:"annotations"`
}
func getFlinkAPIBaseURL(cluster *v1beta1.FlinkCluster) string {
	clusterDomain := os.Getenv("CLUSTER_DOMAIN")
	if clusterDomain == "" {
		clusterDomain = "cluster.local"
	}
	return fmt.Sprintf(
		"http://%s.%s.svc.%s:%d",
		getJobManagerServiceName(cluster.ObjectMeta.Name),
		cluster.ObjectMeta.Namespace,
		clusterDomain,
		*cluster.Spec.JobManager.Ports.UI)
}
func getConfigMapName(clusterName string) string {
	return clusterName + "-configmap"
}
func getJobManagerStatefulSetName(clusterName string) string {
	return clusterName + "-jobmanager"
}
func getJobManagerServiceName(clusterName string) string {
	return clusterName + "-jobmanager"
}
func getJobManagerIngressName(clusterName string) string {
	return clusterName + "-jobmanager"
}
func getTaskManagerStatefulSetName(clusterName string) string {
	return clusterName + "-taskmanager"
}
func getJobName(clusterName string) string {
	return clusterName + "-job-submitter"
}
type TimeConverter struct{}
func (tc *TimeConverter) FromString(timeStr string) time.Time {
	timestamp, err := time.Parse(
		time.RFC3339, timeStr)
	if err != nil {
		panic(fmt.Sprintf("Failed to parse time string: %s", timeStr))
	}
	return timestamp
}
func (tc *TimeConverter) ToString(timestamp time.Time) string {
	return timestamp.Format(time.RFC3339)
}
func setTimestamp(target *string) {
	var tc = &TimeConverter{}
	var now = time.Now()
	*target = tc.ToString(now)
}