Manage Artifacts in Confluent Manager for Apache Flink

Starting with version 2.4.0, Confluent Manager for Apache Flink® (CMF) can store and manage binary artifacts, such as Flink job JAR files, in blob storage that you provide.

You upload an artifact to an environment through the REST API and reference it with a cmf:// URI in a FlinkApplication specification as the job JAR, or in Flink SQL statements, for example as a user-defined function or a custom connector. CMF resolves the reference to the artifact’s storage location when it deploys the application, and the Flink cluster fetches the file directly from storage.

Artifact management removes the need to build a custom Docker image or to define a sidecar download container for every change to your application JAR. For those packaging alternatives, see How to Package a Flink Job for Confluent Manager for Apache Flink.

Artifacts are scoped to an environment: an artifact name is unique within its environment, and access is controlled per environment with RBAC. Artifacts are versioned: uploading new content for an existing artifact creates a new version, and existing versions are never modified in place.

Note

Artifact management is an opt-in feature and is disabled by default. While the feature is disabled, the artifact endpoints return HTTP 404. Artifact operations are supported through the REST API and CMF UI; the Confluent CLI does not support artifact operations.

How it works

You configure one global storage location (the base path) for the CMF installation, for example an Amazon S3 bucket, a Google Cloud Storage bucket, or any other file system supported by Confluent Platform for Apache Flink. CMF stores each artifact version as one object under the base path:

{basePath}/environments/{environment}/{artifact-name}/{version}{extension}

For example, version 2 of an artifact named my-app.jar in environment env-1 with base path s3://my-bucket/cmf is stored at:

s3://my-bucket/cmf/environments/env-1/my-app.jar/2.jar

CMF keeps the artifact metadata such as the artifact name, labels, annotations, versions, sizes, and checksums in its own database and treats the database as the source of truth. Files that are modified or removed directly on the storage, bypassing CMF, are not detected, except that a removed file is reported with the FILE_MISSING status phase (see List and view artifacts).

Two components access the storage, each with its own credentials:

CMF uploads, downloads, and deletes files using the filesystem configuration you provide with cmf.artifacts.configuration.
The Flink clusters fetch referenced artifacts using their own filesystem configuration, the same way they access checkpoint or savepoint storage. The artifact configuration of CMF is not passed to the Flink clusters. For more information, see Storage access for the Flink cluster.

As a best practice, use the same storage for artifacts that you use for Flink checkpoints and savepoints, because Flink supports only a single set of credentials per file system.

Enable artifact management

Enable the feature in the cmf section of your Helm values file and set the storage base path. The following example configures an S3 bucket:

cmf:
  artifacts:
    enabled: true
    basePath: s3://my-bucket/cmf
    configuration:
      fs.s3a.endpoint: s3.us-west-2.amazonaws.com
      fs.s3a.access.key: ${AWS_ACCESS_KEY_ID}
      fs.s3a.secret.key: ${AWS_SECRET_ACCESS_KEY}

Apply the values with helm upgrade. When artifact management is enabled, basePath is required; CMF does not start without it.

The configuration map is forwarded to the Flink filesystem that CMF uses internally, so it accepts the standard Flink filesystem properties for your storage provider (fs.s3a.* for S3, fs.gs.* for Google Cloud Storage, fs.azure.* for Azure). The CMF image bundles the Flink filesystem plugins for the s3://, s3a://, gs://, and abfs:// schemes, so no custom image is required for these providers. For other schemes, see Filesystem plugins.

Do not put real credentials in the values file. Instead, supply them out-of-band, for example as a Kubernetes Secret injected with the extraEnv Helm value, and reference the resulting environment variables as ${ENV_VAR} placeholders as shown in the preceding example. For details, see Inject secrets with environment variables. Alternatives such as IAM roles for service accounts (IRSA), Workload Identity, or a Vault Agent sidecar work the same way.

Note

S3-compatible object stores, such as MinIO or Ceph, are supported with the bundled S3 plugin. Set fs.s3a.endpoint to your store’s endpoint and fs.s3a.path.style.access: "true".

Artifact names and versions

The artifact name identifies the artifact within its environment and appears in the storage path and in cmf:// references. An artifact name follows the same rules as any other CMF resource name (see Identifiers), with one addition: it can end with a single file extension, such as .jar or .py. A dot is allowed only in the extension, so a name like my-app-1.0.jar is rejected.

For example, my-app.jar, etl-pipeline and my-udf.jar are valid names. Use versions or labels instead of encoding version numbers into the name.

Note

An artifact that is referenced as a Flink SQL user-defined function must be named with a .jar extension.

Versions are managed by CMF: the first upload creates version 1, and each content update creates the next version. The version number, size, SHA-256 checksum, and storage path of each version are reported in the artifact status.

Upload an artifact

Upload an artifact with a multipart POST request to the artifacts endpoint of an environment. The request has two parts: artifact contains the resource definition as JSON or YAML, with a matching part media type, and file contains the binary content:

curl -X POST http://cmf:8080/cmf/api/v1/environments/env-1/artifacts \
  -F 'artifact={"apiVersion":"cmf.confluent.io/v1","kind":"Artifact","metadata":{"name":"my-app.jar","labels":{"team":"streaming"}},"spec":{}};type=application/json' \
  -F 'file=@target/my-app.jar'

Note the following:

metadata.name and an empty spec object are required. The name of the uploaded local file is ignored; the artifact is identified by metadata.name.
metadata.labels and metadata.annotations are optional string maps you can use to organize artifacts.
A successful upload returns HTTP 201 with the created artifact at version 1.
If an artifact with the same name already exists in the environment, the request fails with HTTP 409. To upload a new version of an existing artifact, use the update endpoint instead (see Update an artifact).
Uploads larger than the configured maximum size (default 100 MB, see Configuration reference) fail with HTTP 413.

CMF accepts any file type and does not inspect or validate the uploaded content.

List and view artifacts

List the artifacts of an environment. The list contains the latest version of each artifact:

curl http://cmf:8080/cmf/api/v1/environments/env-1/artifacts

Get a single artifact, either the latest or a specific version:

curl http://cmf:8080/cmf/api/v1/environments/env-1/artifacts/my-app.jar
curl "http://cmf:8080/cmf/api/v1/environments/env-1/artifacts/my-app.jar?version=2"

List all versions of an artifact, newest first:

curl http://cmf:8080/cmf/api/v1/environments/env-1/artifacts/my-app.jar/versions

The list endpoints support the page and size query parameters for pagination.

Example response for a single artifact:

{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Artifact",
  "metadata": {
    "name": "my-app.jar",
    "uid": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "creationTimestamp": "2026-06-01T09:00:00Z",
    "updateTimestamp": "2026-06-15T14:30:00Z",
    "labels": {
      "team": "streaming"
    }
  },
  "spec": {},
  "status": {
    "version": 2,
    "path": "s3://my-bucket/cmf/environments/env-1/my-app.jar/2.jar",
    "size": 52428800,
    "checksum": "sha256:9f2c081b6ee2ee76adbf2e33f39a3699fbee6d97a6b5c5f2a0c5a9d40a6a4c1e",
    "phase": "READY"
  }
}

The response includes the following status fields:

Field	Description
`status.version`	Version number, starting at 1. Incremented on each content update.
`status.path`	Full path of this version’s file in blob storage.
`status.size`	File size in bytes.
`status.checksum`	SHA-256 checksum of the content, for example `sha256:9f2c...`. The checksum is informational; CMF does not verify it when a Flink cluster fetches the file.
`status.phase`	`READY`, `UPLOADING`, or `FILE_MISSING`
`status.message`	Human-readable message, populated on error states such as `FILE_MISSING`.

UPLOADING is a transient phase while an upload is in progress. FILE_MISSING indicates that the version’s backing file was removed from the storage outside of CMF; the metadata is still listed, but downloading the version fails with HTTP 404. The FILE_MISSING check runs when you list artifacts or get a single artifact.

Update an artifact

Update an artifact with a multipart PUT request to the artifact’s endpoint. The artifact part is required; the file part is optional.

If a file part is provided, CMF stores the content as a new version and returns HTTP 201:

curl -X PUT http://cmf:8080/cmf/api/v1/environments/env-1/artifacts/my-app.jar \
  -F 'artifact={"apiVersion":"cmf.confluent.io/v1","kind":"Artifact","metadata":{"name":"my-app.jar"},"spec":{}};type=application/json' \
  -F 'file=@target/my-app.jar'

If the uploaded content is identical to the current latest version, no new version is created. The request still succeeds with HTTP 200, and the response status.message states that the content already exists as the latest version.

If the file part is omitted, only the metadata of the artifact is updated and the request returns HTTP 200. Labels and annotations are updated independently of each other, with the following semantics:

Omit the field to leave the existing values unchanged.
Send an empty object ("labels": {}) to clear all entries.
Send a non-empty map to replace all existing entries with the given map.

In the request body, metadata.name can be omitted; if present, it must match the artifact name in the URL path.

Note

Uploading a new version does not affect running applications. An application keeps using the version that was resolved when the application was created or last updated. To move an application to a newer version, update the application specification (see Reference artifacts in applications).

Download an artifact

Download the content of an artifact, either the latest or a specific version:

curl -OJ "http://cmf:8080/cmf/api/v1/environments/env-1/artifacts/my-app.jar/content?version=2"

CMF streams the file from the blob storage. The response is served as application/octet-stream with a Content-Disposition header that names the file {name}-v{version} with the artifact’s extension preserved, for example my-app-v2.jar. The -OJ curl options save the download under that name.

Delete an artifact

Delete a specific version of an artifact, or the entire artifact with all its versions:

# Delete one version
curl -X DELETE "http://cmf:8080/cmf/api/v1/environments/env-1/artifacts/my-app.jar?version=2"

# Delete the artifact with all versions (?version=all is the default)
curl -X DELETE http://cmf:8080/cmf/api/v1/environments/env-1/artifacts/my-app.jar

A successful delete returns HTTP 204. Deleting removes both the metadata records and the backing files from the blob storage. Deleting the last remaining version removes the artifact entirely.

Important

There is no deletion protection: CMF does not check whether an artifact is still referenced by a Flink cluster such as an application or compute pool. Running clusters are not interrupted, because they fetched the file at deployment time, but subsequent deployments that reference a deleted artifact or version fail.

An environment that still contains artifacts cannot be deleted. Delete all artifacts in the environment first.

Reference artifacts in applications

Reference an uploaded artifact in the spec.job.jarURI field of a FlinkApplication with a cmf:// URI:

cmf://{environment}/{artifact-name}[?version={N}]

If version is omitted, the reference resolves to the latest version at the time the application is created or updated. For example:

apiVersion: cmf.confluent.io/v1
kind: FlinkApplication
metadata:
  name: my-app
spec:
  image: confluentinc/cp-flink:1.19.1-cp2
  flinkVersion: v1_19
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: '1'
  serviceAccount: flink
  jobManager:
    resource:
      memory: 1024m
      cpu: 1
  taskManager:
    resource:
      memory: 1024m
      cpu: 1
  job:
    jarURI: cmf://env-1/my-app.jar
    state: running
    parallelism: 1
    upgradeMode: stateless

CMF validates the reference when the application is created or updated: the artifact and version must exist, the caller needs the Use permission on the artifact (see Access control), and artifact management must be enabled. When CMF deploys the application, it resolves the reference to the version’s blob storage path and passes that path to the Flink cluster, which fetches the file directly from storage.

cmf:// references require flinkVersion v1_19 or newer. Applications with flinkVersion: v1_18 cannot use cmf:// URIs, because native artifact fetching is only available in Flink 1.19 and later.

Version pinning

When you submit an application with a version-less reference, such as cmf://env-1/my-app.jar, CMF pins the reference to the newest version at submission time and stores the pinned form, for example cmf://env-1/my-app.jar?version=3. Retrieving the application returns the pinned URI. This makes deployments reproducible: suspending and resuming the application keeps using the pinned version, even if newer versions have been uploaded since.

To move an application to a newer version, update the application with either an explicit ?version={N} or a version-less reference, which re-pins to the newest version.

If an environment’s application defaults (spec.flinkApplicationDefaults) contain a cmf:// jarURI, the default must specify an explicit ?version={N}; version-less references are rejected in environment defaults.

Storage access for the Flink cluster

The Flink cluster fetches the resolved file from blob storage using its own filesystem configuration and plugins.

Warning

CMF does not pass its artifact storage credentials to the Flink cluster. Without its own access to the artifact storage, the cluster cannot fetch the referenced file and the deployment fails.

Configure the cluster’s access to the artifact storage the same way you configure access to checkpoint storage: enable the matching filesystem plugin and provide credentials in flinkConfiguration or through the pod environment. For example, for S3:

spec:
  flinkConfiguration:
    s3.access-key: ${AWS_ACCESS_KEY_ID}
    s3.secret-key: ${AWS_SECRET_ACCESS_KEY}
  podTemplate:
    spec:
      containers:
        - name: flink-main-container
          env:
            - name: ENABLE_BUILT_IN_PLUGINS
              value: "flink-s3-fs-hadoop-1.19.1-cp2.jar"

The ENABLE_BUILT_IN_PLUGINS variable is required: Flink images ship the filesystem plugins in a disabled state, and the variable activates the named plugin when the container starts. Make sure the plugin JAR name matches the Flink version of your image. For a complete example of configuring S3 filesystem access for Flink clusters, see How to Set Up Checkpointing to AWS S3 with Confluent Manager for Apache Flink. You can set these values once per environment through the environment’s application defaults. For more information, see Configure Environments in Confluent Manager for Apache Flink.

Cross-environment references

An artifact is scoped to the environment it was uploaded to, but an application or a Flink SQL statement can reference an artifact in a different environment by naming that environment in the cmf:// URI:

job:
  jarURI: cmf://shared/common-connectors.jar

This is intentional behavior that supports the shared-library use case: a team can publish common utility JARs, connectors, and other artifacts once in a central environment, then reuse them across environments instead of uploading a copy to every environment.

Cross-environment referencing is governed by RBAC on both environments. To reference the artifact, the calling principal needs:

The Use operation on the artifact, checked against the artifact’s own environment (shared in the preceding example).
The usual permissions to create or update the application in the target environment where the application is deployed.

For an example role binding that grants this cross-environment access, see Configure Access Control for Confluent Manager for Apache Flink.

Important

An environment alone is not an isolation boundary for artifacts because CMF allows cross-environment references. A principal that holds the required access on both sides can reference an artifact from another environment. If you treat an environment as a trust boundary, scope the FlinkArtifact Use operation accordingly with RBAC, and restrict direct access to the underlying storage, which bypasses CMF RBAC entirely. To learn more, see Access control.

Raw storage paths

Raw jarURI values, such as s3:// paths to storage that you manage yourself, continue to work as in previous releases. However, artifacts stored in the configured artifact storage location must be referenced through the cmf:// scheme: when artifact management is enabled, a raw jarURI that points into the artifact storage location is rejected with HTTP 400.

Access control

Artifact access is controlled through the FlinkArtifact RBAC resource type, which is scoped to the environment, like FlinkApplication. The following operations exist:

Action	Required operation on `FlinkArtifact`
Upload an artifact, upload a new version, update metadata	`Edit`
List artifacts, view artifact details and versions	`View`
Delete an artifact or a version	`Remove`
Download artifact content	`Use`
Reference an artifact in an application with a `cmf://` URI	`Use`, checked against the artifact’s environment

For the roles that grant these operations and example role bindings, see Configure Access Control for Confluent Manager for Apache Flink.

Important

The FlinkArtifact RBAC rules control access to artifacts through CMF only. The underlying blob storage has its own access controls, which are your responsibility to configure. Because CMF scopes artifacts by environment, configure the storage permissions to reflect the same environment boundaries. Otherwise, a principal with direct storage access can read or change artifacts across environments, bypassing CMF RBAC.

Filesystem plugins

CMF accesses the artifact storage through the Flink filesystem abstraction. The CMF image bundles the following filesystem plugins, which are loaded when artifact management is enabled:

Plugin ID	Storage	URI schemes
`s3-fs-hadoop`	Amazon S3 and S3-compatible stores	`s3://`, `s3a://`
`gs-fs-hadoop`	Google Cloud Storage	`gs://`
`azure-fs-hadoop`	Azure Blob Storage	`abfs://`

These plugins apply only to CMF’s own storage access. The Flink clusters load their own plugins, as described in Storage access for the Flink cluster.

Add or replace plugins

If you need a filesystem plugin that is not bundled, for example HDFS, or you want to use a newer or patched build of a bundled plugin, supply an additional plugin directory with cmf.artifacts.extraPluginsDir. The directory layout is one sub-directory per plugin, named with the plugin ID, containing the plugin’s JAR files:

/opt/cmf/extra-plugins/
  hdfs/
    flink-hadoop-fs-plugin.jar        # new plugin id: added to the bundled set
  s3-fs-hadoop/
    flink-s3-fs-hadoop-patched.jar    # bundled id: replaces the bundled S3 plugin

The merge rule is by plugin ID, that is, the sub-directory name. A sub-directory whose name matches a bundled plugin ID (s3-fs-hadoop, gs-fs-hadoop, azure-fs-hadoop) replaces that bundled plugin, and new IDs are added alongside the bundled set. In the preceding example, CMF loads your patched S3 plugin instead of the bundled one, adds HDFS support, and keeps the bundled Google Cloud Storage and Azure plugins.

To use an extra plugin directory:

Provide the directory contents to the CMF pod with the mountedVolumes Helm value. A PersistentVolumeClaim, ConfigMap, or Secret all work as the volume source:

mountedVolumes:
  volumes:
    - name: extra-plugins
      persistentVolumeClaim:
        claimName: cmf-extra-plugins
  volumeMounts:
    - name: extra-plugins
      mountPath: /opt/cmf/extra-plugins
      readOnly: true

Point cmf.artifacts.extraPluginsDir at the mounted directory:

cmf:
  artifacts:
    enabled: true
    basePath: s3://my-bucket/cmf
    extraPluginsDir: /opt/cmf/extra-plugins

Apply the values with helm upgrade. Plugin directories are scanned once at startup; CMF must be restarted to pick up plugin changes.

Warning

If extraPluginsDir is set but does not exist or cannot be scanned, CMF fails to start. CMF also fails to start if no loaded plugin registers the URI scheme of the configured basePath.

Configuration reference

Set the following properties under the cmf: block of the Helm values file. In log and error messages, CMF reports the property names in kebab-case, for example cmf.artifacts.base-path.

Property	Default	Description
`cmf.artifacts.enabled`	`false`	Enable artifact management.
`cmf.artifacts.basePath`	(unset)	Required when enabled. Base storage path for artifacts, for example `s3://my-bucket/cmf`. CMF refuses to start when the feature is enabled and no base path is set.
`cmf.artifacts.maxUploadSize`	`100MB`	Maximum upload size. Accepts a unit suffix (`B`, `KB`, `MB`, `GB`, `TB`); a bare number is interpreted as bytes. Larger uploads fail with HTTP 413.
`cmf.artifacts.configuration`	`{}`	Key/value map forwarded to the Flink filesystem configuration that CMF uses for its own storage access, for example `fs.s3a.endpoint`. Not passed to Flink clusters.
`cmf.artifacts.extraPluginsDir`	(unset)	Directory with additional filesystem plugins, layered on top of the bundled plugins by plugin ID. See Filesystem plugins.
`cmf.artifacts.cache.maxEntries`	`64`	Maximum number of artifact JAR files kept in the local cache that CMF uses when Flink SQL statements reference artifacts as user-defined functions. Least recently used entries are evicted beyond this limit.
`cmf.artifacts.cache.downloadTimeout`	`60s`	Timeout for a single artifact download into the local cache. Accepts a unit suffix (for example `s`, `m`).
`cmf.artifacts.cache.dir`	(temp dir)	Directory of the local artifact cache. Defaults to `cmf-udf-jar-cache` under the Java temporary directory.

Limitations

All environments share one global artifact storage configuration. In deployments that span more than one Kubernetes cluster, every cluster must be able to reach the same storage backend. Per-cluster or per-environment storage locations are not supported.
The checksum is informational. CMF does not verify file integrity when a Flink cluster fetches an artifact, and modifications made directly on the storage are not detected.
CMF does not track which applications reference an artifact, and there is no deletion protection for artifacts that are in use.
The artifact Use permission governs access to managed artifacts, but is not a hard security boundary: a user who can deploy applications can still point jarURI at any raw storage location outside the artifact storage that the Flink cluster can read. Restrict access at the storage layer to enforce isolation.

Important

CMF does not validate or scan uploaded artifact content, and Flink clusters execute the code that artifacts contain. Apply appropriate controls, such as code review and artifact scanning, before uploading, and restrict direct access to the underlying storage.

Manage Artifacts in Confluent Manager for Apache Flink

How it works

Enable artifact management

Artifact names and versions

Upload an artifact

List and view artifacts

Update an artifact

Download an artifact

Delete an artifact

Reference artifacts in applications

Version pinning

Storage access for the Flink cluster

Cross-environment references

Raw storage paths

Access control

Filesystem plugins

Add or replace plugins

Configuration reference

Limitations

Related content