Apache Flink

Streamtime runs Flink on a Kubernetes fleet and keeps it monitored and up to date. A Flink deployment runs in one of two modes:

• Application (a single-job cluster): the cluster runs one job, from an application JAR or a registered Flink project version, and is sized in Flink Units.
• Session: a shared cluster, sized up to Max Flink Units, that you submit jobs to. Each job claims Flink Units from it.

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) and a State Backend Tier (Stateless, Light, Heavy or Heavy-RocksDB).
• Savepoints and upgrade modes (stateless, savepoint or last-state) for safe job upgrades.
• Flink Web UI, optionally, and metrics, logs and alerts for every deployment.
• Flink projects: CI registers each build from your own artifact store as a version, and deployments run it by version. See Flink projects.

Settings

Basic Configuration: Identifier, Description, Provider, the mode, and Flink Units (application) or Max Flink Units (session).

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

Advanced Configuration

• For application mode: the application JAR location (or a project version), entry class or Python entry file, and arguments.
• Parallelism and upgrade mode; savepoint strategy.
• Optimization Goal and State Backend Tier.
• Web UI Enabled, Flink configuration (flink_conf), and JobManager and TaskManager resources.
• The chart Version, to upgrade.

---

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:

HTTPPythonTypeScriptGo

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.

---

Create a Flink deployment

UIAPICLIMCP

1. In the sidebar, select Flink, then Create Flink.
2. Basic Configuration: enter the Identifier, choose the Provider and the mode, and set Flink Units (application) or Max Flink Units (session).
3. Fleet Selection: choose the fleet.
4. Advanced Configuration: for application mode, the JAR or project version to run; then parallelism, upgrade mode, Optimization Goal, State Backend Tier and Web UI Enabled.
5. Create it. It shows Pending, then Provisioning, then Healthy.

HTTPPythonTypeScriptGo

# 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. For application mode, send flink_units and either advanced_configuration.application_jar or project_version ({"project": "<project>", "version": "<version>"}). 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 \
  --fleet <fleet>

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.

---

Check status

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

UIAPICLIMCP

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

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

HTTPPythonTypeScriptGo

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.

UIAPICLIMCP

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.

HTTPPythonTypeScriptGo

# 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, the advanced settings (parallelism, upgrade mode, resources, Flink configuration), 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.

UIAPICLIMCP

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.

HTTPPythonTypeScriptGo

# 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.

---

View metrics, logs and alerts

UIAPICLIMCP

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.

HTTPPythonTypeScriptGo

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.

UIAPICLIMCP

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.

HTTPPythonTypeScriptGo

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.