2. Data flow

Firebird writes changes of published tables to the replication journal. When a segment is full (or after journal_archive_timeout), Firebird archives it. The journal_archive_command can copy the archived segment to an extra folder. Each fb_streaming task watches one folder. See “How it works?” in fb_streaming.md.

3. Replication events

The parser raises these events. Each plugin decides which events it uses and which it ignores. The same list is in fb_streaming.md and fb_repl_print.md.

4. Transaction buffering (interactive)

The message plugins (Kafka, RabbitMQ, MongoDB, Lucene) keep one buffer per active transaction. Inside the buffer there is a stack of savepoints. Documents go to the top savepoint. RELEASE moves documents one level down. UNDO drops the level and its documents. COMMIT sends everything. ROLLBACK drops everything. SET SEQUENCE does not wait: it is sent at once. Algorithms: rabbitmq_plugin.md, mongodb_events_plugin.md, fts_lucene_plugin.md.

5. Control file and crash recovery

For each task the service keeps a control file named <database GUID>. It is created in controlFileDir (default: the task folder). Source: fb_streaming.md. It holds:

• control data (database GUID, size, and so on);
• the last processed segment and the position in it;
• the segment with the oldest active transaction;
• the list of active transactions as pairs {tnxNumber, segmentNumber} (segment where each started).

Delete rule: a segment is deleted only when its number is lower than the segment of the oldest active transaction. Restart rule: after a crash, the service rereads segments from the oldest-active-transaction segment up to the last processed one, replays the events of the open transactions, and then goes on as usual.

6. Configuration

replication.conf (Firebird side)

Turn on the journal and archive for the database. A task folder must be used by only one task, and processed segments are deleted. So, if you also run a replica or more tasks, copy each archived segment to one folder per consumer:

database = d:\fbdata\5.0\test.fdb
{
   journal_directory         = d:\fbdata\5.0\replication\test\journal
   journal_archive_directory = d:\fbdata\5.0\replication\test\archive
   journal_archive_command   = "copy $(pathname) $(archivepathname) && copy $(pathname) d:\fbdata\5.0\replication\test\kafka_source"
   journal_archive_timeout   = 10
}

If the journal is used only for streaming, set journal_archive_directory straight to the task folder and omit the command. Then publish the tables:

ALTER DATABASE INCLUDE ALL TO PUBLICATION;      -- or: INCLUDE CUSTOMERS TO PUBLICATION
ALTER DATABASE ENABLE PUBLICATION;

fb_streaming.conf (service side)

The file sits next to the executable. Global settings come first, then one task block per source folder. Full parameter list: fb_streaming.md (“Configuring the service”) and the plugin documents below.

# logLevel  = info                     # trace, debug, info, warning, error, critical, off
# pluginDir = $(root)/stream_plugins   # $(root) = service folder

task = d:\fbdata\5.0\replication\test\kafka_source
{
    database            = inet://localhost:3055/test
    username            = SYSDBA
    password            = masterkey
    plugin              = kafka_cdc_plugin
    deleteProcessedFile = true
    # controlFileDir    = ...            # default: the task folder
    # errorTimeout      = 60
    # plugin-specific parameters follow
}

* required. Common plugin filters: include_tables / exclude_tables (regular expressions), dumpBlobs, register_ddl_events, register_sequence_events, charset_for_none_fields.

Install and run

fb_streaming install [service_name]
fb_streaming start   [service_name]
fb_streaming stop    [service_name]
fb_streaming remove  [service_name]

sudo systemctl enable fb_streaming
sudo systemctl start  fb_streaming
sudo systemctl stop    fb_streaming
sudo systemctl disable fb_streaming

7. Plugins

simple_json_plugin — segment to JSON file

Demo plugin shipped with the service. It turns each binary segment into a JSON file with the same name plus .json. Full description: simple_json_plugin.md. The root object has header (guid, sequence, state, version) and an events array.

{
  "header": { "version": 1, "sequence": 1, "state": "archive",
              "guid": "{F396449D-F6E4-4812-875E-248AB7C2BEE7}" },
  "events": [
    { "event": "START TRANSACTION", "tnx": 1937 },
    { "event": "SAVEPOINT", "tnx": 1937 },
    { "event": "UPDATE", "table": "CUSTOMER", "tnx": 1937,
      "changedFields": ["ADDRESS"],
      "oldRecord": { "CUSTOMER_ID": 8, "ADDRESS": null, ... },
      "record":    { "CUSTOMER_ID": 8, "ADDRESS": "555", ... } },
    { "event": "RELEASE SAVEPOINT", "tnx": 1937 },
    { "event": "COMMIT", "tnx": 1937 }
  ]
}

kafka_cdc_plugin — Change Data Capture to Kafka

Writes row-level change events in a format close to Debezium. Events are buffered per transaction and sent on commit. Each event has a key (primary key, or unique key if no PK) and a value (an envelope with before / after). Full description with schemas, all kafka_* options and a Docker Kafka setup: kafka_cdc_plugin.md.

{ "before": null,
  "after":  { "ID": 1, "FIRST_NAME": "Anne", "LAST_NAME": "Kretchmar", "EMAIL": "[email protected]" },
  "source": { "dbguid": "{9D66A972-...}", "sequence": 1, "filename": "TEST.FDB.journal-000000001",
              "table": "CUSTOMERS", "tnx": 200, "ts_ms": 1711288254908 },
  "op": "c", "ts_ms": 1711288255056 }

{ "before": { "ID": 1, "FIRST_NAME": "Anne",       ... },
  "after":  { "ID": 1, "FIRST_NAME": "Anne Marie", ... },
  "source": { ..., "sequence": 2, "tnx": 219 },
  "op": "u", "ts_ms": 1711288256121 }

{ "before": { "ID": 1, "FIRST_NAME": "Anne Marie", ... },
  "after":  null,
  "source": { ..., "sequence": 3, "tnx": 258 },
  "op": "d", "ts_ms": 1711288256152 }

// topic $[topicPrefix].transaction  (transaction_metadata_enable = true)
{ "status": "BEGIN", "id": 901, "ts_ms": 1713099401529, "event_count": null, "data_collections": null }
{ "status": "END",   "id": 901, "ts_ms": 1713099404605, "event_count": 2,
  "data_collections": [ { "data_collection": "fbstreaming.cdc.CUSTOMERS", "event_count": 2 } ] }

// each change event then also gets:
"transaction": { "id": 901, "total_order": 1, "data_collection_order": 1 }

rabbitmq_plugin — events to a RabbitMQ exchange

Each event is a JSON message sent to exchange_name. With exchange_type = fanout (default) every bound queue gets it; with direct / topic the routing_key decides. If queue_name is set, the queue is declared and bound. The exchange is created if it does not exist. Full description: rabbitmq_plugin.md.

task = d:\fbdata\4.0\replication\horses\archive_rabbit
{
    database = inet://localhost:3054/d:\fbdata\4.0\horses.fdb
    username = SYSDBA
    password = masterkey
    plugin   = rabbitmq_plugin
    rabbit_uri    = amqp://guest:guest@localhost
    exchange_name = horses
    exchange_type = direct
    queue_name    = horses_events
    routing_key   = 123
    dumpBlobs = true
}

{ "event": "UPDATE", "table": "COLOR", "tnx": 11771, "changedFields": ["NAME_DE"],
  "oldRecord": { "CODE_COLOR": 14, "NAME_DE": "",  ... },
  "record":    { "CODE_COLOR": 14, "NAME_DE": "g", ... } }
{ "event": "SET SEQUENCE", "sequence": "SEQ1", "value": 1 }
{ "event": "EXECUTE SQL", "sql": "CREATE SEQUENCE SEQ1", "tnx": 6590 }

mongodb_events_plugin — events to MongoDB

Same event model and same buffering as RabbitMQ. Each event becomes a document in the events collection of mongo_database (created on first write). Parameters: mongo_uri (for example mongodb://localhost:27017), mongo_database, plus the common filters. Full description: mongodb_events_plugin.md.

use horses
db.events.find()
{ _id: ObjectId("638f37f9022b0000ad005779"), event: 'INSERT', table: 'TABLE1', tnx: 6597, record: { ID: 1, S: 'Hello' } }

fts_lucene_plugin — keep full-text indexes current

Updates indexes made by IBSurgeon Full Text Search UDR. On each iteration the plugin rereads index metadata, takes only active indexes, and filters DML events by their tables. On COMMIT it adds, updates, or deletes index documents. Full description: fts_lucene_plugin.md.

8. fb_repl_print utility

Prints a binary segment in readable form. Use it to check what a segment holds, or to find the record that broke replication. Full description: fb_repl_print.md.

fb_repl_print [file_path] <options>
  -i, --input-file  file      segment file (or first argument)
  -H, --print-header          segment header: version, GUID, sequence, state, length
  -B, --print-blobs           BLOB contents in hex
  -R, --dump-records          record values as JSON (needs --database)
  -F, --print-format          record format: field names and types (needs -R)
  -d, --database / -u, --username / -p, --password
  -c, --charset-none-fields   charset for NONE fields

BLOCK (offset: 48)
[200] START (offset: 64)
[200] SAVE (offset: 65)
[200] SAVE (offset: 66)
[200] INSERT CUSTOMERS (length: 3082) (offset: 78)
[200] RELEASE (offset: 3169)
[200] RELEASE (offset: 3170)
BLOCK (offset: 3171)
[] SET SEQUENCE RDB$2 = 1 (offset: 3194)
[200] COMMIT (offset: 3207)

9. Writing your own plugin

A plugin is a shared library in C++ or Delphi. It implements three interfaces and exports the entry point fb_stream_plugin, which registers a factory by name. Helper templates with the Impl suffix reduce boilerplate. Full examples: IBSurgeon/FBStreamingExamples.

Records come as IStreamedRecord (getCount, getField, raw data). Fields are IStreamedField (getName, getType, getScale, getCharSet, getData, isKey, keyPosition, isNullable). IStreamLogger writes to the service log. Every function is described in writing_plugin.md.

extern "C" FB_DLL_EXPORT void fb_stream_plugin(IMaster* master, IStreamPluginManager* pm)
{
    auto factory = new MyPlugin::MyStreamPluginFactory(master);
    pm->registerPluginFactory("my_plugin", factory);   // plugin = my_plugin in fb_streaming.conf
}

10. Limits and rules

Further reading

Source documents in this folder (Markdown):

External: FBStreamingExamples (plugin source examples), lucene_udr (Full Text Search UDR), Debezium (event format reference).