Replacing RocksDB with ScyllaDB in Kafka Streams by Almog Gavra

Learn how Responsive replaced embedded RocksDB with ScyllaDB in Kafka Streams, simplifying the architecture and unlocking massive availability and scale. The talk covers unbundling stream processors, key ScyllaDB features tested, and lessons learned from the transition.

1. A ScyllaDB Community
Replacing RocksDB with
ScyllaDB in Kafka Streams
Almog Gavra
Co-Founder, Responsive

2. Realtime Search Ingest @
Stream Processing @
Co-Founder @
Almog Gavra

3. About Kafka Streams
Why Introduce ScyllaDB?
Architecture Deep Dive
Lessons Learned & Practical Tips
Agenda

4. About Kafka Streams

5. Kafka
A storage backend optimized for storing
ordered “event” data

6. Kafka
Kafka Streams
A storage backend optimized for storing
ordered “event” data
A library for building event-driven
applications: realtime, responsive & stateful

7. Kafka Streams is Everywhere
Realtime Inference
Walmart uses Kafka Streams
to power fraud detection and
purchase recommendations

8. Kafka Streams is Everywhere
Realtime Inference
Walmart uses Kafka Streams
to power fraud detection and
purchase recommendations
Logistics
Michelin uses Kafka Streams to
handle their tier distribution, ensuring
delivery is tracked in realtime

9. Kafka Streams is Everywhere
Realtime Inference
Walmart uses Kafka Streams
to power fraud detection and
purchase recommendations
Logistics
Michelin uses Kafka Streams to
handle their tier distribution, ensuring
delivery is tracked in realtime
Liquidity Management
Michelin uses Kafka Streams to
handle their tier distribution, ensuring
delivery is tracked in realtime

10. Why Introduce ScyllaDB?

11. Original Design Goals
Just a Library
Kafka Streams should be easy
to integrate with existing apps.

12. Original Design Goals
Just a Library
Kafka Streams should be easy
to integrate with existing apps.
Complete API
All stream processing use cases
should be possible to write using
Kafka Streams.

13. Original Design Goals
Just a Library
Kafka Streams should be easy
to integrate with existing apps.
Complete API
All stream processing use cases
should be possible to write using
Kafka Streams.
Depend only on Kafka
There should be no dependencies on
external systems (such as HDFS or
YARN).

14. Original Design Goals
Depend only on Kafka
There should be no dependencies on
external systems (such as HDFS or
YARN).

15. Original Design Goals
Availability
Source of truth is in a “changelog” topic in Kafka. If
assignment changes, state must be restored.

16. Original Design Goals
Availability
Source of truth is in a “changelog” topic in Kafka. If
assignment changes, state must be restored.
Flexibility
Dynamic scaling is impractical. Most deployments
are provisioned for peak throughput.

17. Original Design Goals
Availability
Source of truth is in a “changelog” topic in Kafka. If
assignment changes, state must be restored.
Flexibility
Dynamic scaling is impractical. Most deployments
are provisioned for peak throughput.
Performance
Difficult to properly attribute resources to the
stream processing vs. RocksDB storage subsystem

18. What Changed?
2016
Kafka Streams is released with
Apache Kafka 0.10

19. What Changed?
2016
Kafka Streams is released with
Apache Kafka 0.10
2017
Cloud databases gain
mainstream adoption

20. What Changed?
2016
Kafka Streams is released with
Apache Kafka 0.10
2017
Cloud databases gain
mainstream adoption
2019
Kubernetes wins out as de
facto orchestration system

21. What Changed?
2016
Kafka Streams is released with
Apache Kafka 0.10
2017
Cloud databases gain
mainstream adoption
2019
Kubernetes wins out as de
facto orchestration system
2020
Kafka Streams popularity soars,
widening possible use cases

22. What Changed?
2016
Kafka Streams is released with
Apache Kafka 0.10
2017
Cloud databases gain
mainstream adoption
2019
Kubernetes wins out as de
facto orchestration system
2020
Kafka Streams popularity soars,
widening possible use cases
Today
Some assumptions guiding the original
Kafka Streams design are outdated

23. Lifting the “Kafka Only Requirement”

24. Lifting the “Kafka Only Requirement”

25. Lifting the “Kafka Only Requirement”

26. Deep Dive: Metronome
Metronome powers billing for companies like
OpenAI, NVIDIA and Databricks using Kafka
Streams.

Mission Critical Feature: Realtime Spend Limits

27. Key Results Migrating
from RocksDB to Scylla
Availability
Going from regular incidents to “not thinking about
Kafka Streams”
Throughput Growth
ScyllaDB scaled without hiccup as their data size
and throughput scaled
Scale Potential
Decoupled compute and storage means we’ve been
able to scale number of Kafka Partitions and
ScyllaDB cluster size independently
99.99%
3x
∞

28. Architecture Deep Dive

29. Data Model
CREATE TABLE key_value
partitionKey INTEGER,
dataKey BLOB,
dataValue BLOB,
epoch BIGINT,
offset BIGINT,
PRIMARY KEY ((partitionKey),
dataKey);

Raw Data
Since Kafka Streams deals only with serialized
bytes (users supply the serializers) it makes storing
data easy and general purpose

30. Data Model
CREATE TABLE key_value
partitionKey INTEGER,
dataKey BLOB,
dataValue BLOB,
epoch BIGINT,
offset BIGINT,
PRIMARY KEY ((partitionKey),
dataKey);

Primary Key
Using the Kafka partition allows us to implement
LWTs per partition and the data key allows us to
implement efficient lookups and range scans

31. Data Model
CREATE TABLE key_value
partitionKey INTEGER,
dataKey BLOB,
dataValue BLOB,
epoch BIGINT,
offset BIGINT,
PRIMARY KEY ((partitionKey),
dataKey);

Restoring State
Storing the offset with a sentinel dataKey enables
efficient client hand-offs in failure scenarios

32. Data Model
CREATE TABLE key_value
partitionKey INTEGER,
dataKey BLOB,
dataValue BLOB,
epoch BIGINT,
offset BIGINT,
PRIMARY KEY ((partitionKey),
dataKey);

What’s This?

33. Data Model
CREATE TABLE key_value
partitionKey INTEGER,
dataKey BLOB,
dataValue BLOB,
epoch BIGINT,
offset BIGINT,
PRIMARY KEY ((partitionKey),
dataKey);

34. Node B
Node A
process Kafka commit
Fencing Zombies

35. Node B
Node A
process Kafka commit
write to
ScyllaDB
Fencing Zombies

36. Node B
Node A
process Kafka commit
write to
ScyllaDB
process Kafka commit
write to
ScyllaDB
Fencing Zombies

37. Node B
Node A
process Kafka commit
write to
ScyllaDB
process Kafka commit
write to
ScyllaDB
prevents other nodes from
committing to Kafka (but not
writing to Scylla!)
Fencing Zombies

38. Node B
Node A
process Kafka commit
write to
ScyllaDB
process Kafka commit
write to
ScyllaDB
write to
ScyllaDB
Fencing Zombies

39. Node B
Node A
process Kafka commit
write to
ScyllaDB
process Kafka commit
write to
ScyllaDB
write to
ScyllaDB
Might overwrite data from the
previous ScyllaDB write!
Fencing Zombies

40. Data Model
CREATE TABLE key_value
partitionKey INTEGER,
dataKey BLOB,
dataValue BLOB,
epoch BIGINT,
offset BIGINT,
PRIMARY KEY ((partitionKey),
dataKey);

BEGIN BATCH;
starts an Atomic Batch (not
for speed, but for LWT)

41. Data Model
CREATE TABLE key_value
partitionKey INTEGER,
dataKey BLOB,
dataValue BLOB,
epoch BIGINT,
offset BIGINT,
PRIMARY KEY ((partitionKey),
dataKey);

BEGIN BATCH;
UPDATE key_value
SET epoch = 12
WHERE
partitionKey = 1
dataKey = metadata_key
IF epoch <= 12;

42. Data Model
CREATE TABLE key_value
partitionKey INTEGER,
dataKey BLOB,
dataValue BLOB,
epoch BIGINT,
offset BIGINT,
PRIMARY KEY ((partitionKey),
dataKey);

BEGIN BATCH;
UPDATE key_value
SET epoch = 12
WHERE
partitionKey = 1
dataKey = metadata_key
IF epoch <= 12;
INSERT INTO key_value VALUES …;
INSERT INTO key_value VALUES …;
INSERT INTO key_value VALUES …;
INSERT INTO key_value VALUES …;

43. Data Model
CREATE TABLE key_value
partitionKey INTEGER,
dataKey BLOB,
dataValue BLOB,
epoch BIGINT,
offset BIGINT,
PRIMARY KEY ((partitionKey),
dataKey);

BEGIN BATCH;
UPDATE key_value
SET epoch = 12
WHERE
partitionKey = 1
dataKey = metadata_key
IF epoch <= 12;
INSERT INTO key_value VALUES …;
INSERT INTO key_value VALUES …;
INSERT INTO key_value VALUES …;
INSERT INTO key_value VALUES …;
APPLY BATCH;

44. Data Model
CREATE TABLE key_value
partitionKey INTEGER,
dataKey BLOB,
dataValue BLOB,
epoch BIGINT,
offset BIGINT,
PRIMARY KEY ((partitionKey),
dataKey);

BEGIN BATCH;
UPDATE key_value
SET epoch = 12
WHERE
partitionKey = 1
dataKey = metadata_key
IF epoch <= 12;
INSERT INTO key_value VALUES …;
INSERT INTO key_value VALUES …;
INSERT INTO key_value VALUES …;
INSERT INTO key_value VALUES …;
APPLY BATCH;

45. Lessons Learned

46. Latency Issues? Check Disk!
Bloom Filter Usage
An increase in disk reads is often a symptom
something else is wrong. For us, there were a few
times Bloom Filters were not properly constructed
(once due to a config, once due to a bug)!

47. Use LWT Only When Necessary
Cost of Atomic Batches
Atomic Batches significantly slow down write
throughput, even if they’re contained to only a
single partition
Throughput Before/After Removing LWTs

48. Selecting Node Size
Node Size
How to choose your ScyllaDB Cloud node type?

49. Consistency
Don’t Be Inconsistent!
Make sure you’ve set your Read / Write consistency
levels. The default is ONE / ONE, which can give
inconsistent results!
Favor Fast Reads
Favor Fast Writes
Inconsistent ConsistentONE/ONE

50. Consistency
Don’t Be Inconsistent!
Make sure you’ve set your Read / Write consistency
levels. The default is ONE / ONE, which can give
inconsistent results!
Favor Fast Reads
Favor Fast Writes
Inconsistent Consistent
QUORUM/
QUORUM
ONE/ONE

51. Consistency
Don’t Be Inconsistent!
Make sure you’ve set your Read / Write consistency
levels. The default is ONE / ONE, which can give
inconsistent results!
Favor Fast Reads
Favor Fast Writes
Inconsistent Consistent
QUORUM/
QUORUM
ONE/ONE
ONE/ALL
risky, but has niche
use cases

52. Consistency
QUORUM/QUORUM
Need Faster Reads
Option: use ONE/ALL temporarily

53. Consistency
QUORUM/QUORUM
Need Faster Reads
Option: use ONE/ALL temporarily
Migrate to ALL / ALL
This is an intermediate state necessary
to maintain consistency
Run Repair
Ensure all data is available on all nodes
from the QUORUM / QUORUM time

54. Consistency
QUORUM/QUORUM
Need Faster Reads
Option: use ONE/ALL temporarily
Migrate to ALL / ALL
This is an intermediate state necessary
to maintain consistency
Run Repair
Ensure all data is available on all nodes
from the QUORUM / QUORUM time
Enable ONE / ALL
Speedy reads! This can help you
temporarily and migrating back to
QUORUM / QUORUM is safe.

55. Stay in Touch
Almog Gavra
[email protected]
@almog.ai
@agavra
/in/agavra/