bsideup/liiklus repository overview

⁠Liiklus

⁠Why

• horizontally scalable RSocket/gRPC streaming gateway
• supports as many client languages as RSocket+gRPC do (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:$LATEST_VERSION

Where the latest version is:

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"]
   ))
   

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

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

   Copy
3. ACK records

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

   Copy
   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⁠:

var stub = ReactorLiiklusServiceGrpc.newReactorStub(channel);
stub
    .subscribe(
        SubscribeRequest.newBuilder()
            .setTopic("user-events")
            .setGroup("analytics")
            .setAutoOffsetReset(AutoOffsetReset.EARLIEST)
            .build()
    )
    .flatMap(reply -> stub
        .receive(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(
                    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⁠.