Apache Flink

Streamtime runs Flink on a Kubernetes fleet and keeps it monitored and up to date. Apache Flink runs in a Job Workspace (session mode): a shared cluster, sized up to Max Flink Units, that you submit jobs to. Each job claims Flink Units from it. Apache Flink does not offer a single-job cluster.

Confluent Manager for Flink keeps its single-job cluster (application mode): the cluster runs one job, from an application JAR or a registered Flink project version, and is sized in Flink Units.

Key Features

  • Open-source Apache Flink or Confluent Flink, chosen as the Provider.
  • Sized in Flink Units, with an Optimization Goal (Cost, Balanced or Performance). Streamtime derives the rest from the Flink Units: each TaskManager gets min(Flink Units, 4) slots and the cluster gets enough TaskManagers to cover them. A job runs as wide as the Flink Units it was given. These numbers are provisional. What a Flink Unit is, and what it takes from the fleet: Capacity units.
  • RocksDB state in every Job Workspace.
  • Checkpoints and savepoints on by default. On a kind or BYOK fleet they go to the fleet’s RustFS, in the flink bucket under <organization>/<cluster>, and the last 3 checkpoints are kept. A job that stops or restarts resumes from them. Deleting the cluster removes its folder. The fleet’s RustFS holds one shared credential for all clusters, so any cluster on the fleet can read another’s state. Cloud fleets have no durable store yet and keep state on the pods’ temporary disk, so Keep running state and Resume from the latest checkpoint are refused there.
  • State handling per job: Start fresh (stateless), Keep running state (savepoint) or Resume from the latest checkpoint (last-state).
  • Flink Web UI on every deployment, with metrics, logs and alerts.
  • Flink projects: CI registers each build from your own artifact store as a version, and deployments run it by version. See Flink projects.

Settings

Provider: Apache Flink or Confluent Manager for Flink.

Basic Configuration: Identifier, Description, and Max Flink Units (Job Workspace) or Flink Units (Confluent single-job cluster) with the Optimization Goal. For Apache Flink, Jobs run in: Java (the default) or Python. Python installs the image that runs PyFlink jobs and needs no project to exist first. A Python release is refused on a Java workspace, and the language cannot change once the workspace has jobs. Confluent Manager for Flink also asks for the State Backend Tier.

Fleet Selection: the fleet to run on, or let Streamtime choose one.

Advanced Configuration: the Version, to upgrade or for local testing. Confluent Manager for Flink also asks for the application JAR (or a project version), entry class and arguments, and its license. Task slots, TaskManager count, security, Web UI, savepoint and checkpoint settings are not asked for; the API still accepts them as overrides.

Before you start

  • A fleet in Healthy state with room for the Flink workload (see Fleet management).
  • For the API: an access token in STREAMTIME_TOKEN and your organization identifier in place of <org>. Replace api.example.com with your Streamtime API address.

The API examples below share this setup:

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)
}

Saving a Flink resource and provisioning it are separate steps in the API: create or change it, then reconcile its new generation. The reconcile returns 202 with the operation it started; the same Idempotency-Key returns that operation again, and a 409 means the generation is out of date.

  1. In the sidebar, select Flink, then Create Flink.
  2. Provider: choose Apache Flink or Confluent Manager for Flink.
  3. Basic Configuration: enter the Identifier, choose Jobs run in (Apache Flink), and set Max Flink Units and the Optimization Goal.
  4. Fleet Selection: choose the fleet.
  5. Advanced Configuration: change the Version only if you need to.
  6. Create it. It shows Pending, then Provisioning, then Healthy.
# 1. Save a session cluster. The response includes "generation".
curl -X POST "$API/flink/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "identifier": "analytics",
    "mode": "session",
    "provider": "open-source",
    "max_flink_units": 8,
    "advanced_configuration": {"optimization_goal": "Balanced", "state_tier": "Light"}
  }'

# 2. Provision it
curl -X POST "$API/flink/analytics/reconcile/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d '{"generation": 1}'

identifier and mode are required. Apache Flink takes "mode": "session" and max_flink_units; "mode": "application" is refused for it. For Confluent Manager for Flink ("provider": "cmf"), send "mode": "application", flink_units and either advanced_configuration.application_jar or project_version ({"project": "<project>", "version": "<version>"}). Sizing, state backend and checkpoint locations are Streamtime’s defaults; the rest of advanced_configuration is optional. Add "fleet": "<fleet>" to choose the fleet.

import uuid

from streamtime_sdk.api.flink import organizations_flink_reconcile_create
from streamtime_sdk.models import ReconcileRequest

resp = client.get_httpx_client().post(
    f"/organizations/{ORG}/flink/",
    json={
        "identifier": "analytics",
        "mode": "session",
        "provider": "open-source",
        "max_flink_units": 8,
        "advanced_configuration": {"optimization_goal": "Balanced", "state_tier": "Light"},
    },
)
resp.raise_for_status()

organizations_flink_reconcile_create.sync(
    ORG,
    "analytics",
    client=client,
    body=ReconcileRequest(generation=resp.json()["generation"]),
    idempotency_key=str(uuid.uuid4()),
)

import { organizationsFlinkReconcileCreate } from "./client";

const { data: saved } = await client.post({
  url: "/organizations/{org_identifier}/flink/",
  path: { org_identifier: org },
  body: {
    identifier: "analytics",
    mode: "session",
    provider: "open-source",
    max_flink_units: 8,
    advanced_configuration: { optimization_goal: "Balanced", state_tier: "Light" },
  },
});

await organizationsFlinkReconcileCreate({
  path: { org_identifier: org, identifier: "analytics" },
  headers: { "Idempotency-Key": crypto.randomUUID() },
  body: { generation: (saved as { generation: number }).generation },
});

resp, err := call("POST", "/flink/", map[string]any{
    "identifier":      "analytics",
    "mode":            "session",
    "provider":        "open-source",
    "max_flink_units": 8,
    "advanced_configuration": map[string]any{
        "optimization_goal": "Balanced", "state_tier": "Light",
    },
})
if err != nil {
    return err
}
defer resp.Body.Close()

var saved struct {
    Generation int `json:"generation"`
}
if err := json.NewDecoder(resp.Body).Decode(&saved); err != nil {
    return err
}
resp, err = call("POST", "/flink/analytics/reconcile/",
    map[string]any{"generation": saved.Generation}, newKey())

streamtime flink create \
  --identifier analytics \
  --mode session \
  --runtime python \
  --fleet <fleet>

--runtime is java (the default) or python, and applies to session mode only. The API field is advanced_configuration.runtime.

The CLI saves the deployment but can’t start provisioning it yet. Start it from the UI, or with the API’s reconcile request.

Not available in MCP yet. Create the deployment from the UI, API or CLI.

resource "streamtime_flink" "analytics" {
  identifier      = "analytics"
  mode            = "session"
  fleet           = "<fleet>"
  max_flink_units = 8

  advanced_configuration = jsonencode({})
}

terraform apply saves the deployment, starts provisioning it and waits until it is ready, the same two steps as the API. Leave out fleet to let Streamtime choose one. Apache Flink takes mode = "session" and max_flink_units. For Confluent Manager for Flink, set flink_provider = "cmf", mode = "application" and flink_units, and either the JAR in advanced_configuration or a project version:

resource "streamtime_flink" "shared" {
  identifier       = "shared"
  flink_provider   = "open-source"
  mode             = "session"
  fleet            = "<fleet>"
  max_flink_units  = 8
}

# A project runs in a Job Workspace and reserves Flink Units on it.
# The build tool follows the runtime (Java: Maven, Scala: sbt, Python: pip).
resource "streamtime_flink_project" "etl" {
  identifier        = "etl"
  name              = "ETL jobs"
  runtime           = "java"
  cluster           = streamtime_flink.shared.identifier
  flink_unit_budget = 2
}

resource "streamtime_flink" "etl" {
  identifier      = "etl"
  flink_provider  = "cmf"
  mode            = "application"
  fleet           = "<fleet>"
  flink_units = 2

  project_version = {
    project = streamtime_flink_project.etl.identifier
    version = "<version>"
  }
}

Terraform does not manage the project’s CI upload key: create it when you set up CI, on the project page or with streamtime flink-project rotate-key. To manage an existing deployment or project, import it by identifier: terraform import streamtime_flink.analytics analytics. See streamtime_flink and streamtime_flink_project.

Check status

status is one of Pending, Provisioning, Updating, Healthy, Error or Deleting.

  1. In the sidebar, select Flink. Each entry shows its status.

  2. Open it. Overview shows its status, settings and endpoints.

curl -s "$API/flink/analytics/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

# Collect a fresh status report (waits for the answer)
curl -s -X POST "$API/flink/analytics/status/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

from streamtime_sdk.api.flink import organizations_flink_retrieve

res = organizations_flink_retrieve.sync(ORG, "analytics", client=client)
print(res.status)

import { organizationsFlinkRetrieve } from "./client";

const { data } = await organizationsFlinkRetrieve({ path: { org_identifier: org, identifier: "analytics" } });
console.log(data?.status);

resp, err := call("GET", "/flink/analytics/", nil)
// Decode "status" from resp.Body.

streamtime flink list
streamtime flink get analytics

Not available in MCP yet. Check the status from the UI, API or CLI.

Submit and manage jobs (session mode)

A session cluster runs the jobs you submit to it. Each job claims Flink Units from the cluster’s Max Flink Units. Submitting a job starts it straight away.

  1. Open the session deployment from Flink and select Submit Job (Submit First Job when it has none yet).
  2. Enter the Job Identifier and its Flink Units, then the JAR or project version, parallelism and upgrade mode, and select Submit Flink Job.
  3. Open a job to see its status, change it or delete it.
# Submit a job
curl -X POST "$API/flink/analytics/jobs/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "identifier": "clickstream",
    "flink_units": 2,
    "advanced_configuration": {
      "application_jar": "<jar-url>",
      "parallelism": 2,
      "upgrade_mode": "savepoint"
    }
  }'

# List jobs, check one, delete one
curl -s "$API/flink/analytics/jobs/" -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl -s "$API/flink/analytics/jobs/clickstream/" -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl -X DELETE "$API/flink/analytics/jobs/clickstream/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

Change a running job with PATCH $API/flink/analytics/jobs/clickstream/ (for example a new flink_units or advanced_configuration).

from streamtime_sdk.api.flink import (
    organizations_flink_jobs_create,
    organizations_flink_jobs_destroy,
    organizations_flink_jobs_list,
)
from streamtime_sdk.models import FlinkJobCreateSchema

job = organizations_flink_jobs_create.sync(
    ORG,
    "analytics",
    client=client,
    body=FlinkJobCreateSchema.from_dict({
        "identifier": "clickstream",
        "flink_units": 2,
        "advanced_configuration": {
            "application_jar": "<jar-url>",
            "parallelism": 2,
            "upgrade_mode": "savepoint",
        },
    }),
)

jobs = organizations_flink_jobs_list.sync_detailed(ORG, "analytics", client=client)
print(jobs.content.decode())

organizations_flink_jobs_destroy.sync_detailed(ORG, "analytics", "clickstream", client=client)

import {
  organizationsFlinkJobsCreate,
  organizationsFlinkJobsDestroy,
  organizationsFlinkJobsList,
} from "./client";

await organizationsFlinkJobsCreate({
  path: { org_identifier: org, flink_identifier: "analytics" },
  body: {
    identifier: "clickstream",
    flink_units: 2,
    advanced_configuration: {
      application_jar: "<jar-url>",
      parallelism: 2,
      upgrade_mode: "savepoint",
    },
  },
});

const { data: jobs } = await organizationsFlinkJobsList({
  path: { org_identifier: org, flink_identifier: "analytics" },
});
console.log(jobs);

await organizationsFlinkJobsDestroy({
  path: { org_identifier: org, flink_identifier: "analytics", identifier: "clickstream" },
});

resp, err := call("POST", "/flink/analytics/jobs/", map[string]any{
    "identifier":  "clickstream",
    "flink_units": 2,
    "advanced_configuration": map[string]any{
        "application_jar": "<jar-url>",
        "parallelism":     2,
        "upgrade_mode":    "savepoint",
    },
})

resp, err = call("GET", "/flink/analytics/jobs/", nil)
resp, err = call("DELETE", "/flink/analytics/jobs/clickstream/", nil)

streamtime flink job create analytics \
  --identifier clickstream \
  --application-jar <jar-url> \
  --parallelism 2 \
  --upgrade-mode savepoint

streamtime flink job list analytics
streamtime flink job get analytics clickstream
streamtime flink job delete analytics clickstream --yes

Use --project-version PROJECT@VERSION instead of --application-jar to run a registered project version. Setting a job’s Flink Units, or changing a job, isn’t available in the CLI yet; jobs it creates get 1 Flink Unit.

Not available in MCP yet. Submit and manage jobs from the UI or API.

Change the configuration or upgrade

Change the Flink Units, or upgrade to a new Version (advanced_configuration.chart_config.main.version). With the savepoint upgrade mode, Streamtime takes a savepoint first and resumes the job from it.

  1. Open it from Flink and select Configuration (or Edit).
  2. Change the settings, then save. Streamtime applies the change, and the status shows Updating, then Healthy.
# 1. Save the change. The response includes the new "generation".
GEN=$(curl -s -X PATCH "$API/flink/analytics/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"max_flink_units": 12}' | jq .generation)

# 2. Apply it
curl -X POST "$API/flink/analytics/reconcile/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d "{\"generation\": $GEN}"

The SDK has no typed call for this update yet, so save it with the SDK’s HTTP client, then apply it:

import uuid

from streamtime_sdk.api.flink import organizations_flink_reconcile_create
from streamtime_sdk.models import ReconcileRequest

resp = client.get_httpx_client().patch(
    f"/organizations/{ORG}/flink/analytics/",
    json={"max_flink_units": 12},
)
resp.raise_for_status()

organizations_flink_reconcile_create.sync(
    ORG, "analytics",
    client=client,
    body=ReconcileRequest(generation=resp.json()["generation"]),
    idempotency_key=str(uuid.uuid4()),
)

import { organizationsFlinkReconcileCreate } from "./client";

const { data: saved } = await client.patch({
  url: "/organizations/{org_identifier}/flink/analytics/",
  path: { org_identifier: org },
  body: {"max_flink_units": 12},
});

await organizationsFlinkReconcileCreate({
  path: { org_identifier: org, identifier: "analytics" },
  headers: { "Idempotency-Key": crypto.randomUUID() },
  body: { generation: (saved as { generation: number }).generation },
});

resp, err := call("PATCH", "/flink/analytics/", json.RawMessage(`{"max_flink_units": 12}`))
if err != nil {
    return err
}
defer resp.Body.Close()

var saved struct {
    Generation int `json:"generation"`
}
if err := json.NewDecoder(resp.Body).Decode(&saved); err != nil {
    return err
}
resp, err = call("POST", "/flink/analytics/reconcile/",
    map[string]any{"generation": saved.Generation}, newKey())

Not available in the CLI yet. Change the configuration from the UI or API.

Not available in MCP yet. Change the configuration from the UI or API.

Change the attributes in the streamtime_flink block, for example max_flink_units, project_version or the settings in advanced_configuration, and run terraform apply. The plan shows an in-place update; Terraform saves the change, applies it and waits until the deployment runs it. Changing identifier, mode, flink_provider or fleet replaces the deployment.

View metrics, logs and alerts

Open it from Flink and select Monitoring for its metrics and logs, and Alert Rules (where shown) for the alerts that watch it. Alerts that fire also appear on the fleet’s Monitoring tab.

Metrics and logs are read through the fleet that runs it:

curl -s "$API/fleet/<fleet>/metrics/overview/" -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl -s "$API/fleet/<fleet>/alerts/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

resp = client.get_httpx_client().get(f"/organizations/{ORG}/fleet/<fleet>/alerts/")
print(resp.json())

const { data: alerts } = await client.get({
  url: "/organizations/{org_identifier}/fleet/{fleet_id}/alerts/",
  path: { org_identifier: org, fleet_id: "<fleet>" },
});
console.log(alerts);

resp, err := call("GET", "/fleet/<fleet>/alerts/", nil)

# The fleet's logs from the last hour
streamtime logs fleet <fleet> --since 60

Reading metrics or alerts isn’t available in the CLI yet.

Ask your assistant, for example: “Any alerts on the analytics Flink workload today?” or “Show its error logs from the last hour.”

It uses get_fleet_alerts, get_fleet_logs, list_fleet_metrics and get_fleet_metric on the fleet that runs it.

Delete

Deleting removes what Streamtime installed for it on the fleet.

  1. In the sidebar, select Flink.
  2. Open its menu in the list, select Delete, and confirm.

The status shows Deleting, then it disappears from the list.

curl -X DELETE "$API/flink/analytics/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

A draft that was never provisioned is removed at once; otherwise the teardown starts. Poll GET $API/flink/analytics/ until it returns 404.

from streamtime_sdk.api.flink import organizations_flink_destroy

organizations_flink_destroy.sync_detailed(ORG, "analytics", client=client)

import { organizationsFlinkDestroy } from "./client";

await organizationsFlinkDestroy({ path: { org_identifier: org, identifier: "analytics" } });

resp, err := call("DELETE", "/flink/analytics/", nil)

streamtime flink delete analytics --yes

Not available in MCP yet. Delete it from the UI, API or CLI.

Remove the streamtime_flink block and run terraform apply, or run terraform destroy -target=streamtime_flink.analytics. Terraform waits until Streamtime has removed the deployment.