chakra-coder/liiklus

gRPC-based Gateway for the event-based systems

Liiklus

Liiklus [li:klus] ("traffic" in Estonian) - gRPC-based Gateway for the event-based systems from the ones who think that Kafka is too low-level.

Why

• horizontally scalable gRPC streaming gateway
• supports as many client languages as gRPC does (Java, Go, C++, Python, etc...)
• reactive first
• Per-partition backpressure-aware sources
• at-least-once/at-most-once delivery guarantees
• pluggable event storage (Kafka, Pulsar, Kinesis, etc...)
• pluggable positions storage (DynamoDB, Cassandra, Redis, etc...)
• WIP: cold event storage support (S3, Minio, SQL, key/value, etc...)

Who is using

• https://vivy.com/ - 25+ microservices, an abstraction in front of Kafka for the Shared Log Infrastructure (Event Sourcing / CQRS)

Quick Start

The easiest (and recommended) way to run Liiklus is with Docker:

$ docker run \
    -e kafka_bootstrapServers=some.kafka.host:9092 \
    -e storage_positions_type=MEMORY \ # only for testing, DO NOT use in production
    -p 6565:6565 \
    bsideup/liiklus:0.1.8

Now use LiiklusService.proto to generate your client.

The clients must implement the following algorithm:

1. Subscribe to the assignments:

   stub.subscribe(SubscribeRequest(
       topic="your-topic",
       group="your-consumer-group",
       [autoOffsetReset="earliest|latest"]
   ))
   

2. For every emitted reply of Subscribe, using the same channel, subscribe to the records:

   stub.receive(ReceiveRequest(
       assignment=reply.getAssignment()
   ))
   

3. ACK records

   stub.ack(AckRequest(
       assignment=reply.getAssignment(),
       offset=record.getOffset()
   ))
   

   Note 1: If you ACK record before processing it you get at-most-once, after processing - at-least-once
   Note 2: It's recommended to ACK every n-th record, or every n seconds to reduce the load on the positions storage

Java example:

Example code using Project Reactor and reactive-grpc:

val stub = ReactorLiiklusServiceGrpc.newReactorStub(channel);
stub
    .subscribe(Mono.just(
        SubscribeRequest.newBuilder()
            .setTopic("user-events")
            .setGroup("analytics")
            .setAutoOffsetReset(AutoOffsetReset.EARLIEST)
            .build()
    ))
    .flatMap(reply -> stub
        .receive(Mono.just(ReceiveRequest.newBuilder().setAssignment(reply.getAssignment()).build()))
        .window(1000) // ACK every 1000th records
        .concatMap(
            batch -> batch
                .map(ReceiveReply::getRecord)
                // TODO process instead of Mono.delay(), i.e. by indexing to ElasticSearch
                .concatMap(record -> Mono.delay(Duration.ofMillis(100)))
                .sample(Duration.ofSeconds(5)) // ACK every 5 seconds
                .onBackpressureLatest()
                .delayUntil(record -> stub.ack(Mono.just(
                    AckRequest.newBuilder()
                        .setAssignment(reply.getAssignment())
                        .setOffset(record.getOffset())
                        .build()
                    ))
                ),
            1
        )
    )
    .blockLast()

Also check examples/java/ for a complete example

Configuration

The project is based on Spring Boot and uses it's configuration system
Please check application.yml for the available configuration keys.

License

See LICENSE.