SDK, Performance Tuning & Integration

1// No tuning needed, direct engine access
2StreamFlowClient client = StreamFlowClient.embedded(router, router, eventLog);
3StreamFlowProducer producer = client.producer();
4producer.send("orders", "key-1", payload);  // ~µs latency

1StreamFlowConsumer consumer = client.consumer();  // no group needed in embedded
2consumer.subscribe("orders");
3List<StreamFlowRecord> records = consumer.poll(Duration.ofMillis(100));

1// Tip 1: Use sendBatch for bulk operations (amortizes shard dispatch)
2List<StreamEvent> batch = buildEventBatch(1000);
3producer.sendBatch("orders", batch).join();  // 1 dispatch for 1000 events
4
5// Tip 2: Use FlatEventBatch directly for maximum throughput
6// (bypass SDK, call engine directly, 3.8M evt/s)
7FlatEventBatch flatBatch = new FlatEventBatch(1024, 4 * 1024 * 1024);
8for (Order order : orders) {
9    flatBatch.add(order.timestamp(), order.keyHash(), partition,
10                  order.payload(), 0, order.payload().length);
11}
12engine.ingestFlatBatch("orders", partition, flatBatch);
13// → 3.8M evt/s (bypasses StreamEvent object creation)

1// Default producer, sensible defaults built-in
2StreamFlowClient client = StreamFlowClient.remote("streamflow:9092");
3StreamFlowProducer producer = client.producer();

1Properties props = new Properties();
2props.put("batch.size", 524288);           // 512KB, fill bigger batches
3props.put("linger.ms", 10);               // wait 10ms to fill batch
4props.put("compression.type", "lz4");     // fastest compression
5props.put("max.in.flight.requests.per.connection", 10);  // more pipelining
6props.put("buffer.memory", 268435456);     // 256MB buffer
7
8StreamFlowProducer producer = client.producer(props);

1Properties props = new Properties();
2props.put("batch.size", 16384);            // 16KB, send quickly
3props.put("linger.ms", 0);                // no wait
4props.put("compression.type", "none");     // no CPU overhead
5
6StreamFlowProducer producer = client.producer(props);

1Properties props = new Properties();
2props.put("acks", "all");                  // wait for all replicas
3props.put("retries", 10);
4props.put("max.in.flight.requests.per.connection", 5);  // required for ordering
5
6StreamFlowProducer producer = client.producer(props);

1// Default consumer, auto-commit enabled
2StreamFlowConsumer consumer = client.consumer("my-group");

1Properties props = new Properties();
2props.put("fetch.min.bytes", 65536);       // 64KB minimum before returning
3props.put("fetch.max.wait.ms", 100);       // 100ms long-poll
4props.put("max.poll.records", 10000);      // large batches per poll
5props.put("max.partition.fetch.bytes", 4194304);  // 4MB per partition
6
7StreamFlowConsumer consumer = client.consumer("fast-group", props);

1Properties props = new Properties();
2props.put("enable.auto.commit", "false");  // manual commit
3props.put("auto.offset.reset", "earliest");
4
5StreamFlowConsumer consumer = client.consumer("processor-group", props);
6consumer.subscribe("orders");
7
8while (running) {
9    List<StreamFlowRecord> records = consumer.poll(Duration.ofMillis(100));
10    for (StreamFlowRecord r : records) {
11        processOrder(r);
12    }
13    consumer.commitSync();  // commit AFTER processing
14}

1import com.streamflow.sdk.resilience.*;
2
3// Custom resilience
4CircuitBreaker cb = new CircuitBreaker("my-producer",
5    5,          // open after 5 consecutive failures
6    10_000L     // stay open for 10 seconds
7);
8
9RetryPolicy retry = RetryPolicy.builder()
10    .maxRetries(5)
11    .initialDelayMs(100)
12    .maxDelayMs(5000)
13    .circuitBreaker(cb)
14    .build();
15
16StreamFlowProducer producer = new RemoteProducer(
17    "streamflow:9092", new Properties(), retry, cb);

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

1@Configuration
2public class StreamFlowConfig {
3
4    @Bean
5    public StreamFlowClient streamFlowClient(
6            @Value("${streamflow.servers}") String servers) {
7        return StreamFlowClient.remote(servers);
8    }
9
10    @Bean
11    public StreamFlowProducer producer(StreamFlowClient client) {
12        return client.producer();  // SDK defaults
13    }
14}
15
16@RestController
17@RequestMapping("/api/orders")
18public class OrderController {
19
20    private final StreamFlowProducer producer;
21
22    @PostMapping
23    public CompletableFuture<RecordMetadata> create(@RequestBody Order order) {
24        return producer.send("orders", order.getId(),
25            objectMapper.writeValueAsBytes(order));
26    }
27}

1@ApplicationScoped
2public class StreamFlowService {
3    private final StreamFlowProducer producer;
4
5    @Inject
6    public StreamFlowService(
7            @ConfigProperty(name = "streamflow.servers") String servers) {
8        StreamFlowClient client = StreamFlowClient.remote(servers);
9        this.producer = client.producer();
10    }
11
12    public Uni<RecordMetadata> publish(String topic, String key, byte[] value) {
13        return Uni.createFrom().completionStage(
14            producer.send(topic, key, value));
15    }
16}

1import { Kafka } from 'kafkajs';
2
3const kafka = new Kafka({
4  clientId: 'my-web-app',
5  brokers: ['streamflow:9092'],
6});
7
8const producer = kafka.producer();
9await producer.connect();
10await producer.send({
11  topic: 'orders',
12  messages: [{ key: 'order-123', value: JSON.stringify(order) }],
13});
14
15const consumer = kafka.consumer({ groupId: 'web-app' });
16await consumer.subscribe({ topic: 'orders', fromBeginning: true });
17await consumer.run({
18  eachMessage: async ({ message }) => {
19    console.log(`${message.key}: ${message.value}`);
20  },
21});

1from confluent_kafka import Producer, Consumer
2import json
3
4p = Producer({'bootstrap.servers': 'streamflow:9092'})
5p.produce('orders', key='order-123', value=json.dumps(order))
6p.flush()
7
8c = Consumer({
9    'bootstrap.servers': 'streamflow:9092',
10    'group.id': 'python-app',
11    'auto.offset.reset': 'earliest'
12})
13c.subscribe(['orders'])
14while True:
15    msg = c.poll(1.0)
16    if msg:
17        print(f'{msg.key()}: {msg.value()}')

1writer := &kafka.Writer{
2    Addr:  kafka.TCP("streamflow:9092"),
3    Topic: "orders",
4}
5writer.WriteMessages(ctx, kafka.Message{
6    Key: []byte("order-123"), Value: orderJSON,
7})
8
9reader := kafka.NewReader(kafka.ReaderConfig{
10    Brokers: []string{"streamflow:9092"},
11    Topic:   "orders",
12    GroupID: "go-app",
13})
14msg, _ := reader.ReadMessage(ctx)

1@RestController
2public class MobileGateway {
3    private final StreamFlowProducer producer;
4
5    @PostMapping("/api/v1/events/{topic}")
6    public CompletableFuture<Map<String, Object>> publish(
7            @PathVariable String topic,
8            @RequestBody byte[] payload,
9            @RequestHeader("X-Device-Id") String deviceId) {
10        return producer.send(topic, deviceId, payload)
11            .thenApply(m -> Map.of(
12                "partition", m.partition(),
13                "offset", m.offset()));
14    }
15
16    @GetMapping(value = "/api/v1/events/{topic}/stream",
17                produces = MediaType.TEXT_EVENT_STREAM_VALUE)
18    public Flux<StreamFlowRecord> stream(@PathVariable String topic) {
19        StreamFlowConsumer c = client.consumer("mobile-" + topic);
20        c.subscribe(topic);
21        return Flux.interval(Duration.ofMillis(100))
22            .flatMapIterable(tick -> c.poll(Duration.ofMillis(50)));
23    }
24}

1class StreamFlowClient {
2    let baseURL: URL
3
4    func publish(topic: String, key: String, value: Data) async throws -> EventMetadata {
5        var req = URLRequest(url: baseURL.appending(
6            path: "api/v1/events/\(topic)"))
7        req.httpMethod = "POST"
8        req.httpBody = value
9        req.setValue(
10            UIDevice.current.identifierForVendor?.uuidString,
11            forHTTPHeaderField: "X-Device-Id")
12        let (data, _) = try await URLSession.shared.data(for: req)
13        return try JSONDecoder().decode(EventMetadata.self, from: data)
14    }
15
16    func subscribe(topic: String) -> AsyncStream<Data> {
17        AsyncStream { cont in
18            let src = EventSource(url: baseURL.appending(
19                path: "api/v1/events/\(topic)/stream"))
20            src.onMessage = { cont.yield($0.data) }
21            cont.onTermination = { _ in src.disconnect() }
22        }
23    }
24}

1class StreamFlowClient(private val baseUrl: String) {
2    private val client = OkHttpClient()
3
4    suspend fun publish(
5        topic: String, key: String, value: ByteArray
6    ): EventMetadata = withContext(Dispatchers.IO) {
7        val req = Request.Builder()
8            .url("$baseUrl/api/v1/events/$topic")
9            .post(value.toRequestBody(
10                "application/octet-stream".toMediaType()))
11            .addHeader("X-Device-Id", Settings.Secure.ANDROID_ID)
12            .build()
13        client.newCall(req).execute().use { resp ->
14            Json.decodeFromString(resp.body!!.string())
15        }
16    }
17
18    fun subscribe(topic: String): Flow<String> = callbackFlow {
19        val src = EventSources.createFactory(client)
20            .newEventSource(
21                Request.Builder()
22                    .url("$baseUrl/api/v1/events/$topic/stream")
23                    .build(),
24                object : EventSourceListener() {
25                    override fun onEvent(
26                        es: EventSource, id: String?,
27                        type: String?, data: String
28                    ) { trySend(data) }
29                })
30        awaitClose { src.cancel() }
31    }
32}