Flink deployment managers

The deployment managers below run streaming scenarios on infrastructure that provisions a Flink cluster for them. Each one is a JAR in the managers directory of the Designer and is configured as a deployment target of a scenario type, see Scenario Deployment configuration.

Choosing a Flink Deployment Manager

Besides flinkStreaming, three Deployment Managers run Flink scenarios on infrastructure that provisions a Flink cluster for them:

• flinkStreaming deploys to a Flink cluster that is already running - a standalone or session cluster reached through its REST API - or to the MiniCluster embedded in the Designer.
• flink-k8s-operator creates a FlinkDeployment resource per scenario for the Flink Kubernetes Operator, so every scenario gets its own Flink cluster in your Kubernetes cluster. Choose it when you run Kubernetes yourself and want per-scenario isolation without operating a long-running Flink cluster.
• awsManagedFlink runs every scenario as an application of Amazon Managed Service for Apache Flink. Choose it when the data stays in AWS and nobody should operate Flink at all; it supports the fewest operations - no stop with savepoint, no savepoints on demand, no live data preview.
• ververica deploys to Ververica Platform, per-job or to a session cluster. Choose it when VVP already runs your Flink workloads; it is the only one of the three supporting stop with savepoint, savepoints on demand and deployment targets created at runtime. It needs a Nussknacker build for Flink 1.20, the version Ververica Runtime runs.

All three run scenario tests on a Flink MiniCluster inside the Designer, and each of them can be one of the targets of a scenario type, next to flinkStreaming targets.

Flink Kubernetes Operator

Deployment Manager of type flink-k8s-operator runs every scenario as its own Flink application cluster, by creating a FlinkDeployment resource (flink.apache.org/v1beta1) that the Apache Flink Kubernetes Operator turns into a JobManager and TaskManagers. The operator itself has to be installed in the cluster and watch k8sNamespace.

deploymentConfig {
  type: "flink-k8s-operator"
  k8sNamespace: "nussknacker"
  flinkImage: "<Flink image>"
  flinkVersion: "v2_2"
  jobJarStoragePVCName: "job-jars-pvc"
  jobJarStorageMountPoint: "/mnt/jobjars"
  jobStateStoragePVCName: "job-states-pvc"
  jobStateStorageMountPoint: "/mnt/jobstates"
}

Parameter	Importance	Type	Default value	Description
k8sNamespace	High	string		Namespace the FlinkDeployment resources are created, listed and deleted in.
flinkImage	High	string		Image of the JobManager and TaskManager pods.
flinkVersion	High	string		spec.flinkVersion of the FlinkDeployment, in the operator's notation, e.g. v2_2. It has to match flinkImage.
jobJarStoragePVCName	High	string		PersistentVolumeClaim holding the scenario JAR files, mounted in the Flink pods at jobJarStorageMountPoint.
jobJarStorageMountPoint	High	string		Where the JAR storage is mounted - in the Flink pods, and in the Designer, which writes the JAR files there unless podForJobJars is set.
jobStateStoragePVCName	High	string		PersistentVolumeClaim for checkpoints, savepoints and high-availability data, mounted in the Flink pods at jobStateStorageMountPoint.
jobStateStorageMountPoint	High	string		Mount point of the state storage in the Flink pods.
podForJobJars	Low	string		For a Designer that cannot mount the JAR storage, e.g. one running outside the cluster: name of a pod in k8sNamespace that mounts it at jobJarStorageMountPoint, as the Flink pods do. The JAR file is then streamed into that pod with kubectl exec, so kubectl has to be on the Designer's PATH.
serviceAccount	Medium	string	flink-service-account	Service account of the Flink pods (spec.serviceAccount).
deploymentNamePrefix	Low	string	flink-	Prefix of the FlinkDeployment name; the rest of the name is the id of the scenario's first deployment.
jobParallelism	Medium	int	1	spec.job.parallelism.
numberOfTaskSlotsInTaskManager	Medium	int	1	taskmanager.numberOfTaskSlots.
taskManagerReplicas	Medium	int	1	Number of TaskManager pods.
jobManagerMemory	Medium	string	1024m	JobManager memory.
jobManagerCpu	Medium	int		JobManager CPU; the operator's default when not set.
taskManagerMemory	Medium	string	1024m	TaskManager memory.
taskManagerCpu	Medium	int		TaskManager CPU; the operator's default when not set.
jobUpgradeMode	Medium	string	savepoint	spec.job.upgradeMode: how the operator carries the state over when the scenario is redeployed - savepoint, last-state or stateless.
allowNonRestoredState	Low	boolean	true	spec.job.allowNonRestoredState: whether a redeployed scenario may start from a savepoint containing state that no node of the new version claims.
additionalFlinkConfiguration	Medium	map of strings	{}	Added to spec.flinkConfiguration, overriding the entries set by Nussknacker (see below).
nodeSelector	Low	map of strings	{}	Node selector of the Flink pods.
additionalVolumes	Low	list of objects	[]	Extra pod volumes: { name, persistentVolumeClaim: { claimName } }, { name, hostPath: { path, type } } or { name, secret: { secretName } }.
additionalVolumeMounts	Low	list of objects	[]	Extra mounts of the Flink container: { name, mountPath, subPath, readOnly }, subPath and readOnly optional.
additionalLibs.imageName	Low	string		Image of an init container that copies the files from additionalLibs.libsPath into the Flink pods.
additionalLibs.libsPath	Low	string		Directory in that image the files are copied from.
additionalLibs.libsNames	Low	list of strings		The copied files that are mounted into /opt/flink/lib of the Flink container.
additionalLibs.imagePullSecrets	Low	string		Image pull secret for additionalLibs.imageName.
modelArtifact	Low	object		See Reducing the deployment artifact.
miniCluster.config	Low	map of strings	{}	Configuration of the Flink MiniCluster in the Designer that runs scenario tests.
scenarioTesting.timeout	Low	duration	55 seconds	Timeout of a scenario test.
scenarioTesting.parallelism	Low	int	1	Parallelism of a scenario test.

How a scenario is deployed:

• The model is packed into one JAR file named after its SHA-256 (jars/<sha256>.jar) and copied onto the JAR storage, where the Flink pods read it from (jarURI: local://<jobJarStorageMountPoint>/jars/<sha256>.jar). The scenario, its version, the deployment data and the model configuration are passed as program arguments of the job.
• Nussknacker sets these entries of spec.flinkConfiguration: taskmanager.numberOfTaskSlots, a Prometheus metrics reporter on port 9249, high-availability.type: kubernetes, the checkpoint, savepoint and high-availability directories under jobStateStorageMountPoint, and the nu-scenario-name, nu-process-version and nu-deployment-id entries Nussknacker uses to find the resource of a scenario and to report its version. Do not override the last three.
• The job restarts from the latest savepoint on failure (restartPolicy: FromSavepointOnFailure).
• Deploying a scenario that already has a FlinkDeployment replaces the spec of that resource, and the operator upgrades the job according to jobUpgradeMode. Otherwise a new FlinkDeployment is created.
• Cancel deletes the FlinkDeployment and waits up to 30 seconds for it to disappear. Nussknacker takes no savepoint of its own, so the next deployment of a canceled scenario starts without state. Stop works the same as cancel, and making a savepoint on demand is not supported.

Statuses come from the job state the operator reports: CREATED, RECONCILING, SCHEDULED, DEPLOYING and INITIALIZING are shown as during deploy, RUNNING as running, FINISHED as finished (during deploy while the resource is upgrading), CANCELLING as during cancel, CANCELED as canceled, RESTARTING as restarting and FAILED as failed. A scenario without a FlinkDeployment is canceled.

Requirements and limitations:

• The Designer talks to the Kubernetes API with the in-cluster service account when it runs in Kubernetes, and with the default kubeconfig otherwise. It needs get, list, create, update and delete on flinkdeployments in k8sNamespace.
• Scenario tests run on a Flink MiniCluster inside the Designer.
• Live data preview reads modelConfig.liveDataPreview.storage, which has to be DESIGNER_DB or DESIGNER_API when live data preview is enabled - without it the Designer refuses the configuration.
• It can be one of the configured targets, but declares no target type, so targets of this kind cannot be created at runtime.

AWS Managed Service for Apache Flink

Deployment Manager of type awsManagedFlink runs every scenario as an application of Amazon Managed Service for Apache Flink, with the Flink 2.2 runtime.

scenarioTypes {
  "streaming-aws" {
    deploymentConfig {
      type: "awsManagedFlink"
      credentials {
        provider: "default"
      }
      region: "eu-central-1"
      s3BucketName: "nussknacker-applications"
      serviceExecutionRoleArn: "arn:aws:iam::123456789012:role/nussknacker-managed-flink"
      cloudWatchLogStreamArn: "arn:aws:logs:eu-central-1:123456789012:log-group:nussknacker:log-stream:scenarios"
    }
    modelConfig {
      classPath: ["model/defaultModel.jar", "model/flinkExecutor.jar", "model/awsManagedFlinkExecutorPlugin.jar",
                  "components/flink", "components/common"]
    }
    category: "Default"
  }
}

Parameter	Importance	Type	Default value	Description
region	High	string		AWS region of the applications and of the S3 bucket, e.g. eu-central-1.
s3BucketName	High	string		S3 bucket the application JAR file and the deployment properties are uploaded to.
serviceExecutionRoleArn	High	string		IAM role the applications run with.
credentials.provider	High	string		default - the AWS SDK default credentials chain (environment variables, system properties, web identity token, profile, container or instance role); static - the keys below.
credentials.accessKeyId	Medium	string		Access key id, required with provider: "static".
credentials.secretAccessKey	Medium	string		Secret access key, required with provider: "static".
cloudWatchLogStreamArn	Medium	string		CloudWatch log stream the applications log to. When set, it is added to, or updated in, an existing application on the next deployment.
miniCluster.config	Low	map of strings	{}	Configuration of the Flink MiniCluster in the Designer that runs scenario tests.
scenarioTesting.timeout	Low	duration	55 seconds	Timeout of a scenario test.
scenarioTesting.parallelism	Low	int	1	Parallelism of a scenario test.

model/awsManagedFlinkExecutorPlugin.jar has to be in modelConfig.classPath of the scenario type: it is what lets the job read its scenario from S3 at startup. Put it only in the classpath of the scenario types deployed to AWS: it also replaces the way the Flink job environment is prepared, leaving it to the application's configuration in AWS.

How a scenario is deployed:

• The whole model classpath is merged into one JAR file, nussknacker-application-code-<hash>.jar, built once per Designer run. Entries that occur in several JAR files with different content fail the build, unless a rule in modelConfig.applicationJarMergeRules resolves them, e.g. applicationJarMergeRules: [{ pattern: "META-INF/some.properties", strategy: "concat" }]. The pattern is an Ant-style path, and the strategy one of concat, discard, deduplicate, filterDistinctLines, first and rename.
• The scenario, its version, the deployment data and the model configuration are written to deployment-properties-<hash>.json. Both files are uploaded to the root of s3BucketName, each only once - an object that is already there is not overwritten. Nothing is ever removed from the bucket, so set a lifecycle rule on it.
• The application is named after the scenario: characters other than letters, digits, _, . and - become _, and a hash of the name is appended. As AWS limits the name to 128 characters, a scenario name longer than 111 characters cannot be deployed.
• A new application is created with the S3 locations of both files and started. An existing one gets its code and properties updated, and is started if it was not running (READY); a running application is updated in place.
• Cancel stops the application, which stays in AWS; it is not deleted.

Statuses come from the application status: RUNNING and AUTOSCALING are shown as running, STARTING and UPDATING as during deploy, STOPPING, FORCE_STOPPING and DELETING as during cancel, READY as canceled, ROLLING_BACK, ROLLED_BACK and MAINTENANCE as restarting.

Permissions:

• The identity of the Designer: s3:PutObject on the bucket; kinesisanalytics:DescribeApplication, CreateApplication, UpdateApplication, StartApplication, StopApplication and AddApplicationCloudWatchLoggingOption; iam:PassRole on the service execution role.
• The service execution role: trusted by kinesisanalytics.amazonaws.com, with s3:GetObject on the bucket and, with cloudWatchLogStreamArn, the permissions to write to that log group.

Limitations:

• Nussknacker sets no Flink application configuration - checkpointing, parallelism, scaling and snapshots are the defaults of AWS and can be changed on the application in AWS.
• Stop, making a savepoint and deploying from a chosen savepoint are not supported. Nussknacker passes no restore configuration when it starts or updates an application, so whether the state of the previous run is restored follows the snapshot settings of the application in AWS.
• Live data preview is not available, and scenario tests run on a Flink MiniCluster inside the Designer.
• It can be one of the configured targets, but declares no target type, so targets of this kind cannot be created at runtime.

Ververica Platform

Deployment Manager of type ververica runs scenarios as deployments of Ververica Platform (VVP), through its REST API - either one Flink cluster per scenario on a VVP deployment target (perJob), or as jobs on a session cluster that is run outside Nussknacker (session).

Flink version

Ververica Runtime images run Flink 1.20, while Nussknacker is built on Flink 2.3. Deploying to Ververica Platform needs a Nussknacker build made for Flink 1.20 - contact us for one. A scenario built against Flink 2.3 does not run on a Ververica Runtime image.

deploymentConfig {
  type: "ververica"
  baseUrl: "https://vvp.example.com"
  auth {
    type: "apiToken"
    token: ${VVP_API_TOKEN}
  }
  namespace: "default"
  workspace: "default"
  engineVersion: "nussknacker-4.5-jdk17-flink-1.20"
  deploymentMode {
    type: perJob
    deploymentTargetName: "k8s-target"
  }
  flinkConfiguration {
    "state.backend.type": "rocksdb"
  }
}

Parameter	Importance	Type	Default value	Description
baseUrl	High	string		URL of VVP. A path prefix, e.g. of a reverse proxy, is kept.
engineVersion	High	string		VVP engine version of the deployments. It has to resolve to a runtime image prepared for Nussknacker - see below.
deploymentMode.type	High	string		perJob or session.
deploymentMode.deploymentTargetName	High	string		(perJob only, required) VVP deployment target the scenarios run on.
deploymentMode.jobResources.jobmanager.cpu	Medium	number	1	(perJob only) CPUs of the JobManager.
deploymentMode.jobResources.jobmanager.memory	Medium	string	1Gi	(perJob only) Memory of the JobManager.
deploymentMode.jobResources.taskmanager.cpu	Medium	number	1	(perJob only) CPUs of a TaskManager.
deploymentMode.jobResources.taskmanager.memory	Medium	string	1Gi	(perJob only) Memory of a TaskManager.
deploymentMode.clusterName	High	string		(session only, required) Session cluster the scenarios run on. It is provisioned, started and stopped outside Nussknacker and has to be RUNNING when a scenario is deployed.
auth.type	High	string	none	none, apiToken, credentials or sso. Without the auth block no Authorization header is sent.
auth.token	High	string		(apiToken) API token sent as a bearer token.
auth.username, auth.password	High	string		(credentials) Exchanged for a token at POST <baseUrl>/api/v1/auth/tokens.
auth.ssoId	High	string		(sso) Id of the SSO client; the token is taken from the access_token cookie set by GET <baseUrl>/authentication/callback?sso_id=<ssoId>.
namespace	Medium	string	default	VVP namespace.
workspace	Medium	string	default	VVP workspace.
flinkConfiguration	Medium	map of strings	{}	Flink configuration of the deployments (flinkConf).
modelArtifact	Low	object		See Reducing the deployment artifact.
modelConfigOverride	Low	object		Merged over the model configuration the job runs with, e.g. to point at files that only exist in the VVP pods.
deleteFailedDeploymentOnCancel	Low	boolean	false	Canceling a scenario whose job has failed deletes the VVP deployment, with its savepoints and history.
savepointPolling.interval	Low	duration	1 second	How often a savepoint being made, or the one of a stop, is checked.
savepointPolling.timeout	Low	duration	60 seconds	How long to wait for such a savepoint.
jobStopPolling.interval	Low	duration	1 second	How often a redeploy checks that the previous job has stopped.
jobStopPolling.timeout	Low	duration	5 minutes	How long a redeploy waits for the previous job to stop with a savepoint.
requestTimeout	Low	duration	1 minute	Timeout of a call to VVP.
artifactUploadTimeout	Low	duration	10 minutes	Timeout of the model JAR upload.
tokenRefreshMargin	Low	duration	5 minutes	(credentials, sso) How long before its expiry a token is replaced.
miniCluster.config	Low	map of strings	{}	Configuration of the Flink MiniCluster in the Designer that runs scenario tests.
scenarioTesting.timeout	Low	duration	55 seconds	Timeout of a scenario test.
scenarioTesting.parallelism	Low	int	1	Parallelism of a scenario test.

Notes on the configuration:

• Unknown keys are ignored, so a misspelled key silently falls back to its default.
• A savepointPolling or jobStopPolling block that gives only one of its fields gets 1 second / 60 seconds for the other one - jobStopPolling { interval: 2s } waits 60 seconds, not 5 minutes.
• Tokens obtained with credentials or sso are cached and replaced before they expire. On a 401 (or a redirect to the SSO callback) the token is dropped and the call repeated once.

The runtime image: the images of Ververica Runtime ship flink-scala for Scala 2.12, while Nussknacker needs the Scala 2.13 one in /flink/lib. The image VVP runs is chosen by the deployment's engine version, so register an engine version in VVP backed by an image with the Scala 2.13 flink-scala swapped in, and name it in engineVersion.

How a scenario is deployed:

• The model is packed into nussknacker-model-<md5>.jar and uploaded to the VVP artifact storage only when VVP does not have it yet. Next to it, every deployment uploads nussknacker-deployment-<deploymentId>.json with the scenario, its version, the deployment data and the model configuration; its location is the only argument of the job.
• The VVP deployment is named after the scenario. A scenario name made only of letters, digits, _ and -, up to 128 characters, is used as is; otherwise every run of other characters becomes -, the name is cut to 119 characters and a hash of the original name is appended. The deployment carries the labels nussknackerScenarioId, nussknackerVersionId and nussknackerDeploymentId; a VVP deployment of the same name that belongs to another scenario is never taken over - the deployment fails.
• Its parallelism is the scenario's parallelism (default 1).
• First deployment: the VVP deployment is created and started without state.
• Redeploy of a scenario whose job is active: the job is stopped with a savepoint, Nussknacker waits until it has stopped (jobStopPolling), updates the deployment and starts it from that state (LATEST_STATE). When no job is active, the deployment is updated and started without state. Deploying from a chosen savepoint is not supported.
• Cancel stops the job without a savepoint. The VVP deployment is kept - Nussknacker only deletes it on cancel of a failed job with deleteFailedDeploymentOnCancel.
• Stop stops the job with a savepoint and returns its location; making a savepoint of a running scenario is supported too.

Statuses come from the newest job of the deployment: STARTING is shown as during deploy, RUNNING as running, CANCELLING as during cancel, CANCELLED as canceled, FINISHED as finished and FAILED as a problem with the failure message from VVP. A deployment deleted in the VVP console is reported as a problem until the scenario is canceled.

Other features:

• Scenario tests run on a Flink MiniCluster inside the Designer.
• Live data preview reads modelConfig.liveDataPreview.storage, which has to be DESIGNER_DB or DESIGNER_API when live data preview is enabled - without it the Designer refuses the configuration.
• Periodic scenarios are not supported.

Ververica targets created at runtime

The ververica Deployment Manager can be listed in dynamicTargets.deploymentManagers, so that targets can be created at runtime. deploymentMode.type of the entry decides the target type it declares: perJob declares Ververica per-job, session declares Ververica session cluster. To offer both, list ververica twice:

dynamicTargets {
  deploymentManagers: [
    {
      type: "ververica"
      baseUrl: "https://vvp.example.com"
      engineVersion: "nussknacker-4.5-jdk17-flink-1.20"
      deploymentMode { type: perJob }
    }
    {
      type: "ververica"
      baseUrl: "https://vvp.example.com"
      engineVersion: "nussknacker-4.5-jdk17-flink-1.20"
      deploymentMode { type: session }
    }
  ]
}

Parameters of a created target, named by the configuration key they set:

• Ververica per-job: deploymentMode.deploymentTargetName (required) and the four deploymentMode.jobResources.* keys;
• Ververica session cluster: deploymentMode.clusterName (required);
• both: baseUrl, engineVersion and auth.type (required), auth.token, auth.username, auth.password, auth.ssoId, namespace, workspace, requestTimeout, artifactUploadTimeout, tokenRefreshMargin, savepointPolling.* and jobStopPolling.*.

The values given in the entry are offered as the defaults. flinkConfiguration, modelConfigOverride, modelArtifact, deleteFailedDeploymentOnCancel, miniCluster and scenarioTesting are not parameters - they come from the entry only.