Kafka Bridge Tutorial

Learn how to configure the Kafka Bridge to re-publish WAMP events to a Kafka topic.

To enable the Kafka Bridge modify the broker_bridge.kafka.enabled option in the bondy.conf file as shown below:

broker_bridge.kafka.enabled = on

Subscriptions are configured through a specification file, loaded when Bondy starts.

Statically configuring subscriptions

To configure one or more subscriptions you need to define a specification file using the Broker Bridge Specification Object format and modify the bondy.conf to tell Bondy where to find it.

The following snippet provides an example subscriptions specification file.

{
    "id":"subscribers_1",
    "meta":{},
    "subscriptions" : [
        {
            "bridge": "bondy_kafka_bridge",
            "match": {
                "realm": "com.example.realm",
                "topic" : "com.example.user.created",
                "options": {"match": "exact"}
            },
            "action": {
                "type": "produce_sync",
                "topic": "{{kafka.topics.wamp_events}}",
                "key": "\"{{event.topic}}/{{event.publication_id}}\"",
                "value": "{{event}}",
                "options" : {
                    "client_id": "default",
                    "acknowledge": true,
                    "required_acks": "all",
                    "partition": null,
                    "partitioner": {
                        "algorithm": "fnv32a",
                        "value": "\"{{event.topic}}/{{event.publication_id}}\""
                    },
                    "encoding": "json"
                }
            }
        }
    ]
}

The following snippet shows how to modify the bondy.conf file to tell Bondy where to locate the file and how to configure the bridge (this is particular to the Kafka Bridge).

broker_bridge.config_file = $(platform_etc_dir)/broker_bridge_config.json
broker_bridge.kafka.enabled = on
broker_bridge.kafka.clients.default.allow_topic_auto_creation = off
broker_bridge.kafka.clients.default.auto_start_producers = on
broker_bridge.kafka.clients.default.endpoints = [{"127.0.0.1", 9092}]
broker_bridge.kafka.clients.default.max_metadata_sock_retry = 5
broker_bridge.kafka.clients.default.producer.partition_restart_delay_seconds = 2s
broker_bridge.kafka.clients.default.producer.required_acks = 1
broker_bridge.kafka.clients.default.producer.topic_restart_delay_seconds = 10s
broker_bridge.kafka.clients.default.reconnect_cool_down_seconds = 10s
broker_bridge.kafka.clients.default.restart_delay_seconds = 10s
broker_bridge.kafka.topics.wamp_events = com.example.wamp_events

The broker_bridge.kafka.topics.wamp_events key names the Kafka topic. The expression {{kafka.topics.wamp_events}} in the specification file reads it.

Event Object

#{
    <<"realm">> => RealmUri,
    <<"topic">> => Topic,
    <<"subscription_id">> => Event#event.subscription_id,
    <<"publication_id">> => Event#event.publication_id,
    <<"details">> => Event#event.details,
    <<"args">> => Event#event.args,
    <<"kwargs">> => Event#event.kwargs,
    <<"ingestion_timestamp">> => erlang:system_time(millisecond)
}.

Dynamically configuring subscriptions

Bondy has no WAMP or HTTP API for adding or removing a subscription at runtime. The specification file, loaded at startup as shown above, is the only way to configure the Kafka Bridge.