Kafka queues

In uala-backend applications, there are two systems that leverage on Kafka messagging queue: versioning and entity changes.

They both work in a similar way, producing messages into topics and on the other end, there are consumers that consume them.

The waterdrop gem is configured to send messages and the client used for messages producing is:

:::code source="../../uala-core/lib/uala/messaging_queue_producer.rb" title="uala-main/lib/uala/messaging_queue_producer.rb" range="18-34" :::

Setup Kafka locally

In a terminal window, launch Kafka:

cd uala-backend && docker compose up kafka

In another window, run Kafka consumer process for Venue updates:

cd uala-backend/tw-integration && bundle exec racecar VenueConsumer

Finally, produce Kafka messages for venues topic:

cd uala-backend/uala-main && bundle exec rails runner "Uala::EntityChangeTracker.setup!; Venue.first!.touch"

In the racecar terminal window, you should see something like:

...
Received message in topic venues with key 1
...

Versioning

This system is composed by multiple parts and its purpose is to store versions of models that changes over time into a dedicated system. Kafka is used to facilitate an asynchronous implementation with an external system that consumes those messages and store them.

==- Records are stored into a dedicated ClickHouse cluster into a single table:

CREATE TABLE default.model_histories_global
(
    `id` UUID,
    `model_name` String,
    `object_id` Int64,
    `action` String,
    `object_data` String,
    `object_changes` String,
    `whodunnit` String,
    `howdunnit` String,
    `date` Date,
    `created_at` DateTime,
    `updated_at` DateTime
)
ENGINE = MergeTree
PRIMARY KEY (object_id, model_name)
ORDER BY (object_id, model_name, updated_at)

===

A single topic is used for all those messages, and it's defined using the VERSIONING_KAFKA_TOPIC environment variable.

In the ClickHouse cluster, a mechanism is configured to consume those messages and insert them in the table. That's managed in ClickHouse Cloud using "Clickpipes".

The implementation for the versioning logic can be found in uala-core/lib/uala/versioning and models that are versioned include the Uala::Versioning::Serializable module.

Versioning records are then accessible in uala-admin using the versioning panel: in the model page, a button labeled "All versions" will display a page with all versions available for that model instance.

Entity changes

Very similarly to the versioning system, the entity changes system tracks changes that occur to some specific models and produce messages about them. It can be configured to send the full JSON representation of the object or just the ID.

The full implementation is defined in:

:::code source="../../uala-core/lib/uala/model_props/changeable.rb" title="uala-main/lib/uala/model_props/changeable.rb" range="6-11" :::

Differently from versioning system, in here each model has its own topic, that correspond to the table_name of the model.

There can be multiple consumers configured, but at the moment there are one for each model in tw-integration/app/consumers that uses the racecar gem to consume those messages. Additional details of it can be found in marketplace synchronization doc.

How to add new model to entity-changes

1. enable the consumer in IaC application
2. enable the consumer in IaC template and environments
3. create the Kafka topic
4. enable producer and consumer with ACL
5. include Uala::ModelProps::Changeable module and enable it with track_changes method
6. monitor the SumOffsetLag metric making sure processing power is enough
7. if more consumer are needed, remember updating the partitions setting of the topic to match the consumer's replicas

Operations on Kafka cluster

Here you can find some useful commands to interact with a Kafka cluster when running in a bitnami Docker image with Kafka CLI, see also this PR for additional reference.

Please note that if Kafka is using SCRAM authentication, the following configuration file should be created:

echo 'security.protocol: SASL_SSL
sasl.mechanism: SCRAM-SHA-512
sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="..." password="...";' > /tmp/config.properties

and then passed to Kafka commands with --command-config /tmp/config.properties.

List topics

/opt/bitnami/kafka/bin/kafka-topics.sh --list --bootstrap-server url:port

Create a topic

/opt/bitnami/kafka/bin/kafka-topics.sh --create --if-not-exists --bootstrap-server url:port --partitions P --replication-factor N --topic topic-name

Add consumer/producer ACL to a topic for a consumergroup

/opt/bitnami/kafka/bin/kafka-acls.sh --bootstrap-server url:port --add --allow-principal User:... --operation Write --group topic-producer --topic topic-name
/opt/bitnami/kafka/bin/kafka-acls.sh --bootstrap-server url:port --add --allow-principal User:... --operation Read --group topic-consumer --topic topic-name