Migrating from Apache Kafka to StreamFlow

1# 1. Start StreamFlow (single node for testing)
2./scripts/start-streamflow.sh
3
4# 2. Create the same topic as in Kafka
5kafka-topics.sh --create --topic orders \
6  --partitions 16 --bootstrap-server localhost:9092
7
8# 3. Run the same benchmark as Kafka
9kafka-producer-perf-test.sh --topic orders \
10  --num-records 1000000 --record-size 256 \
11  --throughput -1 \
12  --producer-props bootstrap.servers=localhost:9092 \
13  batch.size=65536 linger.ms=5 acks=1
14
15kafka-consumer-perf-test.sh --topic orders \
16  --messages 1000000 --broker-list localhost:9092 --timeout 60000
17
18# 4. Compare results with Kafka

1// Dual-write producer (during migration)
2public class DualWriteProducer {
3    private final KafkaProducer<String, byte[]> kafkaProducer;   // old
4    private final StreamFlowProducer streamFlowProducer;          // new
5
6    public void send(String topic, String key, byte[] value) {
7        // Write to both
8        kafkaProducer.send(new ProducerRecord<>(topic, key, value));
9        streamFlowProducer.send(topic, key, value);
10    }
11}

1# Count events in each system
2kafka-run-class.sh kafka.tools.GetOffsetShell \
3  --broker-list kafka:9092 --topic orders
4# vs
5kafka-run-class.sh kafka.tools.GetOffsetShell \
6  --broker-list streamflow:9092 --topic orders

1# 1. Stop Kafka consumers
2# 2. Change bootstrap.servers in your app config
3sed -i 's/kafka-cluster:9092/streamflow-cluster:9092/' application.yml
4
5# 3. Redeploy apps
6kubectl rollout restart deployment/my-app
7
8# 4. Verify everything works
9curl http://streamflow:8080/metrics | grep events_published

1# Once StreamFlow is stable in production
2docker compose -f kafka-cluster.yml down
3# or
4kubectl delete statefulset kafka

1Properties props = new Properties();
2props.put("bootstrap.servers", "kafka-cluster:9092");
3props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
4props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
5KafkaProducer<String, byte[]> producer = new KafkaProducer<>(props);
6producer.send(new ProducerRecord<>("orders", "key-1", payload));

1StreamFlowClient client = StreamFlowClient.remote("streamflow-cluster:9092");
2StreamFlowProducer producer = client.producer();
3producer.send("orders", "key-1", payload);

1// ONLY CHANGE: the cluster address
2props.put("bootstrap.servers", "streamflow-cluster:9092");
3// The rest of the KafkaProducer code works as-is

1producer = Producer({'bootstrap.servers': 'kafka-cluster:9092'})

1producer = Producer({'bootstrap.servers': 'streamflow-cluster:9092'})
2# That's it. Same code, same lib.

1const kafka = new Kafka({ brokers: ['kafka-cluster:9092'] });

1const kafka = new Kafka({ brokers: ['streamflow-cluster:9092'] });

1writer := &kafka.Writer{Addr: kafka.TCP("kafka-cluster:9092"), Topic: "orders"}

1writer := &kafka.Writer{Addr: kafka.TCP("streamflow-cluster:9092"), Topic: "orders"}

1{
2  "name": "jdbc-source",
3  "config": {
4    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
5    "connection.url": "jdbc:postgresql://db:5432/shop",
6    "table.whitelist": "orders",
7    "mode": "incrementing",
8    "incrementing.column.name": "id",
9    "topic.prefix": "cdc-"
10  }
11}

1# connect-source.properties
2source.jdbc-orders.type=jdbc
3source.jdbc-orders.topic=cdc-orders
4source.jdbc-orders.connection.url=jdbc:postgresql://db:5432/shop
5source.jdbc-orders.table=orders
6source.jdbc-orders.mode=incrementing
7source.jdbc-orders.incrementing.column=id
8source.jdbc-orders.poll.interval.ms=1000

1StreamsBuilder builder = new StreamsBuilder();
2
3builder.stream("transactions")
4    .filter((key, value) -> value.getAmount() > 10000)
5    .mapValues(tx -> new FraudCheck(tx))
6    .to("fraud-alerts");
7
8KafkaStreams streams = new KafkaStreams(builder.build(), props);
9streams.start();

1CompiledGraph graph = StreamflowGraph.define("fraud-detection")
2    .start("filter")
3        .agent("amount-filter")
4        .prompt("Filter: keep only transactions with amount > 10000. " +
5               "Return the transaction unchanged or null to drop.")
6    .then("analyze")
7        .agent("fraud-analyzer")
8        .prompt("Analyze this high-value transaction for fraud indicators. " +
9               "Return {fraud: true/false, confidence: 0-1, reason: '...'}")
10    .end("alert")
11        .agent("alert-generator")
12        .prompt("Generate a fraud alert for the security team. " +
13               "Include transaction details and risk assessment.")
14    .compile();
15
16GraphRunner runner = new GraphRunner();
17runner.register(graph);