kfka

This project aims to give a subset of the benefits of Apache Kafka (persistent, queryable event queue) without the overhead of managing a Kafka cluster, which sometimes may be a bit overkill for lower message volumes.

Kfka currently supports MySQL as a back-end, and has zero dependencies which makes it ideal as an embedded message queue.

Usage

Configuration

final Duration retentionTime = Duration.ofDays(3);
final Duration cleanInterval = Duration.ofMinutes(30);

final TaskScheduler taskScheduler = ...;
final DataSource dataSource = ...;
final RowMapper<MyKfkaMessage> rowMapper = rs ->
    new MyKfkaMessage.MessageBuilder()
            .userId(rs.getInt("userId"))
            .payload(rs.getBytes("payload"))
            .timestamp(rs.getLong("timestamp"))
            .topic(rs.getString("topic"))
            .type(rs.getString("type"))
            .id(rs.getLong("id"))
            .messageId(rs.getString("message_id"))
            .build();

final KfkaMessageStore msgStore = new MysqlKfkaMessageStore<>(dataSource, rowMapper, retentionTime);
final KfkaManager kfkaManager = new KfkaManagerImpl(msgStore);
taskScheduler.scheduleAtFixedRate(kfkaManager::evictExpired, cleanInterval);

MySQL table definition

CREATE TABLE `kfka` (
  `id` bigint AUTO_INCREMENT NOT NULL,
  `message_id` char(12) NOT NULL,
  `timestamp` bigint NOT NULL,
  `payload` blob NOT NULL,
  `topic` varchar(255) NOT NULL,
  `type` varchar(255) NOT NULL,
  `userId` int,
  PRIMARY KEY (`id`)
);

Listen for events

kfkaManager.addListener((msg)->
{
  // TODO: Handle message
},
new KfkaPredicate()
  .topic("my_topic")
  .rewind(100) // Rewind (up to) 100 messages

Use as a backend for SSE/Websockets messages

kfkaManager.addListener((msg)->
{
  // TODO: Handle message
},
new KfkaPredicate()
  .topic("chat")
  .lastSeenMessageId(45_610) // Next message will be 45,611
  .addPropertyMatch("userId", 123); // Filtering on custom property