Herbert

Introduction

Herbert is a Kafka-inspired event streaming system built in Rust. Its goal is to act as a data exchange point for transactional and analytic workloads building on top of stream processing patterns, while still allowing batch SQL workflows. It should blur the border between batch and stream processing by allowing to easily switch between them. Key foundation for this is the stream-table duality, where in contrast to Kafka, every Topic is not just a log, but also materialized as a table.

Current implementation

Currently Herbert is a simple single instance multi-threaded TCP server, that implements a small subset of the Kafka protocol and missing functionality in its own Herbert protocol. For the topic implementation it uses an in-memory log which is backed by a write-ahead log (WAL) for durability. Either arbitrary bytes or Apache Arrow RecordBatches can be pushed to it, for the latter the topic is schema-aware and validates incoming data first. Consumers of a topic can commit offsets on a record level, which are also persisted for durability.

Usage

Start a Herbert instance by running

RUST_LOG=info cargo run --bin herbert

To create a schema aware topic, first prepare a schema in a json format, e.g. see the examples/basic_usage dir

{
  "fields": [
    {
      "name": "id",
      "nullable": false,
      "data_type": "Int64",
      "metadata": {},
      "dict_id": 0,
      "dict_is_ordered": false
    },
    {
      "name": "name",
      "nullable": false,
      "data_type": "Utf8",
      "metadata": {},
      "dict_id": 0,
      "dict_is_ordered": false
    }
  ],
  "metadata": {}
}

and then use

RUST_LOG=info cargo run --bin herbert-cli create-topic --broker 127.0.0.1:9002 --topic foobar --schema-path schema.json

To produce data prepare records as a JSON line file, e.g.

{"id": 1, "name": "Stefan"}
{"id": 2, "name": "Herbert"}

and then run

RUST_LOG=info cargo run --bin herbert-cli produce-record-batch --broker 127.0.0.1:9001 --topic foobar --schema-path schema.json --data-path data.jsonl

To consume the data run:

RUST_LOG=info cargo run --bin herbert-cli consume --broker 127.0.0.1:9001 --topic foobar --consumer-group "xyz" --max-messages 1

Roadmap

Polish current state

• Create error.rs with comprehensive error types
• Remove all unwrap() and panic() calls
• Add graceful shutdown handling -> avoid until migrate to tokio
• Write README.md, LICENSE
• Add inline documentation to all public APIs
• Add unit tests for all modules
• Add integration tests (produce/consume flows)
• Add config file support (figment + TOML)
• Set up GitHub Actions CI

Features

• Use tokio runtime.
• Add benchmarks
• Improve concurrent performance by replacing RwLock with one tokio task per topic
• Intoduce static partitions (and think about options for dynamic partitioning)
• Add background task that compacts in-memory Arrow Topic to table using either parquet or vortex.
• Add support for open table formats.
• Add a python API with PyO3.
• Use object storage for durability, see object_store.
• Integrate SQL API with Apache Datafusion