StreamFlow SDK Guide

1<dependency>
2    <groupId>com.streamflow</groupId>
3    <artifactId>streamflow-sdk</artifactId>
4    <version>0.1.0-SNAPSHOT</version>
5</dependency>

1# On your server or locally
2./scripts/start-streamflow.sh
3# StreamFlow is now listening on port 9092 (Kafka wire protocol)

1<!-- pom.xml -->
2<dependencies>
3    <dependency>
4        <groupId>com.streamflow</groupId>
5        <artifactId>streamflow-sdk</artifactId>
6        <version>0.1.0-SNAPSHOT</version>
7    </dependency>
8    <!-- Required for remote mode -->
9    <dependency>
10        <groupId>org.apache.kafka</groupId>
11        <artifactId>kafka-clients</artifactId>
12        <version>3.7.0</version>
13    </dependency>
14</dependencies>

1import com.streamflow.sdk.StreamFlowClient;
2import com.streamflow.sdk.StreamFlowProducer;
3import com.streamflow.sdk.RecordMetadata;
4import java.nio.charset.StandardCharsets;
5
6public class ProducerExample {
7    public static void main(String[] args) throws Exception {
8        // 1. Connect to StreamFlow (same as connecting to Kafka)
9        StreamFlowClient client = StreamFlowClient.remote("localhost:9092");
10
11        // 2. Create a producer
12        try (StreamFlowProducer producer = client.producer()) {
13
14            // 3. Send 10 events to topic "user-signups"
15            for (int i = 1; i <= 10; i++) {
16                String key = "user-" + i;
17                String json = """
18                    {"userId": %d, "name": "User %d", "email": "user%d@example.com"}
19                    """.formatted(i, i, i);
20                byte[] value = json.getBytes(StandardCharsets.UTF_8);
21
22                RecordMetadata meta = producer.send("user-signups", key, value).join();
23
24                System.out.printf("Sent user-%d → partition %d, offset %d%n",
25                    i, meta.partition(), meta.offset());
26            }
27
28            // 4. Ensure all buffered events are sent
29            producer.flush();
30        }
31
32        System.out.println("Done! 10 events published to 'user-signups'");
33    }
34}

1Sent user-1 → partition 3, offset 0
2Sent user-2 → partition 7, offset 0
3Sent user-3 → partition 12, offset 0
4...
5Done! 10 events published to 'user-signups'

1import com.streamflow.sdk.StreamFlowClient;
2import com.streamflow.sdk.StreamFlowConsumer;
3import com.streamflow.sdk.StreamFlowRecord;
4import java.time.Duration;
5import java.util.List;
6
7public class ConsumerExample {
8    public static void main(String[] args) {
9        // 1. Connect
10        StreamFlowClient client = StreamFlowClient.remote("localhost:9092");
11
12        // 2. Create a consumer with a group ID
13        try (StreamFlowConsumer consumer = client.consumer("signup-processor")) {
14
15            // 3. Subscribe to the topic
16            consumer.subscribe("user-signups");
17
18            // 4. Poll for events (blocks up to 5 seconds)
19            System.out.println("Waiting for events...");
20            List<StreamFlowRecord> records = consumer.poll(Duration.ofSeconds(5));
21
22            // 5. Process each event
23            for (StreamFlowRecord record : records) {
24                System.out.printf("Received: key=%s partition=%d offset=%d%n",
25                    record.key(), record.partition(), record.offset());
26                System.out.printf("  Value: %s%n", new String(record.value()));
27            }
28
29            System.out.printf("Total: %d events received%n", records.size());
30
31            // 6. Commit offsets
32            consumer.commitSync();
33        }
34    }
35}

1Waiting for events...
2Received: key=user-1 partition=3 offset=0
3  Value: {"userId": 1, "name": "User 1", "email": "user1@example.com"}
4Received: key=user-2 partition=7 offset=0
5  Value: {"userId": 2, "name": "User 2", "email": "user2@example.com"}
6...
7Total: 10 events received

1import java.util.Map;
2
3// Headers are key-value metadata attached to each event
4// Useful for routing, tracing, filtering
5RecordMetadata meta = producer.send(
6    "user-signups",
7    "user-42",
8    """{"userId": 42, "name": "Alice"}""".getBytes(),
9    Map.of(
10        "source", "mobile-app",
11        "region", "eu-west",
12        "trace-id", "abc-123"
13    )
14).join();

1import com.streamflow.sdk.StreamFlowClient;
2import com.streamflow.sdk.StreamFlowProducer;
3import com.streamflow.sdk.StreamFlowConsumer;
4import com.streamflow.sdk.StreamFlowRecord;
5import com.streamflow.sdk.RecordMetadata;
6import com.streamflow.app.StreamFlowApplication;
7import java.nio.charset.StandardCharsets;
8import java.time.Duration;
9import java.util.List;
10
11public class EmbeddedExample {
12    public static void main(String[] args) throws Exception {
13        // 1. Start StreamFlow engine in this JVM
14        StreamFlowApplication app = new StreamFlowApplication(
15            "streamflow-app/src/main/resources/streamflow.properties");
16        app.start();
17
18        // 2. Create embedded client (direct engine access, no network)
19        StreamFlowClient client = StreamFlowClient.embedded(
20            app.getRouter(),      // produces events
21            app.getRouter(),      // reads events
22            app.getEventLog()     // topic management
23        );
24
25        // 3. Create a topic
26        try (var admin = client.admin()) {
27            admin.createTopic("temperature-readings", 4);
28            System.out.println("Topic 'temperature-readings' created with 4 partitions");
29        }
30
31        // 4. Produce sensor readings
32        try (StreamFlowProducer producer = client.producer()) {
33            String[] sensors = {"sensor-A", "sensor-B", "sensor-C"};
34            for (int i = 0; i < 100; i++) {
35                String sensorId = sensors[i % sensors.length];
36                double temperature = 20.0 + Math.random() * 10.0;
37                String json = """
38                    {"sensorId": "%s", "temperature": %.1f, "timestamp": %d}
39                    """.formatted(sensorId, temperature, System.currentTimeMillis());
40
41                producer.send("temperature-readings", sensorId,
42                    json.getBytes(StandardCharsets.UTF_8));
43            }
44            // No flush() needed in embedded, writes are synchronous
45        }
46        System.out.println("Produced 100 temperature readings");
47
48        // 5. Consume all readings
49        try (StreamFlowConsumer consumer = client.consumer()) {
50            consumer.subscribe("temperature-readings");
51            List<StreamFlowRecord> records = consumer.poll(Duration.ofSeconds(1));
52            System.out.printf("Consumed %d readings%n", records.size());
53
54            records.stream().skip(Math.max(0, records.size() - 3)).forEach(r ->
55                System.out.printf("  %s: %s%n", r.key(), new String(r.value()))
56            );
57        }
58
59        // 6. Cleanup
60        app.close();
61    }
62}

1Topic 'temperature-readings' created with 4 partitions
2Produced 100 temperature readings
3Consumed 100 readings
4  sensor-A: {"sensorId": "sensor-A", "temperature": 27.3, "timestamp": 1711929600000}
5  sensor-B: {"sensorId": "sensor-B", "temperature": 22.1, "timestamp": 1711929600001}
6  sensor-C: {"sensorId": "sensor-C", "temperature": 25.8, "timestamp": 1711929600002}

1// Step 1: Produce with StreamFlow SDK
2StreamFlowClient sfClient = StreamFlowClient.remote("localhost:9092");
3try (StreamFlowProducer producer = sfClient.producer()) {
4    producer.send("shared-topic", "key-1",
5        "Hello from StreamFlow SDK".getBytes()).join();
6}
7
8// Step 2: Consume with standard Apache Kafka client
9import org.apache.kafka.clients.consumer.KafkaConsumer;
10import org.apache.kafka.clients.consumer.ConsumerRecords;
11import org.apache.kafka.clients.consumer.ConsumerRecord;
12
13Properties kafkaProps = new Properties();
14kafkaProps.put("bootstrap.servers", "localhost:9092");
15kafkaProps.put("group.id", "kafka-reader");
16kafkaProps.put("key.deserializer",
17    "org.apache.kafka.common.serialization.StringDeserializer");
18kafkaProps.put("value.deserializer",
19    "org.apache.kafka.common.serialization.ByteArrayDeserializer");
20kafkaProps.put("auto.offset.reset", "earliest");
21
22try (KafkaConsumer<String, byte[]> kafkaConsumer = new KafkaConsumer<>(kafkaProps)) {
23    kafkaConsumer.subscribe(List.of("shared-topic"));
24    ConsumerRecords<String, byte[]> records =
25        kafkaConsumer.poll(Duration.ofSeconds(5));
26    for (ConsumerRecord<String, byte[]> record : records) {
27        System.out.printf("Kafka client read: key=%s value=%s%n",
28            record.key(), new String(record.value()));
29        // Output: Kafka client read: key=key-1 value=Hello from StreamFlow SDK
30    }
31}

1// Connect to remote cluster
2StreamFlowClient client = StreamFlowClient.remote("host1:9092,host2:9092");
3
4// Connect to embedded engine
5StreamFlowClient client = StreamFlowClient.embedded(publisher, reader, eventLog);
6
7// Create producer/consumer/admin
8StreamFlowProducer producer = client.producer();           // default config
9StreamFlowProducer producer = client.producer(props);      // custom config (remote only)
10StreamFlowConsumer consumer = client.consumer("group-id"); // named group
11StreamFlowConsumer consumer = client.consumer("group", props);  // custom config
12StreamFlowAdmin admin = client.admin();

1// Send one event (async, returns a Future)
2CompletableFuture<RecordMetadata> future = producer.send("topic", "key", value);
3
4// Send with headers
5producer.send("topic", "key", value, Map.of("trace-id", "abc"));
6
7// Send and wait for confirmation
8RecordMetadata meta = producer.send("topic", "key", value).join();
9
10// Flush buffered events (remote mode, forces network send)
11producer.flush();
12
13// Always close when done
14producer.close();

1// Subscribe to topics
2consumer.subscribe("orders");
3consumer.subscribe("inventory", "shipping");  // multiple topics
4
5// Poll for events (blocks up to the specified duration)
6List<StreamFlowRecord> records = consumer.poll(Duration.ofMillis(100));
7
8// Each record contains:
9for (StreamFlowRecord r : records) {
10    r.topic();      // "orders"
11    r.partition();   // 3
12    r.offset();      // 42
13    r.key();         // "order-123"
14    r.value();       // byte[], your data
15    r.headers();     // Map<String, String>
16    r.timestamp();   // epoch millis
17}
18
19// Commit offsets (marks events as processed)
20consumer.commitSync();
21
22// Seek to a specific offset (replay)
23consumer.seek("orders", 0, 1000L);
24
25// Query offsets
26long latest = consumer.latestOffset("orders", 0);
27long earliest = consumer.earliestOffset("orders", 0);

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

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> submitOrder(
12            String orderId, byte[] orderData) {
13        return producer.send("orders", orderId, orderData);
14    }
15
16    public List<StreamFlowRecord> pollOrders() {
17        return consumer.poll(Duration.ofMillis(100));
18    }
19}
20
21// Production: connects over network
22StreamFlowClient prodClient = StreamFlowClient.remote("prod-cluster:9092");
23OrderService prodService = new OrderService(prodClient);
24
25// Test: same code, no network, instant
26StreamFlowClient testClient = StreamFlowClient.embedded(router, router, eventLog);
27OrderService testService = new OrderService(testClient);

1StreamFlowClient client = StreamFlowClient.remote("localhost:9092");
2StreamFlowConsumer consumer = client.consumer("payment-processor");
3consumer.subscribe("payments");
4
5while (true) {
6    List<StreamFlowRecord> records = consumer.poll(Duration.ofMillis(100));
7
8    for (StreamFlowRecord record : records) {
9        String paymentJson = new String(record.value());
10        System.out.printf("[partition=%d offset=%d] %s: %s%n",
11            record.partition(), record.offset(),
12            record.key(), paymentJson);
13
14        // Process the payment...
15    }
16
17    if (!records.isEmpty()) {
18        consumer.commitSync();  // commit after processing batch
19    }
20}

1StreamFlowConsumer consumer = client.consumer("replay-group");
2consumer.subscribe("orders");
3
4// Go back to offset 500 on partition 0
5consumer.seek("orders", 0, 500L);
6
7// Read from there
8List<StreamFlowRecord> records = consumer.poll(Duration.ofSeconds(5));
9System.out.printf("Replayed %d events starting from offset 500%n",
10    records.size());
11consumer.commitSync();