Flink projects

A Flink project is the home of one Flink application. Your CI system builds the application and adds each build to the project as a version; Streamtime runs a version as a job in a Job Workspace, and moves it forward or back when you choose.

Runtime: Java, Scala or Python. It decides what a version is: a JAR, or a .zip for Python.
Budget: the project books Flink Units on its Job Workspace, and its jobs run within them. See Capacity units.
Versions come one of two ways: uploaded to Streamtime, or kept in your own artifact store (an OCI registry such as GHCR, or a JFrog repository) and registered by digest.
HEAD is the version the project runs. Moving HEAD deploys that version, newer or older than the current one.
A CI key lets your pipeline add versions. It is shown once.

The API examples below share this setup:

HTTP Python TypeScript Go

export STREAMTIME_TOKEN=<your-access-token>
export API=https://api.example.com/organizations/<org>

import os
from streamtime_sdk.client import AuthenticatedClient

client = AuthenticatedClient(
    base_url="https://api.example.com",
    token=os.environ["STREAMTIME_TOKEN"],
)
ORG = "<org>"

Uses the generated client in streamtime-sdk/typescript/client.

import { client } from "./client/client.gen";

client.setConfig({
  baseUrl: "https://api.example.com",
  headers: { Authorization: `Bearer ${process.env.STREAMTIME_TOKEN}` },
});
const org = "<org>";

package main

import (
    "bytes"
    "crypto/rand"
    "encoding/hex"
    "encoding/json"
    "net/http"
    "os"
)

const api = "https://api.example.com/organizations/<org>"

// newKey returns a random idempotency key.
func newKey() string {
    b := make([]byte, 16)
    rand.Read(b)
    return hex.EncodeToString(b)
}

// call sends one request with the access token and a JSON body (or none).
// An optional idempotency key is sent as the Idempotency-Key header.
func call(method, path string, body any, idempotencyKey ...string) (*http.Response, error) {
    var buf bytes.Buffer
    if body != nil {
        if err := json.NewEncoder(&buf).Encode(body); err != nil {
            return nil, err
        }
    }
    req, err := http.NewRequest(method, api+path, &buf)
    if err != nil {
        return nil, err
    }
    req.Header.Set("Authorization", "Bearer "+os.Getenv("STREAMTIME_TOKEN"))
    req.Header.Set("Content-Type", "application/json")
    if len(idempotencyKey) > 0 {
        req.Header.Set("Idempotency-Key", idempotencyKey[0])
    }
    return http.DefaultClient.Do(req)
}

Create a project

UI API CLI MCP Terraform

In the sidebar, open Flink projects and create a project.
Enter Identifier, Name, Runtime and, optionally, Description.
Choose the Job Workspace the project runs in and its Flink Units.
Save. On the project’s CI/CD tab, create the CI key: the dialog Save your CI upload key shows it once. Copy it, tick I have stored this key somewhere safe, and close.

HTTP Python TypeScript Go

curl -X POST "$API/flink-projects/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" -H "Content-Type: application/json" \
  -d '{"identifier": "etl", "name": "ETL jobs", "runtime": "java",
       "cluster": "<job-workspace>", "flink_unit_budget": 2}'

# The CI key, shown once (rotating replaces it).
curl -X POST "$API/flink-projects/etl/rotate-key/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

from streamtime_sdk.api.flink_projects import flink_projects_create, flink_projects_rotate_key
from streamtime_sdk.models import FlinkProject

flink_projects_create.sync(
    ORG,
    client=client,
    body=FlinkProject.from_dict({
        "identifier": "etl", "name": "ETL jobs", "runtime": "java",
        "cluster": "<job-workspace>", "flink_unit_budget": 2,
    }),
)
key = flink_projects_rotate_key.sync(ORG, "etl", client=client)

import { flinkProjectsCreate, flinkProjectsRotateKey } from "./client";

await flinkProjectsCreate({
  path: { org_identifier: org },
  body: { identifier: "etl", name: "ETL jobs", runtime: "java",
          cluster: "<job-workspace>", flink_unit_budget: 2 },
});
const { data: key } = await flinkProjectsRotateKey({
  path: { org_identifier: org, identifier: "etl" },
});

resp, err := call("POST", "/flink-projects/", map[string]any{
    "identifier": "etl", "name": "ETL jobs", "runtime": "java",
    "cluster": "<job-workspace>", "flink_unit_budget": 2,
})
if err != nil {
    return err
}
defer resp.Body.Close()

streamtime flink-project create --identifier etl --name "ETL jobs" --runtime java \
  --cluster <job-workspace> --flink-units 2
streamtime flink-project rotate-key etl     # the CI key, shown once

Not available in MCP.

resource "streamtime_flink_project" "etl" {
  identifier        = "etl"
  name              = "ETL jobs"
  runtime           = "java"
  cluster           = streamtime_flink.shared.identifier # a Job Workspace
  flink_unit_budget = 2
}

Terraform doesn’t manage the CI key: create it on the project page or with streamtime flink-project rotate-key etl. See streamtime_flink_project.

Connect your own artifact store

Optional. Keep builds in your own registry instead of uploading them, and give Streamtime a read-only credential limited to this project’s artifacts.

UI API CLI MCP Terraform

On the project, open the store settings and choose the Store type (OCI registry or JFrog), its address, and the Sign-in: none, Username and Password, or an Access token. Add a CA certificate if the store doesn’t use a public certificate.

HTTP Python TypeScript Go

curl -X PUT "$API/flink-projects/etl/store/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" -H "Content-Type: application/json" \
  -d '{"store_type": "oras", "url": "https://ghcr.io",
       "authentication": "token", "token": "<read-only-token>"}'

store_type is oras (an OCI registry) or jfrog; authentication is none, basic (with username and password) or token.

from streamtime_sdk.api.flink_projects import flink_project_store_update
from streamtime_sdk.models import FlinkProjectStore

flink_project_store_update.sync(
    ORG, "etl", client=client,
    body=FlinkProjectStore.from_dict({
        "store_type": "oras", "url": "https://ghcr.io",
        "authentication": "token", "token": "<read-only-token>",
    }),
)

import { flinkProjectStoreUpdate } from "./client";

await flinkProjectStoreUpdate({
  path: { org_identifier: org, identifier: "etl" },
  body: { store_type: "oras", url: "https://ghcr.io",
          authentication: "token", token: "<read-only-token>" },
});

resp, err := call("PUT", "/flink-projects/etl/store/", map[string]any{
    "store_type": "oras", "url": "https://ghcr.io",
    "authentication": "token", "token": "<read-only-token>",
})
if err != nil {
    return err
}
defer resp.Body.Close()

export STREAMTIME_STORE_TOKEN=<read-only-token>   # rather than --token on the command line
streamtime flink-project store set etl --store oras --url https://ghcr.io
streamtime flink-project store show etl

For password sign-in use --username with STREAMTIME_STORE_PASSWORD; --ca-file adds a CA certificate.

Not available in MCP.

Set the store block on streamtime_flink_project: store_type, url, authentication and the credential, from a sensitive variable.

Add a version from CI

Your pipeline adds each build with the project’s CI key. Streamtime generates the pipeline for you: GitHub, GitLab, Bitbucket or a generic script, on the project’s CI/CD tab or with streamtime flink-project ci-template etl --provider github. streamtime flink-project init etl writes a local development setup.

UI API CLI MCP

On the project’s Versions tab, Upload Version adds a build by hand. The CI/CD tab shows the pipeline file to commit.

HTTP Python TypeScript Go

# A version kept in your own store: register it by digest.
curl -X POST "$API/flink-projects/etl/releases/" \
  -H "X-Streamtime-Key: $CI_KEY" -H "Content-Type: application/json" \
  -d '{"version": "1.4.0", "store_type": "oras",
       "location": "ghcr.io/acme/etl:1.4.0",
       "digest": "sha256:<64-hex-characters>"}'

To upload the build to Streamtime instead, POST the file to $API/flink-projects/etl/versions/. List versions with GET $API/flink-projects/etl/versions/.

from streamtime_sdk.api.flink_projects import flink_project_releases_register
from streamtime_sdk.models import FlinkReleaseRegister

flink_project_releases_register.sync(
    ORG, "etl", client=client,
    body=FlinkReleaseRegister.from_dict({
        "version": "1.4.0", "store_type": "oras",
        "location": "ghcr.io/acme/etl:1.4.0",
        "digest": "sha256:<64-hex-characters>",
    }),
)

import { flinkProjectReleasesRegister } from "./client";

await flinkProjectReleasesRegister({
  path: { org_identifier: org, identifier: "etl" },
  body: { version: "1.4.0", store_type: "oras",
          location: "ghcr.io/acme/etl:1.4.0", digest: "sha256:<64-hex-characters>" },
});

resp, err := call("POST", "/flink-projects/etl/releases/", map[string]any{
    "version": "1.4.0", "store_type": "oras",
    "location": "ghcr.io/acme/etl:1.4.0", "digest": "sha256:<64-hex-characters>",
})
if err != nil {
    return err
}
defer resp.Body.Close()

# Upload a build to Streamtime
streamtime flink-project upload etl ./target/etl.jar --version 1.4.0

# Or register one in your own store
streamtime flink-project release register etl --version 1.4.0 --store oras \
  --location ghcr.io/acme/etl:1.4.0 --digest sha256:<64-hex-characters>

streamtime flink-project versions etl

Not available in MCP.

Run a version

Moving the project’s HEAD to a version deploys it. The first time, name the job and its Job Workspace.

UI API CLI MCP

On the project’s Deployments tab, choose the Version; the first time also the Job Workspace and Job (and for Python the Python file to run). Choose the State handling: from a savepoint, the last state, or stateless. Moving to an older version often needs stateless, because newer state may not load in older code.

HTTP Python TypeScript Go

curl -X PUT "$API/flink-projects/etl/head/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" -H "Content-Type: application/json" \
  -d '{"version": "1.4.0", "target_type": "job", "target": "etl-main",
       "cluster": "<job-workspace>", "upgrade_mode": "savepoint"}'

Returns 202: the job rolls out in the background. Later moves need only version and upgrade_mode. See where it runs with GET $API/flink-projects/etl/deployments/.

from streamtime_sdk.api.flink_projects import flink_projects_head_update
from streamtime_sdk.models import FlinkProjectHeadRequest

flink_projects_head_update.sync(
    ORG, "etl", client=client,
    body=FlinkProjectHeadRequest.from_dict({
        "version": "1.4.0", "target_type": "job", "target": "etl-main",
        "cluster": "<job-workspace>", "upgrade_mode": "savepoint",
    }),
)

import { flinkProjectsHeadUpdate } from "./client";

await flinkProjectsHeadUpdate({
  path: { org_identifier: org, identifier: "etl" },
  body: { version: "1.4.0", target_type: "job", target: "etl-main",
          cluster: "<job-workspace>", upgrade_mode: "savepoint" },
});

resp, err := call("PUT", "/flink-projects/etl/head/", map[string]any{
    "version": "1.4.0", "target_type": "job", "target": "etl-main",
    "cluster": "<job-workspace>", "upgrade_mode": "savepoint",
})
if err != nil {
    return err
}
defer resp.Body.Close()

streamtime flink-project head set etl --version 1.4.0 \
  --job etl-main --session-cluster <job-workspace> --wait
streamtime flink-project head show etl
streamtime flink-project deployments etl

MCP reads the job’s status, metrics and logs: “Is the etl-main job healthy?” Deploying isn’t available in MCP.

Delete a project

Deleting removes the project, every version and its CI key. It is refused while a Flink deployment or job still runs one of its versions. streamtime flink-project delete etl, DELETE $API/flink-projects/etl/, or remove the streamtime_flink_project resource.