apiVersion: batch/v1
kind: Job
metadata:
  name: cluster-job-submit
  labels:
    app: flink
spec:
  template:
    spec:
      serviceAccountName: flink
      restartPolicy: OnFailure
      containers:
        - name: job-cluster
          image: wordcount2:latest
          imagePullPolicy: Never
          command: ["./bin/flink", "run-application",
                    "-p", "1",
                    "-t", "kubernetes-application",
                    "-Dkubernetes.cluster-id=wordcount-cluster",
                    "-Dkubernetes.jobmanager.service-account=flink",
                    "-Dkubernetes.namespace=default",
                    "-Dkubernetes.container.image.pull-policy=Never",
                    "-Dkubernetes.container.image=wordcount2:latest",
                    "-Dkubernetes.taskmanager.cpu=1",
                    "-Dtaskmanager.memory.process.size=1024m",
                    "-Djobmanager.memory.process.size=768m",
                    "-Dresourcemanager.taskmanager-timeout=3600000",
                    "-Dtaskmanager.numberOfTaskSlots=1",
                    "-Dkubernetes.container-start-command-template=%java% %classpath% %jvmmem% %jvmopts% %logging% %class% %args%",
                    "local:///opt/flink/usrlib/wordcount.jar",
                    "--input-topic", "words",
                    "--output-topic", "wordcount",
                    "--kafka-endpoint", "my-cluster-kafka-bootstrap.kafka:9092",
                    "--pipeline-type", "json"]
