Write once.
Run embedded.
Scale remotely.

Name: StreamFlow
Author: StreamFlow

The same code runs embedded for tests and benchmarks, then connects to a remote StreamFlow cluster in production. Your business code stays unchanged, only client construction and transport-specific lifecycle calls differ.

client.producer()client.consumer("group")client.admin()

1. Quick Start

Start a local cluster, then send one event. Ten lines. That's it.

bash

1# Start a local StreamFlow cluster
2./scripts/start-streamflow.sh
3# StreamFlow is listening on localhost:9092

Java

1StreamFlowClient client = StreamFlowClient.remote("localhost:9092");
2
3try (var producer = client.producer()) {
4    producer.send(
5        "orders",
6        "42",
7        orderBytes
8    ).join();
9}

The event has been acknowledged by StreamFlow and is available to every consumer of the topic. Durability depends on the active storage and replication profile of the cluster.

2. Develop locally, deploy remotely

One API. Two transports. Zero rewrites.

Develop locally

StreamFlowClient.embedded(...)

Application
    ↓
StreamFlow SDK
    ↓
Engine (in-process, ~µs)

Tests, benchmarks, monoliths.

Deploy

StreamFlowClient.remote(...)

Application
    ↓
StreamFlow SDK
    ↓
TCP (Kafka wire)
    ↓
Cluster (~ms)

Microservices, multi-JVM, production.

The promise: your business code stays unchanged. Only client construction and a few transport-specific lifecycle calls (flush, commitSync, consumer groups, custom properties) differ between the two modes.

Same business logic, two transports

Java

1StreamFlowClient client =
2    environment.equals("prod")
3        ? StreamFlowClient.remote(System.getenv("STREAMFLOW_BOOTSTRAP"))
4        : StreamFlowClient.embedded(router, router, eventLog);
5
6OrderService service = new OrderService(client);
7service.submit("order-42", orderBytes);

3. Kafka Interoperability

Coming from Kafka? Every method you know is a one-to-one mapping.

Apache Kafka Client	StreamFlow SDK
new KafkaProducer<>(props)	client.producer()
producer.send(new ProducerRecord<>(...))	producer.send("topic", key, val)
new KafkaConsumer<>(props)	client.consumer("group")
consumer.subscribe(List.of("topic"))	consumer.subscribe("topic")
consumer.poll(Duration)	consumer.poll(Duration)
consumer.commitSync()	consumer.commitSync()
AdminClient.create(props)	client.admin()

StreamFlow implements the Kafka wire protocol. Any Kafka client, in Java, Python, Go or Node.js, can read what the SDK writes and vice versa.

4. Install

XML

1<properties>
2    <streamflow.version>0.1.0</streamflow.version>
3</properties>
4
5<dependency>
6    <groupId>com.streamflow</groupId>
7    <artifactId>streamflow-sdk</artifactId>
8    <version>${streamflow.version}</version>
9</dependency>

Remote also needs kafka-clients
Embedded also needs streamflow-core
Both are declared <optional> in the SDK, add only what you need.

5. Producing Events

Send one event

Java

1try (var producer = client.producer()) {
2    RecordMetadata meta = producer
3        .send("orders", "order-42", orderBytes)
4        .join();
5
6    System.out.printf("partition=%d offset=%d%n",
7        meta.partition(), meta.offset());
8}

Send with headers

Java

1producer.send(
2    "orders",
3    "order-42",
4    orderBytes,
5    Map.of(
6        "source", "mobile-app",
7        "region", "eu-west",
8        "trace-id", "abc-123"
9    )
10).join();

Advanced: batch loop

Java

1try (var producer = client.producer()) {
2    for (int i = 1; i <= 10_000; i++) {
3        producer.send("orders", "order-" + i, payload(i));
4    }
5    producer.flush();   // remote only, force network send
6}

6. Consuming Events

Receive one batch

Java

1try (var consumer = client.consumer("order-service")) {
2    consumer.subscribe("orders");
3
4    List<StreamFlowRecord> records =
5        consumer.poll(Duration.ofSeconds(1));
6
7    for (var r : records) {
8        process(r.key(), r.value());
9    }
10    consumer.commitSync();
11}

Advanced: continuous loop

Java

1var consumer = client.consumer("payment-processor");
2consumer.subscribe("payments");
3
4while (running) {
5    var records = consumer.poll(Duration.ofMillis(100));
6    for (var r : records) {
7        handle(r);
8    }
9    if (!records.isEmpty()) consumer.commitSync();
10}

7. Replay

Rewind a partition to any offset and read forward.

Java

1consumer.subscribe("orders");
2consumer.seek("orders", 0, 500L);
3
4var replayed = consumer.poll(Duration.ofSeconds(5));

8. Embedded Mode

StreamFlow runs in the same JVM as your application. Ideal for tests, benchmarks and monoliths.

Java

1StreamFlowApplication app = new StreamFlowApplication(
2    "streamflow.properties");
3app.start();
4
5StreamFlowClient client = StreamFlowClient.embedded(
6    app.getRouter(),
7    app.getRouter(),
8    app.getEventLog()
9);
10
11try (var admin = client.admin()) {
12    admin.createTopic("temperature-readings", 4);
13}
14
15try (var producer = client.producer()) {
16    producer.send("temperature-readings", "sensor-A", reading);
17    // no flush() needed, writes are synchronous
18}
19
20app.close();

Differences with remote: no flush(), no consumer group required, no commitSync() required. No network hop. Writes are synchronous, and execution is deterministic. Storage follows the embedded engine configuration (in-memory or local journal).

9. Remote Mode

Your application connects to a StreamFlow cluster over TCP using the Kafka wire protocol.

bash

1# Start a cluster (or point at an existing one)
2./scripts/start-streamflow.sh
3# Listening on port 9092

Java

1StreamFlowClient client = StreamFlowClient.remote(
2    "host1:9092,host2:9092");
3
4try (var producer = client.producer()) {
5    producer.send("orders", "42", orderBytes).join();
6    producer.flush();
7}

10. Patterns

Pattern 1: Mode-agnostic service

Business logic written once. Injected embedded in tests, remote in production.

Java

1public class OrderService {
2    private final StreamFlowProducer producer;
3    private final StreamFlowConsumer consumer;
4
5    public OrderService(StreamFlowClient client) {
6        this.producer = client.producer();
7        this.consumer = client.consumer("order-service");
8        this.consumer.subscribe("orders");
9    }
10
11    public CompletableFuture<RecordMetadata> submit(String id, byte[] data) {
12        return producer.send("orders", id, data);
13    }
14}
15
16// Test
17var testClient = StreamFlowClient.embedded(router, router, eventLog);
18new OrderService(testClient);
19
20// Production
21var prodClient = StreamFlowClient.remote("prod-cluster:9092");
22new OrderService(prodClient);

Pattern 2: Replay from a specific offset

Java

1var consumer = client.consumer("replay-group");
2consumer.subscribe("orders");
3consumer.seek("orders", 0, 500L);
4
5var records = consumer.poll(Duration.ofSeconds(5));
6consumer.commitSync();

11. API Reference

StreamFlowClient

Java

1StreamFlowClient.remote("host1:9092,host2:9092");
2StreamFlowClient.embedded(publisher, reader, eventLog);
3
4client.producer();                     // default config
5client.producer(props);                // custom config (remote only)
6client.consumer("group-id");
7client.consumer("group-id", props);
8client.admin();

StreamFlowProducer

Java

1CompletableFuture<RecordMetadata> f = producer.send("topic", "key", value);
2producer.send("topic", "key", value, Map.of("trace-id", "abc"));
3RecordMetadata meta = producer.send("topic", "key", value).join();
4producer.flush();
5producer.close();

StreamFlowConsumer

Java

1consumer.subscribe("orders");
2consumer.subscribe("inventory", "shipping");
3
4List<StreamFlowRecord> records = consumer.poll(Duration.ofMillis(100));
5
6for (StreamFlowRecord r : records) {
7    r.topic(); r.partition(); r.offset();
8    r.key(); r.value(); r.headers(); r.timestamp();
9}
10
11consumer.commitSync();
12consumer.seek("orders", 0, 1000L);
13long latest = consumer.latestOffset("orders", 0);
14long earliest = consumer.earliestOffset("orders", 0);

StreamFlowAdmin

Java

1try (StreamFlowAdmin admin = client.admin()) {
2    admin.createTopic("orders", 8);
3    boolean exists = admin.topicExists("orders");
4    TopicInfo info = admin.describeTopic("orders");
5    admin.deleteTopic("orders");
6}

Module dependencies

streamflow-sdk
├── streamflow-common (required)  , StreamEvent, EventPublisher, EventReader
├── streamflow-core   (optional)  , ShardRouter, EventLog (embedded only)
└── kafka-clients     (optional)  , KafkaProducer/Consumer (remote only)

Next Steps

📊 Tuning: performance profiles (high-throughput, low-latency, durability)
🔌 Source Connectors: build custom data sources with the Connector SDK
🧪 Examples: runnable benchmarks in examples/kafka-replication-demo/

Write once.Run embedded.Scale remotely.

On this page

1. Quick Start

2. Develop locally, deploy remotely

Same business logic, two transports

3. Kafka Interoperability

4. Install

5. Producing Events

Send one event

Send with headers

Advanced: batch loop

6. Consuming Events

Receive one batch

Advanced: continuous loop

7. Replay

8. Embedded Mode

9. Remote Mode

10. Patterns

Pattern 1: Mode-agnostic service

Pattern 2: Replay from a specific offset

11. API Reference

StreamFlowClient

StreamFlowProducer

StreamFlowConsumer

StreamFlowAdmin

Module dependencies

Next Steps

Write once.
Run embedded.
Scale remotely.