Feature di.kafka

[alowrydi]

Kafka producer/consumer module, di.kafka

Extracts code/common/kafka.q from TorQ and packages it as a standalone kdb-x module. Wraps the kafkaq native shared library, providing consumer and producer lifecycle management and a configurable message callback for kdb-x processes that need to produce or consume Kafka messages.

Trello ticket - https://trello.com/c/eDFHYhSx/103-kdb-x-kafka-module

---

Files created

**Files created**

File	Description
di/kafka/init.q	Loads kafka.q, defines export of 8 functions
di/kafka/kafka.q	Full implementation - module state, native function stubs/forwarders, internal helpers, public API, init
di/kafka/kafka.md	Full module documentation - see below
di/kafka/test.csv	34 k4unit unit tests across 9 test groups, runs on any kdb-x machine
di/kafka/test_integration.csv	7 k4unit integration tests verifying C library bindings, requires KDBLIB

---

How to test

Unit tests:

q

k4unit:use`di.k4unit
k4unit.moduletest`di.kafka

q)k4unit.moduletest`di.kafka
2026.06.30T17:14:26.636 start
2026.06.30T17:14:26.636 :/home/alowry/bin/local-kdbx-modules/di/kafka/test.csv 34 test(s)
2026.06.30D16:14:26.656770138 warn PID[1398203] HOST[homer.aquaq.co.uk] kafka: di.kafka initialised (native library not loaded - platform not supported)
2026.06.30D16:14:26.656907082 info PID[1398203] HOST[homer.aquaq.co.uk] kafka: normalisation test
2026.06.30T17:14:26.657 end
All tests passed

Integration tests (requires KDBLIB set, e.g. source setenv.sh in a TorQ checkout, or export KDBLIB=/path/to/your/kafkaq/lib on a non-TorQ system):

q

k4unit:use`di.k4unit
.m.di.0k4unit.KUltf .Q.dd[hsym`$.Q.m.mp`di.kafka;`test_integration.csv]
.m.di.0k4unit.KUrt[]
.m.di.0k4unit.getresults[]

q).m.di.0k4unit.getresults[]
action ms bytes lang code                                      repeat file                                                               msx bytesx ok okms okbytes valid timestamp
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
run    0  0     q    kafka.init[`log`libpath!(logdep;libpath)] 1      :/home/alowry/bin/local-kdbx-modules/di/kafka/test_integration.csv 1   1440   1  1    1       1     2026.06.30T17:15:15.365
true   0  0     q    112h=type .m.di.0kafka.nativeinitconsumer 1      :/home/alowry/bin/local-kdbx-modules/di/kafka/test_integration.csv 0   0      1  1    1       1     2026.06.30T17:15:15.365
true   0  0     q    112h=type .m.di.0kafka.nativeinitproducer 1      :/home/alowry/bin/local-kdbx-modules/di/kafka/test_integration.csv 0   0      1  1    1       1     2026.06.30T17:15:15.365
true   0  0     q    112h=type .m.di.0kafka.rawcleanupconsumer 1      :/home/alowry/bin/local-kdbx-modules/di/kafka/test_integration.csv 0   0      1  1    1       1     2026.06.30T17:15:15.365
true   0  0     q    112h=type .m.di.0kafka.rawcleanupproducer 1      :/home/alowry/bin/local-kdbx-modules/di/kafka/test_integration.csv 0   0      1  1    1       1     2026.06.30T17:15:15.365
true   0  0     q    112h=type .m.di.0kafka.nativesubscribe    1      :/home/alowry/bin/local-kdbx-modules/di/kafka/test_integration.csv 0   0      1  1    1       1     2026.06.30T17:15:15.365
true   0  0     q    112h=type .m.di.0kafka.nativepublish      1      :/home/alowry/bin/local-kdbx-modules/di/kafka/test_integration.csv 0   0      1  1    1       1     2026.06.30T17:15:15.365

Unit test coverage includes: dependency validation (all four failure modes plus error prefix), enabled:0b platform skip, stub behaviour for all six native functions, setkupd and the .kupd forwarder pattern (replacement, delegation, immediate reflection), kx.log instance normalisation via normlog, and log call verification under the binary {[c;m]} contract.

Integration test coverage confirms kafka.init loads a real kafkaq library and that all six native function bindings are type 112h (C function) in module state - proving the forwarder pattern reaches genuine C code after init, not the "not initialised" stub.

---

Bug fix

The kdbx module system snapshots exported function values at use time. The original implementation overwrote the exported native functions directly inside bindfunctions, so the C bindings were unreachable via the public API - every call to kafka.initconsumer etc. after a successful init would have hit the "not initialised" stub regardless of whether the library loaded. Fixed by renaming the six stubs to native* internal targets and adding thin exported forwarders that delegate by bare name dynamically, the same pattern already used for the .kupd callback. Verified via the integration tests' 112h type assertions, which confirm the C bindings are reachable through the exported functions after init.

---

Design decisions

1. Required log dependency, all three keys - Unlike di.eodtime, this module calls info, warn, and error, so init validates and requires all three. Errors immediately with a di.kafka: prefixed message if absent or malformed.

2. normlog for kx.log normalisation - The internal normlog function detects a kx.log instance by the presence of getlvl, sinks, fmts keys and wraps its monadic functions into the binary {[c;m]} contract automatically, embedding context as "kafka: message". Callers can pass a kx.log instance directly without manual wrapping.

3. enabled flag for graceful degradation - Defaults to 1b on l64, 0b elsewhere. When disabled, init succeeds without requiring libpath, logs a warning rather than an error, and all six native functions remain as stubs. This is the basis of the entire unit test suite - all 34 unit tests run with enabled:0b, so they execute on any kdb-x machine with no native library required.

4. KDBLIB fallback for libpath - If libpath is omitted, the module falls back to the standard TorQ KDBLIB environment variable. This is not a strict TorQ requirement - any user with their own kafkaq.so build can export KDBLIB themselves to point at it. The integration test guard checks only for KDBLIB and the resolved library file, not for TorQ itself.

5. Stubs distinguish "not initialised" from genuine errors - All six native functions throw a descriptive di.kafka: prefixed error before init, rather than failing with an obscure type error or silently returning nothing.

6. Single deps dict - Config values (libpath, enabled, kupd) and the log dependency are passed together in a single deps dict to init, consistent with the project dependency injection guidelines. All config keys are optional with sensible defaults; only log is required.

7. Root namespace side effect for .kupd - The native kafkaq library is compiled to call .kupd in the root namespace on message receipt; this is unavoidable given the C library's design. init installs a forwarder at .kupd that delegates to module-local state, so setkupd can replace the actual callback at runtime without re-initialising or re-registering with the C library.

8. Export list reviewed against the public API criterion - All eight exports (init, initconsumer, initproducer, cleanupconsumer, cleanupproducer, subscribe, publish, setkupd) are genuine consumer-facing functions. No testing hooks or internal helpers are present in init.q.

---

Checklist

• 34/34 unit tests passing
• 7/7 integration tests passing
• Follows consistency.md and style.md
• Follows dependency injection guidelines

---

Documentation

See kafka.md for full reference including dependency table, configuration options with defaults, exported function documentation with examples, usage example, and notes.

[DI-Software-Engineering]

DIReview Summary

2 critical | 8 warning(s) | 0 suggestion(s)

⚠️ Spec check skipped — tracker lookup failed (NO_REF_FOUND). Standards axis only.

[DI-Software-Engineering]

DIReview Summary

0 critical | 1 warning(s) | 0 suggestion(s)

⚠️ Spec check skipped — tracker lookup failed (NO_REF_FOUND). Standards axis only.

[DI-Software-Engineering]

DIReview Summary

1 critical | 0 warning(s) | 0 suggestion(s)

⚠️ Spec check skipped — tracker lookup failed (NO_REF_FOUND). Standards axis only.