Streams

streams is Skippr Cloud’s append-only partitioned log: produce records to topics, consume with consumer groups and offsets, retain data for a configurable period (max 7 days).

Status: Preview — streams protocol clients connect with SASL/OAUTHBEARER (Cloud JWT) to streams.eu-central-1.cloud.skippr.io:9092. Logical topic names only; broker SCRAM is never customer-visible.

Not a work queue. Competing-consumer jobs (visibility timeout, long poll, DLQ) belong to Queue.

Concepts

Term	Meaning
Topic	Named durable append log (logical name, e.g. orders)
Partition	Ordered unit inside a topic
Producer	Client that appends records (Produce)
Consumer group	Cooperative readers with shared committed offsets
Offset	Position of a record within a partition
Retention	How long records remain readable (default 24h, max 7 days)

Access

Bootstrap	streams.eu-central-1.cloud.skippr.io:9092 (TLS)
Protocol	streams protocol (Admin + Produce/Fetch + consumer groups)
Auth	SASL/OAUTHBEARER with a Cloud JWT (tenant_id from claims). You never receive broker SCRAM passwords.
Tenant	From JWT claims only
Optional HTTPS Admin	Through Gateway only (JSON). streams protocol produce/consume does not go through the gateway.

Connect streams protocol clients directly to the bootstrap address. Do not use api.cloud.skippr.io as a streams bootstrap.

Token sources

Environment	How to get the JWT
Dev	skippr user login → ~/.skippr/credentials.json access_token (set SKIPPR_AUTH_URL=https://auth.cloud.skippr.io)
CI	SKIPPR_API_KEY → POST /auth/api-key-exchange
Prod	Workload identity → Cloud JWT

Override with SKIPPR_ACCESS_TOKEN when your client’s OAUTHBEARER callback reads from the environment.

There is no skippr streams CLI in Preview — manage topics with your language’s streams protocol client library.

Operations (Preview)

Operation	Status	Notes
CreateTopics / DeleteTopics / Metadata	Supported	Self-service topic lifecycle; logical names only; default retention 24h
Produce / Fetch	Supported	Per-partition order; at-least-once
Consumer groups (FindCoordinator, JoinGroup, SyncGroup, Heartbeat, LeaveGroup, OffsetCommit, OffsetFetch)	Supported	Logical group ids
ListOffsets	Supported	Earliest / latest offsets
DescribeConfigs / AlterConfigs	Supported	Topic configs; retention.ms max 7 days
ListGroups / DescribeGroups	Supported	Tenant-scoped; logical group ids
CreatePartitions	Coming soon	Increase partition count
Transactions / exactly-once	Not in Preview	—

Limits

Limit	Value
Retention default	24 hours
Retention max	7 days
Max record size	1 MiB
Topic / group names	Logical only — do not send t. prefixes

Client snippets

Bootstrap and auth are the same for every language:

bootstrap.servers=streams.eu-central-1.cloud.skippr.io:9092
security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
# token = ~/.skippr/credentials.json access_token or SKIPPR_ACCESS_TOKEN

Use logical topic names (orders), never t.{tenant}.*.

Java

Properties props = new Properties();
props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
    "streams.eu-central-1.cloud.skippr.io:9092");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
props.put(SaslConfigs.SASL_MECHANISM, "OAUTHBEARER");
props.put(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL, ""); // unused — callback supplies JWT
props.put(SaslConfigs.SASL_JAAS_CONFIG,
    "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;");
props.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS,
    SkipprOauthBearerCallback.class.getName()); // reads SKIPPR_ACCESS_TOKEN or credentials.json

try (AdminClient admin = AdminClient.create(props)) {
    admin.createTopics(List.of(new NewTopic("orders", 1, (short) 1))).all().get();
}

Spring

@Bean
public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> cfg = new HashMap<>();
    cfg.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
        "streams.eu-central-1.cloud.skippr.io:9092");
    cfg.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
    cfg.put(SaslConfigs.SASL_MECHANISM, "OAUTHBEARER");
    cfg.put(SaslConfigs.SASL_JAAS_CONFIG,
        "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;");
    cfg.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS,
        SkipprOauthBearerCallback.class.getName());
    cfg.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    cfg.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return new DefaultKafkaProducerFactory<>(cfg);
}

Python

import os, json, pathlib
from confluent_kafka import Producer
from confluent_kafka.admin import AdminClient, NewTopic

def token():
    if t := os.environ.get("SKIPPR_ACCESS_TOKEN"):
        return t
    path = pathlib.Path.home() / ".skippr" / "credentials.json"
    return json.loads(path.read_text())["access_token"]

def oauth_cb(_config):
    return token(), 3600_000  # (token, lifetime_ms)

conf = {
    "bootstrap.servers": "streams.eu-central-1.cloud.skippr.io:9092",
    "security.protocol": "SASL_SSL",
    "sasl.mechanisms": "OAUTHBEARER",
    "oauth_cb": oauth_cb,
}
AdminClient(conf).create_topics([NewTopic("orders", num_partitions=1)])
p = Producer(conf)
p.produce("orders", b"hello")
p.flush()

Go

mechanism := kafka.OAuthBearerMechanism{
    TokenProvider: func(ctx context.Context) (kafka.OAuthBearerToken, error) {
        tok := os.Getenv("SKIPPR_ACCESS_TOKEN")
        // or read ~/.skippr/credentials.json
        return kafka.OAuthBearerToken{Token: tok}, nil
    },
}
dialer := &kafka.Dialer{
    TLS:           &tls.Config{},
    SASLMechanism: mechanism,
}
conn, err := dialer.DialContext(ctx, "tcp", "streams.eu-central-1.cloud.skippr.io:9092")
// CreateTopic / WriteMessages with logical name "orders"

Node

import { Kafka, logLevel } from "kafkajs";
import fs from "fs";
import os from "os";
import path from "path";

function accessToken() {
  if (process.env.SKIPPR_ACCESS_TOKEN) return process.env.SKIPPR_ACCESS_TOKEN;
  const creds = JSON.parse(
    fs.readFileSync(path.join(os.homedir(), ".skippr", "credentials.json"), "utf8"),
  );
  return creds.access_token;
}

const client = new Kafka({
  clientId: "skippr-demo",
  brokers: ["streams.eu-central-1.cloud.skippr.io:9092"],
  ssl: true,
  sasl: {
    mechanism: "oauthbearer",
    oauthBearerProvider: async () => ({ value: accessToken() }),
  },
  logLevel: logLevel.INFO,
});

const admin = client.admin();
await admin.connect();
await admin.createTopics({ topics: [{ topic: "orders", numPartitions: 1 }] });

Related

• Queue — SQS-shaped work queues
• Auth — JWT, API keys, refresh
• Events — bus rules that may target streams or queues
• Gateway — HTTPS/JSON only (not streams protocol wire)
• Roadmap