[k8s][flink-operator][java] Hi, How can I submit a job to a session clusster using flink k8s operator Java API? I see examples only for application cluster with a Job. I've created a session cluster that is running on k8s. I have a

FlinkSessionJobSpec

object but not sure how to submit it, what API should be called.

FSJ is a custom resource, check Custom Resources APIs https://itnext.io/handling-kubernetes-custom-resources-in-java-using-fabric8-kubernetes-client-part-2-typed-cc0560d9af0d

hm tried that:

@Test
public void submitSessionJob() {

    JobSpec jobSpec = JobSpec.builder()
        .jarURI(
            "local:///opt/flink/examples/streaming/StateMachineExample.jar")
        .parallelism(2)
        .upgradeMode(UpgradeMode.STATELESS)
        .build();

    FlinkSessionJobSpec sessionJobSpec = FlinkSessionJobSpec.builder()
        .job(jobSpec)
        .deploymentName("basic-session-deployment-only-example")
        .build();

    FlinkSessionJob flinkSessionJob = new FlinkSessionJob();
    flinkSessionJob.setKind("FlinkSessionJob");
    flinkSessionJob.setApiVersion("<http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>");

    ObjectMeta meta = new ObjectMeta();
    meta.setNamespace("default");
    meta.setName("basic-session-job-only-example");
    flinkSessionJob.setMetadata(meta);
    flinkSessionJob.setSpec(sessionJobSpec);

    try (KubernetesClient kubernetesClient = new KubernetesClientBuilder().build()) {
        kubernetesClient.resource(flinkSessionJob).create();
    }
}

That ran without any exceptions but Job was not submitted to the session cluster. When I called

kubectl get FlinkSessionJob

I got this:

kristoff@LAPTOP-6V16VCBI:~/flinkK8sPoc$ kubectl get FlinkSessionJob
NAME                               JOB STATUS    LIFECYCLE STATE
basic-session-job-only-example     RECONCILING   STABLE

you should check flink-kubernetes-operators logs then