From 9e07e4fa795ed25850b31d6dbf1a0e5615128865 Mon Sep 17 00:00:00 2001
From: alowrydi <aaron.lowry@dataintellect.com>
Date: Mon, 15 Jun 2026 17:44:25 +0100
Subject: [PATCH 01/10] first draft of core module implementation. init.q and
 kafka.q

---
 di/kafka/init.q  |  5 ++++
 di/kafka/kafka.q | 73 ++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 78 insertions(+)
 create mode 100644 di/kafka/init.q
 create mode 100644 di/kafka/kafka.q

diff --git a/di/kafka/init.q b/di/kafka/init.q
new file mode 100644
index 00000000..eac93cc8
--- /dev/null
+++ b/di/kafka/init.q
@@ -0,0 +1,5 @@
+/ kafka consumer/producer interface - wraps kafkaq native library
+
+\l ::kafka.q
+
+export:([init;initconsumer;initproducer;cleanupconsumer;cleanupproducer;subscribe;publish;setkupd])
diff --git a/di/kafka/kafka.q b/di/kafka/kafka.q
new file mode 100644
index 00000000..273975cb
--- /dev/null
+++ b/di/kafka/kafka.q
@@ -0,0 +1,73 @@
+/ kafka consumer/producer interface - wraps kafkaq native library
+
+/ default message handler - discards message silently
+defaultkupd:{[k;x]};
+
+/ default lib path derived from KDBLIB env var and os string
+/ override with libpath config key if KDBLIB is not set in the environment
+defaultlib:`$getenv[`KDBLIB],"/",string[.z.o],"/kafkaq";
+
+/ ============================================================
+/ module state and defaults
+/ ============================================================
+
+/ whether native library is loaded - overwritten by init
+enabled:0b;
+
+/ path to kafkaq library without file extension - overwritten by init
+lib:defaultlib;
+
+/ message handler called by C library on message receipt - overwritten by init
+kupd:defaultkupd;
+
+/ ============================================================
+/ native function stubs - replaced by init when enabled:1b
+/ ============================================================
+
+/ initialise consumer with broker address and option dictionary
+initconsumer:{[s;o]'"kafka not enabled"};
+
+/ initialise producer with broker address and option dictionary
+initproducer:{[s;o]'"kafka not enabled"};
+
+/ disconnect and free consumer object and stop subscription thread
+cleanupconsumer:{'"kafka not enabled"};
+
+/ disconnect and free producer object
+cleanupproducer:{'"kafka not enabled"};
+
+/ start subscription thread for topic on partition - messages delivered to kupd
+subscribe:{[t;p]'"kafka not enabled"};
+
+/ publish byte vector to topic and partition with given key
+publish:{[t;p;k;m]'"kafka not enabled"};
+
+/ ============================================================
+/ public api
+/ ============================================================
+
+setkupd:{[f]
+  / update message handler; propagate to global kupd for C callback if enabled
+  .z.m.kupd:f;
+  if[enabled;@[`.;`kupd;:;f]];
+  };
+
+init:{[config]
+  / normalise config - handles (::) and ()!() identically
+  cfg:$[99h=type config;config;()!()];
+  .z.m.enabled:$[`enabled in key cfg;cfg`enabled;0b];
+  .z.m.lib:$[`libpath in key cfg;cfg`libpath;defaultlib];
+  .z.m.kupd:$[`kupd in key cfg;cfg`kupd;defaultkupd];
+  if[enabled;
+    libfile:hsym ` sv lib,$[.z.o like "w*";`dll;`so];
+    if[()~key libfile;'"kafka lib not found: ",1_string libfile];
+    .z.m.initconsumer:lib 2:(`initconsumer;2);
+    .z.m.initproducer:lib 2:(`initproducer;2);
+    .z.m.cleanupconsumer:lib 2:(`cleanupconsumer;1);
+    .z.m.cleanupproducer:lib 2:(`cleanupproducer;1);
+    .z.m.subscribe:lib 2:(`subscribe;2);
+    .z.m.publish:lib 2:(`publish;4);
+    / set global kupd - unavoidable side effect of native library design
+    @[`.;`kupd;:;kupd];
+    ];
+  };

From 9ca102ed3dfafd30373a2656f6490149ae58b70b Mon Sep 17 00:00:00 2001
From: alowrydi <aaron.lowry@dataintellect.com>
Date: Tue, 16 Jun 2026 14:56:12 +0100
Subject: [PATCH 02/10] Adjusting code to be more faithful to original TorQ
 code, adding logging following the injected dependency pattern observed in
 other developed modules and suite of k4unit test added (25/25 pass)

---
 di/kafka/kafka.q  | 63 ++++++++++++++++++++++++++++++-----------------
 di/kafka/test.csv | 49 ++++++++++++++++++++++++++++++++++++
 2 files changed, 89 insertions(+), 23 deletions(-)
 create mode 100644 di/kafka/test.csv

diff --git a/di/kafka/kafka.q b/di/kafka/kafka.q
index 273975cb..ffe5b9c1 100644
--- a/di/kafka/kafka.q
+++ b/di/kafka/kafka.q
@@ -1,7 +1,7 @@
 / kafka consumer/producer interface - wraps kafkaq native library
 
-/ default message handler - discards message silently
-defaultkupd:{[k;x]};
+/ default message handler - routes message through injected logger
+defaultkupd:{[k;x].z.m.loginfo[`kafka;"kupd: ","c"$x]};
 
 / default lib path derived from KDBLIB env var and os string
 / override with libpath config key if KDBLIB is not set in the environment
@@ -11,8 +11,8 @@ defaultlib:`$getenv[`KDBLIB],"/",string[.z.o],"/kafkaq";
 / module state and defaults
 / ============================================================
 
-/ whether native library is loaded - overwritten by init
-enabled:0b;
+/ whether native library is loaded - overwritten by init; true by default on l64
+enabled:.z.o in `l64;
 
 / path to kafkaq library without file extension - overwritten by init
 lib:defaultlib;
@@ -21,7 +21,7 @@ lib:defaultlib;
 kupd:defaultkupd;
 
 / ============================================================
-/ native function stubs - replaced by init when enabled:1b
+/ native function stubs - replaced by init when enabled:1b and library loads
 / ============================================================
 
 / initialise consumer with broker address and option dictionary
@@ -30,11 +30,11 @@ initconsumer:{[s;o]'"kafka not enabled"};
 / initialise producer with broker address and option dictionary
 initproducer:{[s;o]'"kafka not enabled"};
 
-/ disconnect and free consumer object and stop subscription thread
-cleanupconsumer:{'"kafka not enabled"};
+/ disconnect and free consumer object and stop subscription thread - rank-1 matches C lib 2:(`cleanupconsumer;1)
+cleanupconsumer:{[x]'"kafka not enabled"};
 
-/ disconnect and free producer object
-cleanupproducer:{'"kafka not enabled"};
+/ disconnect and free producer object - rank-1 matches C lib 2:(`cleanupproducer;1)
+cleanupproducer:{[x]'"kafka not enabled"};
 
 / start subscription thread for topic on partition - messages delivered to kupd
 subscribe:{[t;p]'"kafka not enabled"};
@@ -49,25 +49,42 @@ publish:{[t;p;k;m]'"kafka not enabled"};
 setkupd:{[f]
   / update message handler; propagate to global kupd for C callback if enabled
   .z.m.kupd:f;
-  if[enabled;@[`.;`kupd;:;f]];
+  if[.z.m.enabled;@[`.;`kupd;:;f]];
   };
 
-init:{[config]
+init:{[config;deps]
+  / config: dict with optional keys `enabled`libpath`kupd
+  / deps: `log!(logdict)
+  /   `log: `info`warn`error!(infofunc;warnfunc;errfunc) - required
+  logdict:$[99h=type deps;$[(`log in key deps) and not (::)~deps`log;deps`log;()!()];()!()];
+  if[not all `info`warn`error in key logdict;
+    '"di.kafka: log dependency is required; pass `info`warn`error functions - see di.log or refer to confluence documentation";
+    ];
+  .z.m.loginfo:logdict`info;
+  .z.m.logwarn:logdict`warn;
+  .z.m.logerr:logdict`error;
   / normalise config - handles (::) and ()!() identically
   cfg:$[99h=type config;config;()!()];
-  .z.m.enabled:$[`enabled in key cfg;cfg`enabled;0b];
+  .z.m.enabled:$[`enabled in key cfg;cfg`enabled;.z.o in `l64];
   .z.m.lib:$[`libpath in key cfg;cfg`libpath;defaultlib];
   .z.m.kupd:$[`kupd in key cfg;cfg`kupd;defaultkupd];
-  if[enabled;
-    libfile:hsym ` sv lib,$[.z.o like "w*";`dll;`so];
-    if[()~key libfile;'"kafka lib not found: ",1_string libfile];
-    .z.m.initconsumer:lib 2:(`initconsumer;2);
-    .z.m.initproducer:lib 2:(`initproducer;2);
-    .z.m.cleanupconsumer:lib 2:(`cleanupconsumer;1);
-    .z.m.cleanupproducer:lib 2:(`cleanupproducer;1);
-    .z.m.subscribe:lib 2:(`subscribe;2);
-    .z.m.publish:lib 2:(`publish;4);
-    / set global kupd - unavoidable side effect of native library design
-    @[`.;`kupd;:;kupd];
+  if[.z.m.enabled;
+    libfile:hsym ` sv .z.m.lib,$[.z.o like "w*";`dll;`so];
+    / protected key - kdbx throws on paths with non-existent ancestors
+    libexists:@[{not ()~key x};libfile;{0b}];
+    if[not libexists;
+      .z.m.logerr[`kafka;"no such file ",1_string libfile]
+      ];
+    if[libexists;
+      .z.m.initconsumer:.z.m.lib 2:(`initconsumer;2);
+      .z.m.initproducer:.z.m.lib 2:(`initproducer;2);
+      .z.m.cleanupconsumer:.z.m.lib 2:(`cleanupconsumer;1);
+      .z.m.cleanupproducer:.z.m.lib 2:(`cleanupproducer;1);
+      .z.m.subscribe:.z.m.lib 2:(`subscribe;2);
+      .z.m.publish:.z.m.lib 2:(`publish;4);
+      / set global kupd - unavoidable side effect of native library design
+      @[`.;`kupd;:;.z.m.kupd];
+      .z.m.loginfo[`kafka;"kupd is set to ",-3!.z.m.kupd];
+      ];
     ];
   };
diff --git a/di/kafka/test.csv b/di/kafka/test.csv
new file mode 100644
index 00000000..305895c2
--- /dev/null
+++ b/di/kafka/test.csv
@@ -0,0 +1,49 @@
+action,ms,bytes,lang,code,repeat,minver,comment
+/ pre-test setup
+before,0,0,q,kafka:use`di.kafka,1,1,load di.kafka module
+before,0,0,q,mocklog:`info`warn`error!({[c;m]};{[c;m]};{[c;m]}),1,1,no-op mock log dependency
+before,0,0,q,logdep:enlist[`log]!enlist mocklog,1,1,log dep dict
+before,0,0,q,kafkaterrored:0b,1,1,flag used to prove enabled:0b branch was skipped not merely unsuccessful
+before,0,0,q,capturelog:`info`warn`error!({[c;m]};{[c;m]};{[c;m] `kafkaterrored set 1b}),1,1,capturing mock logger - error fn uses set builtin; only reliable when not called from within module execution context
+
+/ logdict extraction - invalid deps inputs throw correct message
+true,0,0,q,.[kafka.init;(()!();()!());{x}] like "*log dependency*",1,1,empty dict deps throws correct message
+true,0,0,q,.[kafka.init;(()!();(::));{x}] like "*log dependency*",1,1,:: as deps throws correct message
+true,0,0,q,.[kafka.init;(()!();42);{x}] like "*log dependency*",1,1,non-dict deps throws correct message
+true,0,0,q,.[kafka.init;(()!();enlist[`log]!enlist(::));{x}] like "*log dependency*",1,1,log value of :: throws correct message
+true,0,0,q,.[kafka.init;(()!();enlist[`notlog]!enlist mocklog);{x}] like "*log dependency*",1,1,deps without log key throws correct message
+true,0,0,q,.[kafka.init;(()!();enlist[`log]!enlist `info`warn!(mocklog`info;mocklog`warn));{x}] like "*log dependency*",1,1,partial log dict missing error key throws correct message
+
+/ init disabled - stubs throw with correct message
+run,0,0,q,kafka.init[()!();logdep],1,1,init with empty config does not error
+true,0,0,q,.[kafka.initconsumer;(`localhost:9092;()!());{x}] like "*kafka not enabled*",1,1,initconsumer stub throws correct message
+true,0,0,q,.[kafka.initproducer;(`localhost:9092;()!());{x}] like "*kafka not enabled*",1,1,initproducer stub throws correct message
+true,0,0,q,.[kafka.cleanupconsumer;enlist(::);{x}] like "*kafka not enabled*",1,1,cleanupconsumer stub throws correct message
+true,0,0,q,.[kafka.cleanupproducer;enlist(::);{x}] like "*kafka not enabled*",1,1,cleanupproducer stub throws correct message
+true,0,0,q,.[kafka.subscribe;(`test;0);{x}] like "*kafka not enabled*",1,1,subscribe stub throws correct message
+true,0,0,q,.[kafka.publish;(`test;0;`;0x68656c6c6f);{x}] like "*kafka not enabled*",1,1,publish stub throws correct message
+
+/ init - explicit enabled:0b: capturelog proves lib-loading branch was skipped not merely unsuccessful
+run,0,0,q,kafka.init[enlist[`enabled]!enlist 0b;enlist[`log]!enlist capturelog],1,1,init with enabled:0b does not error
+true,0,0,q,not kafkaterrored,1,1,logerr was not called - lib-loading branch was genuinely skipped
+true,0,0,q,.[kafka.subscribe;(`test;0);{x}] like "*kafka not enabled*",1,1,subscribe throws correct message after explicit disabled init
+
+/ init - :: config
+run,0,0,q,kafka.init[::;logdep],1,1,init with :: config does not error
+
+/ init - enabled:1b with missing lib: process continues with stubs intact
+/ note: logerr is called internally but cannot be captured from within kdbx module execution context
+run,0,0,q,kafka.init[`enabled`libpath!(1b;`fakekafkaq);logdep],1,1,init with enabled:1b and missing lib does not throw
+true,0,0,q,.[kafka.subscribe;(`test;0);{x}] like "*kafka not enabled*",1,1,stubs remain after failed lib load
+
+/ setkupd - state unobservable without live broker; test confirms no throw only
+run,0,0,q,kafka.init[()!();logdep],1,1,reinit with no-op log before setkupd test
+run,0,0,q,kafka.setkupd[{[k;x]}],1,1,setkupd does not error when disabled - internal state unobservable without live broker
+
+/ init - custom kupd
+run,0,0,q,kafka.init[(enlist`kupd)!enlist{[k;x]};logdep],1,1,init with custom kupd does not error
+
+/ multiple consecutive init calls do not corrupt state
+run,0,0,q,kafka.init[()!();logdep],1,1,second consecutive init does not error
+run,0,0,q,kafka.init[enlist[`enabled]!enlist 0b;logdep],1,1,third consecutive init with enabled:0b does not error
+true,0,0,q,.[kafka.subscribe;(`test;0);{x}] like "*kafka not enabled*",1,1,stubs remain correct after multiple init calls

From 55b04956f4de641ef9d25eab528564687a371fe5 Mon Sep 17 00:00:00 2001
From: alowrydi <aaron.lowry@dataintellect.com>
Date: Tue, 16 Jun 2026 16:33:02 +0100
Subject: [PATCH 03/10] markdown documentation

---
 di/kafka/kafka.md | 203 ++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 203 insertions(+)
 create mode 100644 di/kafka/kafka.md

diff --git a/di/kafka/kafka.md b/di/kafka/kafka.md
new file mode 100644
index 00000000..8c222f74
--- /dev/null
+++ b/di/kafka/kafka.md
@@ -0,0 +1,203 @@
+# di.kafka
+
+Kafka consumer and producer interface for kdb+ processes. Wraps the `kafkaq` native C shared library to provide consumer initialisation, producer initialisation, topic subscription, and message publishing. Incoming messages are delivered to a configurable `kupd` callback in the root namespace.
+
+---
+
+## Features
+
+- Connect kdb+ processes to a Kafka broker as consumer, producer, or both
+- Subscribe to topics and receive messages via a configurable callback function
+- Publish byte vectors to topics with optional message keys
+- Graceful disabled mode on unsupported platforms - all native functions are replaced with informative stubs
+- Message handler (`kupd`) configurable at init or updated live via `setkupd`
+- Compatible with any logging implementation via the injected `log` dependency
+
+---
+
+## Dependencies
+
+**Native library:** `kafkaq.so` (Linux) or `kafkaq.dll` (Windows). Must be present on the host. Path is provided via the `libpath` config key or derived from the `KDBLIB` environment variable.
+
+**Injected:** `log` - required. Pass `info`, `warn`, and `error` log functions via the `deps` argument to `init`. See `di.log` or the Confluence documentation for the expected function signatures.
+
+---
+
+## Initialisation
+
+```q
+log:use`di.log
+log.init[logconfig]
+logdep:enlist[`log]!enlist `info`warn`error!(log.info;log.warn;log.error)
+
+kafka:use`di.kafka
+kafka.init[`enabled`libpath!(1b;`$/opt/TorQ/lib/l64/kafkaq);logdep]
+```
+
+All config keys are optional. Passing `(::)` or `()!()` as config applies defaults.
+
+| Key | Type | Default | Description |
+|---|---|---|---|
+| `enabled` | boolean | `1b` on `l64`, `0b` on all other platforms | Whether to load the native library. On a non-`l64` host the module loads in disabled mode unless explicitly overridden. |
+| `libpath` | symbol | derived from `KDBLIB` env var | Path to the kafkaq library **without** file extension. If `KDBLIB` is not set, provide this explicitly. |
+| `kupd` | function | logs message via injected logger | Message handler called by the C library on Kafka message receipt. Signature: `{[key;bytes]}`. |
+
+The `log` dependency is **required**. `init` throws if it is absent, `(::)`, or missing any of `info`/`warn`/`error`.
+
+`init` must be called before any other function is used. It is safe to call multiple times.
+
+---
+
+## Functions
+
+### `init`
+```q
+kafka.init[config;deps]
+```
+Initialises the module. Loads the native library when `enabled:1b` and the library is found. Sets the global `kupd` callback for the C library.
+
+### `initconsumer`
+```q
+kafka.initconsumer[`localhost:9092;`fetch.wait.max.ms`fetch.error.backoff.ms!`5`5]
+```
+Initialises a Kafka consumer and connects to the broker.
+
+### `initproducer`
+```q
+kafka.initproducer[`localhost:9092;`queue.buffering.max.ms`batch.num.messages!`5`1]
+```
+Initialises a Kafka producer and connects to the broker.
+
+### `cleanupconsumer`
+```q
+kafka.cleanupconsumer[(::)]
+```
+Disconnects and frees the consumer object, stopping the subscription thread.
+
+### `cleanupproducer`
+```q
+kafka.cleanupproducer[(::)]
+```
+Disconnects and frees the producer object.
+
+### `subscribe`
+```q
+kafka.subscribe[`mytopic;0]
+```
+Starts the subscription thread for a topic on a given partition. Messages are delivered to `kupd`.
+
+### `publish`
+```q
+kafka.publish[`mytopic;0;`;`byte$"hello world"]
+kafka.publish[`mytopic;0;`mykey;`byte$"hello world"]
+```
+Publishes a byte vector to a topic and partition with an optional symbol key.
+
+### `setkupd`
+```q
+kafka.setkupd[{[k;x] upd[`kafkadata;(enlist k;enlist x)]}]
+```
+Updates the message handler after `init` without requiring a full reinitialisation. Also updates the global `kupd` in the root namespace if the module is enabled.
+
+`init` must be called before `setkupd`.
+
+> **Warning:** Avoid calling `setkupd` while a subscription is active. The Kafka C library delivers messages on a background thread and may invoke the old handler after the swap.
+
+---
+
+## Usage example
+
+```q
+log:use`di.log
+log.init[logconfig]
+logdep:enlist[`log]!enlist `info`warn`error!(log.info;log.warn;log.error)
+
+myhandler:{[k;x]
+  upd[`kafkadata;(enlist .z.p;enlist k;enlist "c"$x)];
+  };
+
+kafka:use`di.kafka
+kafka.init[`enabled`libpath`kupd!(1b;`$/opt/TorQ/lib/l64/kafkaq;myhandler);logdep]
+
+kafka.initconsumer[`localhost:9092;()!()]
+kafka.subscribe[`trades;0]
+
+kafka.initproducer[`localhost:9092;()!()]
+kafka.publish[`trades;0;`;`byte$"test message"]
+
+kafka.cleanupconsumer[(::)]
+kafka.cleanupproducer[(::)]
+```
+
+---
+
+## Global side effect
+
+When `enabled:1b` and the library loads successfully, `init` sets `kupd` in the **root namespace**:
+
+```q
+@[`.;`kupd;:;.z.m.kupd]
+```
+
+This is an unavoidable consequence of the native C library's design - kafkaq is compiled to call `kupd` by name in the root namespace. `setkupd` also updates the global when enabled.
+
+---
+
+## Disabled mode
+
+When `enabled:0b` - either because the platform default resolved to `0b` or it was explicitly set - every native function (`initconsumer`, `initproducer`, `cleanupconsumer`, `cleanupproducer`, `subscribe`, `publish`) is a stub that throws:
+
+```
+'kafka not enabled
+```
+
+This is the expected behaviour when kafka is not set up. It distinguishes "kafka isn't configured" from a genuine broker or argument error.
+
+If `enabled:1b` and the library file is not found, `init` logs an error and continues rather than throwing - the process stays alive and all functions remain as stubs. This matches TorQ's original behaviour: a missing native library degrades the module to disabled rather than halting the process.
+
+> **Note:** If kafka silently isn't working after `init` with `enabled:1b`, check the injected logger's error output. `init` logs the missing library path but does not throw - there is nothing to catch.
+
+---
+
+## TorQ migration
+
+| TorQ pattern | Module equivalent |
+|---|---|
+| `enabled:@[value;\`enabled;.z.o in \`l64]` | `kafka.init[\`enabled!enlist 1b;logdep]` |
+| `kupd:{[k;x]...}` defined in settings | `kafka.init[\`kupd!enlist{[k;x]...};logdep]` |
+| default `kupd` prints bytes to stdout (`-1 \`char$x`) | default `kupd` routes through injected logger - override with `kupd` config key if stdout behaviour is needed |
+| `.kafka.initconsumer[s;o]` | `kafka.initconsumer[s;o]` |
+| `.kafka.initproducer[s;o]` | `kafka.initproducer[s;o]` |
+| `.kafka.cleanupconsumer[(::)]` | `kafka.cleanupconsumer[(::)]` |
+| `.kafka.cleanupproducer[(::)]` | `kafka.cleanupproducer[(::)]` |
+| `.kafka.subscribe[t;p]` | `kafka.subscribe[t;p]` |
+| `.kafka.publish[t;p;k;m]` | `kafka.publish[t;p;k;m]` |
+
+---
+
+## Manual verification (requires kafkaq.so and Kafka broker)
+
+```q
+/ verify lib loads
+kafka.init[`enabled`libpath!(1b;`$/path/to/l64/kafkaq);logdep]
+
+/ verify consumer and subscription
+kafka.initconsumer[`localhost:9092;()!()]
+kafka.subscribe[`test;0]
+
+/ verify publish
+kafka.initproducer[`localhost:9092;()!()]
+kafka.publish[`test;0;`;`byte$"hello from kdb+"]
+
+/ cleanup
+kafka.cleanupconsumer[(::)]
+kafka.cleanupproducer[(::)]
+```
+
+---
+
+## Notes
+
+- `enabled` defaults to `1b` only on `l64` hosts; everywhere else it defaults to `0b`. Pass `enabled:1b` explicitly to force-enable on a platform where a native library build is available.
+- The `log` dependency is required with no fallback. This differs from TorQ, which has no equivalent concept of injected logging - this module requires it for consistency with other `di.*` modules.
+- The default `kupd` routes through the injected logger rather than TorQ's original raw stdout write (`-1 \`char$x`). Override with the `kupd` config key if stdout behaviour is needed.

From 03213e0c02c686a412cb4c984950803a4445aeb7 Mon Sep 17 00:00:00 2001
From: alowrydi <aaron.lowry@dataintellect.com>
Date: Mon, 22 Jun 2026 14:04:47 +0100
Subject: [PATCH 04/10] Refactor log dep to kx.log standard - optional with
 no-op fallback, single-arg log functions (22/22 tests pass)

---
 di/kafka/kafka.md | 30 ++++++++++++++++--------------
 di/kafka/kafka.q  | 21 ++++++++++++---------
 di/kafka/test.csv | 17 +++++++----------
 3 files changed, 35 insertions(+), 33 deletions(-)

diff --git a/di/kafka/kafka.md b/di/kafka/kafka.md
index 8c222f74..19909298 100644
--- a/di/kafka/kafka.md
+++ b/di/kafka/kafka.md
@@ -11,7 +11,7 @@ Kafka consumer and producer interface for kdb+ processes. Wraps the `kafkaq` nat
 - Publish byte vectors to topics with optional message keys
 - Graceful disabled mode on unsupported platforms - all native functions are replaced with informative stubs
 - Message handler (`kupd`) configurable at init or updated live via `setkupd`
-- Compatible with any logging implementation via the injected `log` dependency
+- Compatible with `kx.log` and any logger that exposes unary `info`/`warn`/`error` functions
 
 ---
 
@@ -19,19 +19,18 @@ Kafka consumer and producer interface for kdb+ processes. Wraps the `kafkaq` nat
 
 **Native library:** `kafkaq.so` (Linux) or `kafkaq.dll` (Windows). Must be present on the host. Path is provided via the `libpath` config key or derived from the `KDBLIB` environment variable.
 
-**Injected:** `log` - required. Pass `info`, `warn`, and `error` log functions via the `deps` argument to `init`. See `di.log` or the Confluence documentation for the expected function signatures.
+**Injected:** `log` - optional. Pass a `kx.log` logger instance (from `kx.log.createLog[]`) under the `log` key in `deps`. If absent, or if the instance does not provide `info`/`warn`/`error` functions, the module falls back to no-op logging.
 
 ---
 
 ## Initialisation
 
 ```q
-log:use`di.log
-log.init[logconfig]
-logdep:enlist[`log]!enlist `info`warn`error!(log.info;log.warn;log.error)
+kxlog:use`kx.log
+logger:kxlog.createLog[]
 
 kafka:use`di.kafka
-kafka.init[`enabled`libpath!(1b;`$/opt/TorQ/lib/l64/kafkaq);logdep]
+kafka.init[`enabled`libpath!(1b;`$/opt/TorQ/lib/l64/kafkaq);enlist[`log]!enlist logger]
 ```
 
 All config keys are optional. Passing `(::)` or `()!()` as config applies defaults.
@@ -42,7 +41,7 @@ All config keys are optional. Passing `(::)` or `()!()` as config applies defaul
 | `libpath` | symbol | derived from `KDBLIB` env var | Path to the kafkaq library **without** file extension. If `KDBLIB` is not set, provide this explicitly. |
 | `kupd` | function | logs message via injected logger | Message handler called by the C library on Kafka message receipt. Signature: `{[key;bytes]}`. |
 
-The `log` dependency is **required**. `init` throws if it is absent, `(::)`, or missing any of `info`/`warn`/`error`.
+If no `log` dep is provided, or it is missing `info`/`warn`/`error` functions, the module falls back to no-op logging silently.
 
 `init` must be called before any other function is used. It is safe to call multiple times.
 
@@ -108,16 +107,15 @@ Updates the message handler after `init` without requiring a full reinitialisati
 ## Usage example
 
 ```q
-log:use`di.log
-log.init[logconfig]
-logdep:enlist[`log]!enlist `info`warn`error!(log.info;log.warn;log.error)
+kxlog:use`kx.log
+logger:kxlog.createLog[]
 
 myhandler:{[k;x]
   upd[`kafkadata;(enlist .z.p;enlist k;enlist "c"$x)];
   };
 
 kafka:use`di.kafka
-kafka.init[`enabled`libpath`kupd!(1b;`$/opt/TorQ/lib/l64/kafkaq;myhandler);logdep]
+kafka.init[`enabled`libpath`kupd!(1b;`$/opt/TorQ/lib/l64/kafkaq;myhandler);enlist[`log]!enlist logger]
 
 kafka.initconsumer[`localhost:9092;()!()]
 kafka.subscribe[`trades;0]
@@ -165,7 +163,7 @@ If `enabled:1b` and the library file is not found, `init` logs an error and cont
 |---|---|
 | `enabled:@[value;\`enabled;.z.o in \`l64]` | `kafka.init[\`enabled!enlist 1b;logdep]` |
 | `kupd:{[k;x]...}` defined in settings | `kafka.init[\`kupd!enlist{[k;x]...};logdep]` |
-| default `kupd` prints bytes to stdout (`-1 \`char$x`) | default `kupd` routes through injected logger - override with `kupd` config key if stdout behaviour is needed |
+| default `kupd` prints bytes to stdout (`-1 \`char$x`) | default `kupd` routes through injected `kx.log` logger (unary call) - override with `kupd` config key for custom behaviour |
 | `.kafka.initconsumer[s;o]` | `kafka.initconsumer[s;o]` |
 | `.kafka.initproducer[s;o]` | `kafka.initproducer[s;o]` |
 | `.kafka.cleanupconsumer[(::)]` | `kafka.cleanupconsumer[(::)]` |
@@ -178,6 +176,10 @@ If `enabled:1b` and the library file is not found, `init` logs an error and cont
 ## Manual verification (requires kafkaq.so and Kafka broker)
 
 ```q
+kxlog:use`kx.log
+logger:kxlog.createLog[]
+logdep:enlist[`log]!enlist logger
+
 / verify lib loads
 kafka.init[`enabled`libpath!(1b;`$/path/to/l64/kafkaq);logdep]
 
@@ -199,5 +201,5 @@ kafka.cleanupproducer[(::)]
 ## Notes
 
 - `enabled` defaults to `1b` only on `l64` hosts; everywhere else it defaults to `0b`. Pass `enabled:1b` explicitly to force-enable on a platform where a native library build is available.
-- The `log` dependency is required with no fallback. This differs from TorQ, which has no equivalent concept of injected logging - this module requires it for consistency with other `di.*` modules.
-- The default `kupd` routes through the injected logger rather than TorQ's original raw stdout write (`-1 \`char$x`). Override with the `kupd` config key if stdout behaviour is needed.
+- The `log` dependency is optional. Pass a `kx.log` logger instance for production observability; omit it for standalone or test use. Logger functions use a unary `{[msg]}` signature — context is embedded in the message string (e.g. `"kafka: message"`).
+- The default `kupd` routes through the injected logger rather than TorQ's original raw stdout write (`-1 \`char$x`). Override with the `kupd` config key if custom behaviour is needed.
diff --git a/di/kafka/kafka.q b/di/kafka/kafka.q
index ffe5b9c1..58794db7 100644
--- a/di/kafka/kafka.q
+++ b/di/kafka/kafka.q
@@ -1,12 +1,16 @@
 / kafka consumer/producer interface - wraps kafkaq native library
 
 / default message handler - routes message through injected logger
-defaultkupd:{[k;x].z.m.loginfo[`kafka;"kupd: ","c"$x]};
+/ safe to reference .z.m.loginfo here: kupd can only fire after initconsumer+subscribe, both of which require init
+defaultkupd:{[k;x].z.m.loginfo["kafka: kupd: ","c"$x]};
 
 / default lib path derived from KDBLIB env var and os string
 / override with libpath config key if KDBLIB is not set in the environment
 defaultlib:`$getenv[`KDBLIB],"/",string[.z.o],"/kafkaq";
 
+/ default no-op logger - used when no log dep is injected
+defaultlog:`info`warn`error!({[m]};{[m]};{[m]});
+
 / ============================================================
 / module state and defaults
 / ============================================================
@@ -54,12 +58,11 @@ setkupd:{[f]
 
 init:{[config;deps]
   / config: dict with optional keys `enabled`libpath`kupd
-  / deps: `log!(logdict)
-  /   `log: `info`warn`error!(infofunc;warnfunc;errfunc) - required
-  logdict:$[99h=type deps;$[(`log in key deps) and not (::)~deps`log;deps`log;()!()];()!()];
-  if[not all `info`warn`error in key logdict;
-    '"di.kafka: log dependency is required; pass `info`warn`error functions - see di.log or refer to confluence documentation";
-    ];
+  / deps: optional. `log key should be a kx.log logger instance (from kx.log.createLog[])
+  /   if absent or missing info/warn/error keys, falls back to no-op logger
+  / extract deps`log safely; type check before key avoids crash on non-dict lograw (e.g. deps=(::))
+  lograw:$[99h=type deps;@[deps;`log;{(::)}];(::)];
+  logdict:$[99h=type lograw;$[all `info`warn`error in key lograw;lograw;defaultlog];defaultlog];
   .z.m.loginfo:logdict`info;
   .z.m.logwarn:logdict`warn;
   .z.m.logerr:logdict`error;
@@ -73,7 +76,7 @@ init:{[config;deps]
     / protected key - kdbx throws on paths with non-existent ancestors
     libexists:@[{not ()~key x};libfile;{0b}];
     if[not libexists;
-      .z.m.logerr[`kafka;"no such file ",1_string libfile]
+      .z.m.logerr["kafka: no such file ",1_string libfile]
       ];
     if[libexists;
       .z.m.initconsumer:.z.m.lib 2:(`initconsumer;2);
@@ -84,7 +87,7 @@ init:{[config;deps]
       .z.m.publish:.z.m.lib 2:(`publish;4);
       / set global kupd - unavoidable side effect of native library design
       @[`.;`kupd;:;.z.m.kupd];
-      .z.m.loginfo[`kafka;"kupd is set to ",-3!.z.m.kupd];
+      .z.m.loginfo["kafka: kupd set to ",-3!.z.m.kupd];
       ];
     ];
   };
diff --git a/di/kafka/test.csv b/di/kafka/test.csv
index 305895c2..ad4ff6f2 100644
--- a/di/kafka/test.csv
+++ b/di/kafka/test.csv
@@ -1,18 +1,15 @@
 action,ms,bytes,lang,code,repeat,minver,comment
 / pre-test setup
 before,0,0,q,kafka:use`di.kafka,1,1,load di.kafka module
-before,0,0,q,mocklog:`info`warn`error!({[c;m]};{[c;m]};{[c;m]}),1,1,no-op mock log dependency
+before,0,0,q,mocklog:`info`warn`error!({[m]};{[m]};{[m]}),1,1,no-op mock log dependency - unary {[m]} matches kx.log contract
 before,0,0,q,logdep:enlist[`log]!enlist mocklog,1,1,log dep dict
 before,0,0,q,kafkaterrored:0b,1,1,flag used to prove enabled:0b branch was skipped not merely unsuccessful
-before,0,0,q,capturelog:`info`warn`error!({[c;m]};{[c;m]};{[c;m] `kafkaterrored set 1b}),1,1,capturing mock logger - error fn uses set builtin; only reliable when not called from within module execution context
-
-/ logdict extraction - invalid deps inputs throw correct message
-true,0,0,q,.[kafka.init;(()!();()!());{x}] like "*log dependency*",1,1,empty dict deps throws correct message
-true,0,0,q,.[kafka.init;(()!();(::));{x}] like "*log dependency*",1,1,:: as deps throws correct message
-true,0,0,q,.[kafka.init;(()!();42);{x}] like "*log dependency*",1,1,non-dict deps throws correct message
-true,0,0,q,.[kafka.init;(()!();enlist[`log]!enlist(::));{x}] like "*log dependency*",1,1,log value of :: throws correct message
-true,0,0,q,.[kafka.init;(()!();enlist[`notlog]!enlist mocklog);{x}] like "*log dependency*",1,1,deps without log key throws correct message
-true,0,0,q,.[kafka.init;(()!();enlist[`log]!enlist `info`warn!(mocklog`info;mocklog`warn));{x}] like "*log dependency*",1,1,partial log dict missing error key throws correct message
+before,0,0,q,capturelog:`info`warn`error!({[m]};{[m]};{[m] `kafkaterrored set 1b}),1,1,capturing mock logger - error fn uses set builtin; only reliable when not called from within module execution context
+
+/ log dep is optional - invalid or absent deps fall back to no-op logger without error
+run,0,0,q,kafka.init[()!();()!()],1,1,no log dep provided - falls back to defaultlog no-ops without error
+run,0,0,q,kafka.init[()!();42],1,1,non-dict deps - falls back gracefully without error
+run,0,0,q,kafka.init[()!();enlist[`notlog]!enlist mocklog],1,1,missing log key in deps - falls back gracefully without error
 
 / init disabled - stubs throw with correct message
 run,0,0,q,kafka.init[()!();logdep],1,1,init with empty config does not error

From c4f9d99f9230791d815ec5ae33ccc4017176902c Mon Sep 17 00:00:00 2001
From: alowrydi <aaron.lowry@dataintellect.com>
Date: Tue, 23 Jun 2026 18:54:09 +0100
Subject: [PATCH 05/10] Refactoring kafka module with newly consolidated
 standards outlined in initial team review. Required log dep, kx.log
 ergonomics, full test coverage

---
 di/kafka/init.q   |   7 +-
 di/kafka/kafka.md | 223 +++++++++++++++++-----------------------------
 di/kafka/kafka.q  | 151 ++++++++++++++++---------------
 di/kafka/test.csv | 102 +++++++++++----------
 4 files changed, 223 insertions(+), 260 deletions(-)

diff --git a/di/kafka/init.q b/di/kafka/init.q
index eac93cc8..fc6c2109 100644
--- a/di/kafka/init.q
+++ b/di/kafka/init.q
@@ -1,5 +1,6 @@
-/ kafka consumer/producer interface - wraps kafkaq native library
-
+/ kafka module - wrapper around the kafkaq native library for producer/consumer operations
 \l ::kafka.q
 
-export:([init;initconsumer;initproducer;cleanupconsumer;cleanupproducer;subscribe;publish;setkupd])
+version:"0.1.0";
+
+export:([init;initconsumer;initproducer;cleanupconsumer;cleanupproducer;subscribe;publish;setkupd;version])
diff --git a/di/kafka/kafka.md b/di/kafka/kafka.md
index 19909298..01045130 100644
--- a/di/kafka/kafka.md
+++ b/di/kafka/kafka.md
@@ -1,205 +1,144 @@
 # di.kafka
 
-Kafka consumer and producer interface for kdb+ processes. Wraps the `kafkaq` native C shared library to provide consumer initialisation, producer initialisation, topic subscription, and message publishing. Incoming messages are delivered to a configurable `kupd` callback in the root namespace.
-
----
-
-## Features
-
-- Connect kdb+ processes to a Kafka broker as consumer, producer, or both
-- Subscribe to topics and receive messages via a configurable callback function
-- Publish byte vectors to topics with optional message keys
-- Graceful disabled mode on unsupported platforms - all native functions are replaced with informative stubs
-- Message handler (`kupd`) configurable at init or updated live via `setkupd`
-- Compatible with `kx.log` and any logger that exposes unary `info`/`warn`/`error` functions
-
----
+Wrapper around the `kafkaq` native shared library. Provides consumer and producer
+lifecycle management and a configurable message callback for kdb-x processes that
+need to produce or consume Kafka messages.
 
 ## Dependencies
 
-**Native library:** `kafkaq.so` (Linux) or `kafkaq.dll` (Windows). Must be present on the host. Path is provided via the `libpath` config key or derived from the `KDBLIB` environment variable.
-
-**Injected:** `log` - optional. Pass a `kx.log` logger instance (from `kx.log.createLog[]`) under the `log` key in `deps`. If absent, or if the instance does not provide `info`/`warn`/`error` functions, the module falls back to no-op logging.
-
----
+| Dependency | Key | Required | Description |
+|---|---|---|---|
+| logger | `log` | yes | `info`, `warn`, `error` — each monadic `{[msg] ...}` |
 
-## Initialisation
+The `log` dependency must be passed to `init`. The module throws immediately if it
+is absent or missing any of the three required keys. A `kx.log` instance can be
+passed directly — no manual wrapping required:
 
 ```q
 kxlog:use`kx.log
-logger:kxlog.createLog[]
-
 kafka:use`di.kafka
-kafka.init[`enabled`libpath!(1b;`$/opt/TorQ/lib/l64/kafkaq);enlist[`log]!enlist logger]
+kafka.init[enlist[`libpath]!enlist`$/opt/kdb/lib;kxlog.createLog[]]
 ```
 
-All config keys are optional. Passing `(::)` or `()!()` as config applies defaults.
+Alternatively, pass a deps dict with a `log` key for use alongside other dependencies:
 
-| Key | Type | Default | Description |
-|---|---|---|---|
-| `enabled` | boolean | `1b` on `l64`, `0b` on all other platforms | Whether to load the native library. On a non-`l64` host the module loads in disabled mode unless explicitly overridden. |
-| `libpath` | symbol | derived from `KDBLIB` env var | Path to the kafkaq library **without** file extension. If `KDBLIB` is not set, provide this explicitly. |
-| `kupd` | function | logs message via injected logger | Message handler called by the C library on Kafka message receipt. Signature: `{[key;bytes]}`. |
-
-If no `log` dep is provided, or it is missing `info`/`warn`/`error` functions, the module falls back to no-op logging silently.
-
-`init` must be called before any other function is used. It is safe to call multiple times.
-
----
-
-## Functions
-
-### `init`
 ```q
-kafka.init[config;deps]
+kafka.init[enlist[`libpath]!enlist`$/opt/kdb/lib;enlist[`log]!enlist kxlog.createLog[]]
 ```
-Initialises the module. Loads the native library when `enabled:1b` and the library is found. Sets the global `kupd` callback for the C library.
 
-### `initconsumer`
-```q
-kafka.initconsumer[`localhost:9092;`fetch.wait.max.ms`fetch.error.backoff.ms!`5`5]
-```
-Initialises a Kafka consumer and connects to the broker.
+## Configuration
 
-### `initproducer`
-```q
-kafka.initproducer[`localhost:9092;`queue.buffering.max.ms`batch.num.messages!`5`1]
-```
-Initialises a Kafka producer and connects to the broker.
+`init[config;deps]` takes a configuration dictionary as its first argument.
 
-### `cleanupconsumer`
-```q
-kafka.cleanupconsumer[(::)]
-```
-Disconnects and frees the consumer object, stopping the subscription thread.
+| Key | Required | Description |
+|---|---|---|
+| `libpath` | yes (when enabled) | Root library directory — the OS-specific subdirectory and `kafkaq` filename are appended automatically (e.g. `/opt/kdb/lib` → `/opt/kdb/lib/l64/kafkaq.so`) |
+| `kupd` | no | Initial message callback `{[k;x]}` — defaults to printing bytes as chars to stdout |
+| `enabled` | no | Whether to load the native library. Defaults to `1b` on `l64`, `0b` elsewhere. Set to `0b` to suppress library loading on unsupported platforms — all native functions remain stubs |
 
-### `cleanupproducer`
-```q
-kafka.cleanupproducer[(::)]
-```
-Disconnects and frees the producer object.
+## The message callback
 
-### `subscribe`
-```q
-kafka.subscribe[`mytopic;0]
-```
-Starts the subscription thread for a topic on a given partition. Messages are delivered to `kupd`.
+When a subscription is active, the native `kafkaq` library delivers messages by
+calling `.kupd` in the root namespace with a key (symbol) and payload (bytes).
+`init` installs a forwarder at `.kupd` that delegates to the module-local callback,
+which can be replaced at any time via `setkupd` without re-initialising.
 
-### `publish`
-```q
-kafka.publish[`mytopic;0;`;`byte$"hello world"]
-kafka.publish[`mytopic;0;`mykey;`byte$"hello world"]
-```
-Publishes a byte vector to a topic and partition with an optional symbol key.
+> **Warning:** Avoid calling `setkupd` while a subscription is active. The Kafka
+> C library delivers messages on a background thread and may invoke the old handler
+> after the swap.
 
-### `setkupd`
 ```q
-kafka.setkupd[{[k;x] upd[`kafkadata;(enlist k;enlist x)]}]
+kafka.setkupd[{[k;x] upd[`kafkadata;(enlist .z.p;enlist k;enlist "c"$x)]}]
 ```
-Updates the message handler after `init` without requiring a full reinitialisation. Also updates the global `kupd` in the root namespace if the module is enabled.
 
-`init` must be called before `setkupd`.
+## Public API
+
+| Function | Description |
+|---|---|
+| `init[config;deps]` | Load the native library, wire dependencies, install the message callback |
+| `initconsumer[server;optiondict]` | Initialise a Kafka consumer |
+| `initproducer[server;optiondict]` | Initialise a Kafka producer |
+| `cleanupconsumer[]` | Disconnect and free the consumer |
+| `cleanupproducer[]` | Disconnect and free the producer |
+| `subscribe[topic;partition]` | Start the subscription thread for a topic/partition |
+| `publish[topic;partition;key;msg]` | Publish a byte vector to a topic/partition |
+| `setkupd[f]` | Replace the message callback |
+
+## Before `init` is called
 
-> **Warning:** Avoid calling `setkupd` while a subscription is active. The Kafka C library delivers messages on a background thread and may invoke the old handler after the swap.
+All six native functions (`initconsumer`, `initproducer`, `cleanupconsumer`,
+`cleanupproducer`, `subscribe`, `publish`) are stubs that throw:
+'di.kafka: kafka not initialised - call init first
 
----
+This distinguishes "init not called" from a genuine broker or argument error.
 
-## Usage example
+## Example
 
 ```q
 kxlog:use`kx.log
-logger:kxlog.createLog[]
-
-myhandler:{[k;x]
-  upd[`kafkadata;(enlist .z.p;enlist k;enlist "c"$x)];
-  };
 
 kafka:use`di.kafka
-kafka.init[`enabled`libpath`kupd!(1b;`$/opt/TorQ/lib/l64/kafkaq;myhandler);enlist[`log]!enlist logger]
+kafka.init[enlist[`libpath]!enlist`$/opt/kdb/lib;kxlog.createLog[]]
 
-kafka.initconsumer[`localhost:9092;()!()]
+/ consume messages with a custom callback
+kafka.setkupd[{[k;x] upd[`kafkadata;(enlist .z.p;enlist k;enlist "c"$x)]}]
+kafka.initconsumer[`localhost:9092;`fetch.wait.max.ms`fetch.error.backoff.ms!`5`5]
 kafka.subscribe[`trades;0]
 
-kafka.initproducer[`localhost:9092;()!()]
-kafka.publish[`trades;0;`;`byte$"test message"]
-
-kafka.cleanupconsumer[(::)]
-kafka.cleanupproducer[(::)]
-```
-
----
-
-## Global side effect
-
-When `enabled:1b` and the library loads successfully, `init` sets `kupd` in the **root namespace**:
-
-```q
-@[`.;`kupd;:;.z.m.kupd]
-```
-
-This is an unavoidable consequence of the native C library's design - kafkaq is compiled to call `kupd` by name in the root namespace. `setkupd` also updates the global when enabled.
-
----
-
-## Disabled mode
-
-When `enabled:0b` - either because the platform default resolved to `0b` or it was explicitly set - every native function (`initconsumer`, `initproducer`, `cleanupconsumer`, `cleanupproducer`, `subscribe`, `publish`) is a stub that throws:
+/ publish a message
+kafka.initproducer[`localhost:9092;`queue.buffering.max.ms`batch.num.messages!`5`1]
+kafka.publish[`trades;0;`;`byte$"hello world"]
 
+/ cleanup
+kafka.cleanupconsumer[]
+kafka.cleanupproducer[]
 ```
-'kafka not enabled
-```
-
-This is the expected behaviour when kafka is not set up. It distinguishes "kafka isn't configured" from a genuine broker or argument error.
-
-If `enabled:1b` and the library file is not found, `init` logs an error and continues rather than throwing - the process stays alive and all functions remain as stubs. This matches TorQ's original behaviour: a missing native library degrades the module to disabled rather than halting the process.
-
-> **Note:** If kafka silently isn't working after `init` with `enabled:1b`, check the injected logger's error output. `init` logs the missing library path but does not throw - there is nothing to catch.
-
----
 
 ## TorQ migration
 
 | TorQ pattern | Module equivalent |
 |---|---|
-| `enabled:@[value;\`enabled;.z.o in \`l64]` | `kafka.init[\`enabled!enlist 1b;logdep]` |
-| `kupd:{[k;x]...}` defined in settings | `kafka.init[\`kupd!enlist{[k;x]...};logdep]` |
-| default `kupd` prints bytes to stdout (`-1 \`char$x`) | default `kupd` routes through injected `kx.log` logger (unary call) - override with `kupd` config key for custom behaviour |
+| `enabled:@[value;\`enabled;.z.o in \`l64]` | pass `enabled:0b` in config to suppress library loading |
+| `kupd:{[k;x]...}` defined in settings | pass `kupd` key in config to `init`, or call `setkupd` after |
+| default `kupd` prints bytes to stdout | default `kupd` prints bytes to stdout — same behaviour |
 | `.kafka.initconsumer[s;o]` | `kafka.initconsumer[s;o]` |
 | `.kafka.initproducer[s;o]` | `kafka.initproducer[s;o]` |
-| `.kafka.cleanupconsumer[(::)]` | `kafka.cleanupconsumer[(::)]` |
-| `.kafka.cleanupproducer[(::)]` | `kafka.cleanupproducer[(::)]` |
+| `.kafka.cleanupconsumer[(::)]` | `kafka.cleanupconsumer[]` |
+| `.kafka.cleanupproducer[(::)]` | `kafka.cleanupproducer[]` |
 | `.kafka.subscribe[t;p]` | `kafka.subscribe[t;p]` |
 | `.kafka.publish[t;p;k;m]` | `kafka.publish[t;p;k;m]` |
 
----
+## Global side effect
+
+`init` sets `.kupd` in the root namespace as a forwarder into module state:
 
-## Manual verification (requires kafkaq.so and Kafka broker)
+```q
+`.kupd set {[k;x] .z.m.kupd[k;x]}
+```
+
+This is an unavoidable consequence of the native C library's design — `kafkaq` is
+compiled to call `kupd` by name in the root namespace.
+
+## Manual verification (requires kafkaq.so and a Kafka broker)
 
 ```q
 kxlog:use`kx.log
-logger:kxlog.createLog[]
-logdep:enlist[`log]!enlist logger
 
-/ verify lib loads
-kafka.init[`enabled`libpath!(1b;`$/path/to/l64/kafkaq);logdep]
+kafka.init[enlist[`libpath]!enlist`$/path/to/lib;kxlog.createLog[]]
 
-/ verify consumer and subscription
 kafka.initconsumer[`localhost:9092;()!()]
 kafka.subscribe[`test;0]
 
-/ verify publish
 kafka.initproducer[`localhost:9092;()!()]
 kafka.publish[`test;0;`;`byte$"hello from kdb+"]
 
-/ cleanup
-kafka.cleanupconsumer[(::)]
-kafka.cleanupproducer[(::)]
+kafka.cleanupconsumer[]
+kafka.cleanupproducer[]
 ```
 
----
-
-## Notes
+## Running tests
 
-- `enabled` defaults to `1b` only on `l64` hosts; everywhere else it defaults to `0b`. Pass `enabled:1b` explicitly to force-enable on a platform where a native library build is available.
-- The `log` dependency is optional. Pass a `kx.log` logger instance for production observability; omit it for standalone or test use. Logger functions use a unary `{[msg]}` signature — context is embedded in the message string (e.g. `"kafka: message"`).
-- The default `kupd` routes through the injected logger rather than TorQ's original raw stdout write (`-1 \`char$x`). Override with the `kupd` config key if custom behaviour is needed.
+```q
+k4unit:use`di.k4unit
+k4unit.moduletest`di.kafka
+```
diff --git a/di/kafka/kafka.q b/di/kafka/kafka.q
index 58794db7..18d7fd68 100644
--- a/di/kafka/kafka.q
+++ b/di/kafka/kafka.q
@@ -1,93 +1,104 @@
-/ kafka consumer/producer interface - wraps kafkaq native library
-
-/ default message handler - routes message through injected logger
-/ safe to reference .z.m.loginfo here: kupd can only fire after initconsumer+subscribe, both of which require init
-defaultkupd:{[k;x].z.m.loginfo["kafka: kupd: ","c"$x]};
-
-/ default lib path derived from KDBLIB env var and os string
-/ override with libpath config key if KDBLIB is not set in the environment
-defaultlib:`$getenv[`KDBLIB],"/",string[.z.o],"/kafkaq";
-
-/ default no-op logger - used when no log dep is injected
-defaultlog:`info`warn`error!({[m]};{[m]};{[m]});
+/ kafka - wrapper around the kafkaq native library
+/ provides consumer and producer lifecycle management plus a configurable message callback
+/ the native library calls .kupd in the root namespace on message receipt; init sets a forwarder
+/ into .z.m.kupd so the callback can be replaced at runtime via setkupd without re-initialising
+/ the log dependency is required - init errors immediately if absent or malformed
+/ log functions are monadic {[msg]} loggers; a kx.log instance satisfies the contract
+
+/ default message callback - prints message bytes as chars to stdout
+/ safe to use stdout here: kupd can only fire after initconsumer+subscribe, both require init
+/ override via setkupd or by passing kupd in the config dict to init
+defaultkupd:{[k;x] -1 `char$x;};
+
+/ configuration defaults
+kupd:defaultkupd;
+enabled:.z.o in `l64;
 
 / ============================================================
-/ module state and defaults
+/ native function stubs - replaced by bindfunctions when library loads
 / ============================================================
 
-/ whether native library is loaded - overwritten by init; true by default on l64
-enabled:.z.o in `l64;
-
-/ path to kafkaq library without file extension - overwritten by init
-lib:defaultlib;
-
-/ message handler called by C library on message receipt - overwritten by init
-kupd:defaultkupd;
+initconsumer:{[s;o]'"di.kafka: kafka not initialised - call init first"};
+initproducer:{[s;o]'"di.kafka: kafka not initialised - call init first"};
+cleanupconsumer:{[h]'"di.kafka: kafka not initialised - call init first"};
+cleanupproducer:{[h]'"di.kafka: kafka not initialised - call init first"};
+subscribe:{[t;p]'"di.kafka: kafka not initialised - call init first"};
+publish:{[t;p;k;m]'"di.kafka: kafka not initialised - call init first"};
 
 / ============================================================
-/ native function stubs - replaced by init when enabled:1b and library loads
+/ internal functions
 / ============================================================
 
-/ initialise consumer with broker address and option dictionary
-initconsumer:{[s;o]'"kafka not enabled"};
-
-/ initialise producer with broker address and option dictionary
-initproducer:{[s;o]'"kafka not enabled"};
-
-/ disconnect and free consumer object and stop subscription thread - rank-1 matches C lib 2:(`cleanupconsumer;1)
-cleanupconsumer:{[x]'"kafka not enabled"};
+setdeps:{[deps]
+  / accept a bare kx.log instance (info/warn/error at top level) or a full deps dict keyed on `log
+  d:$[(99h=type deps)and not `log in key deps;enlist[`log]!enlist deps;deps];
+  logval:$[99h=type d;$[`log in key d;d`log;(::)];(::)];
+  if[not $[99h=type logval;all `info`warn`error in key logval;0b];
+    '"di.kafka: log dep required - pass a kx.log instance or `info`warn`error!(infofn;warnfn;errfn) keyed on `log"];
+  .z.m.log:logval;
+  };
 
-/ disconnect and free producer object - rank-1 matches C lib 2:(`cleanupproducer;1)
-cleanupproducer:{[x]'"kafka not enabled"};
+setconfig:{[config]
+  / apply recognised configuration overrides on top of current defaults; returns normalised cfg dict
+  cfg:$[99h=type config;config;()!()];
+  if[`enabled in key cfg; .z.m.enabled:cfg`enabled];
+  if[`libpath in key cfg; .z.m.lib:cfg`libpath];
+  if[`kupd in key cfg; .z.m.kupd:cfg`kupd];
+  cfg
+  };
 
-/ start subscription thread for topic on partition - messages delivered to kupd
-subscribe:{[t;p]'"kafka not enabled"};
+loadlib:{[libpath]
+  / resolve and load the native kafkaq shared library from the configured path
+  / the os-appropriate extension (.so or .dll) is appended automatically
+  lib:`$string[libpath],"/",string[.z.o],"/kafkaq";
+  libfile:hsym ` sv lib,$[.z.o like "w*";`dll;`so];
+  libexists:@[{not ()~key x};libfile;{0b}];
+  if[not libexists;
+    .z.m.log[`error]["kafka: native library not found at ",string libfile];
+    '"di.kafka: native library not found at ",string libfile];
+  .z.m.log[`info]["kafka: loading library ",string libfile];
+  .z.m.lib:lib;
+  };
 
-/ publish byte vector to topic and partition with given key
-publish:{[t;p;k;m]'"kafka not enabled"};
+bindfunctions:{[]
+  / bind the six c functions from the loaded kafkaq library into module-local state
+  / overwrites the stubs defined at module level
+  .z.m.initconsumer:.z.m.lib 2:(`initconsumer;2);
+  .z.m.initproducer:.z.m.lib 2:(`initproducer;2);
+  .z.m.cleanupconsumer:.z.m.lib 2:(`cleanupconsumer;1);
+  .z.m.cleanupproducer:.z.m.lib 2:(`cleanupproducer;1);
+  .z.m.subscribe:.z.m.lib 2:(`subscribe;2);
+  .z.m.publish:.z.m.lib 2:(`publish;4);
+  };
 
 / ============================================================
 / public api
 / ============================================================
 
 setkupd:{[f]
-  / update message handler; propagate to global kupd for C callback if enabled
+  / replace the message callback invoked when a subscribed message arrives
+  / f must be a binary function {[k;x]} where k is the message key (symbol) and x is the payload (bytes)
+  / the root .kupd forwarder always delegates to the current value - swap takes effect immediately
+  / note: messages in-flight from the c background thread may briefly invoke the previous handler
   .z.m.kupd:f;
-  if[.z.m.enabled;@[`.;`kupd;:;f]];
   };
 
 init:{[config;deps]
-  / config: dict with optional keys `enabled`libpath`kupd
-  / deps: optional. `log key should be a kx.log logger instance (from kx.log.createLog[])
-  /   if absent or missing info/warn/error keys, falls back to no-op logger
-  / extract deps`log safely; type check before key avoids crash on non-dict lograw (e.g. deps=(::))
-  lograw:$[99h=type deps;@[deps;`log;{(::)}];(::)];
-  logdict:$[99h=type lograw;$[all `info`warn`error in key lograw;lograw;defaultlog];defaultlog];
-  .z.m.loginfo:logdict`info;
-  .z.m.logwarn:logdict`warn;
-  .z.m.logerr:logdict`error;
-  / normalise config - handles (::) and ()!() identically
-  cfg:$[99h=type config;config;()!()];
-  .z.m.enabled:$[`enabled in key cfg;cfg`enabled;.z.o in `l64];
-  .z.m.lib:$[`libpath in key cfg;cfg`libpath;defaultlib];
-  .z.m.kupd:$[`kupd in key cfg;cfg`kupd;defaultkupd];
-  if[.z.m.enabled;
-    libfile:hsym ` sv .z.m.lib,$[.z.o like "w*";`dll;`so];
-    / protected key - kdbx throws on paths with non-existent ancestors
-    libexists:@[{not ()~key x};libfile;{0b}];
-    if[not libexists;
-      .z.m.logerr["kafka: no such file ",1_string libfile]
-      ];
-    if[libexists;
-      .z.m.initconsumer:.z.m.lib 2:(`initconsumer;2);
-      .z.m.initproducer:.z.m.lib 2:(`initproducer;2);
-      .z.m.cleanupconsumer:.z.m.lib 2:(`cleanupconsumer;1);
-      .z.m.cleanupproducer:.z.m.lib 2:(`cleanupproducer;1);
-      .z.m.subscribe:.z.m.lib 2:(`subscribe;2);
-      .z.m.publish:.z.m.lib 2:(`publish;4);
-      / set global kupd - unavoidable side effect of native library design
-      @[`.;`kupd;:;.z.m.kupd];
-      .z.m.loginfo["kafka: kupd set to ",-3!.z.m.kupd];
-      ];
+  / initialise the kafka module - validate deps, apply config, load native library
+  / config: required dict with `libpath (root library directory - os subdirectory is appended automatically)
+  /         optionally `kupd to set a custom message callback
+  / deps:   kx.log instance (passed directly) or dict with `log -> `info`warn`error!(infofn;warnfn;errfn)
+  / example:
+  /   kafka.init[enlist[`libpath]!enlist`$/opt/kdb/lib;kxlog.createLog[]]
+  setdeps deps;
+  cfg:setconfig config;
+  if[enabled;
+    if[not `libpath in key cfg;
+      '"di.kafka: config must contain `libpath - root library directory containing the os-specific kafkaq build"];
+    loadlib cfg`libpath;
+    bindfunctions[];
+    / install the root .kupd forwarder so the native library delegates to our configurable callback
+    `.kupd set {[k;x] .z.m.kupd[k;x]};
     ];
+  .z.m.log[$[enabled;`info;`warn]]["kafka: di.kafka initialised",$[enabled;"";" (native library not loaded - platform not supported)"]];
   };
diff --git a/di/kafka/test.csv b/di/kafka/test.csv
index ad4ff6f2..bd6c69cf 100644
--- a/di/kafka/test.csv
+++ b/di/kafka/test.csv
@@ -1,46 +1,58 @@
 action,ms,bytes,lang,code,repeat,minver,comment
-/ pre-test setup
-before,0,0,q,kafka:use`di.kafka,1,1,load di.kafka module
-before,0,0,q,mocklog:`info`warn`error!({[m]};{[m]};{[m]}),1,1,no-op mock log dependency - unary {[m]} matches kx.log contract
-before,0,0,q,logdep:enlist[`log]!enlist mocklog,1,1,log dep dict
-before,0,0,q,kafkaterrored:0b,1,1,flag used to prove enabled:0b branch was skipped not merely unsuccessful
-before,0,0,q,capturelog:`info`warn`error!({[m]};{[m]};{[m] `kafkaterrored set 1b}),1,1,capturing mock logger - error fn uses set builtin; only reliable when not called from within module execution context
-
-/ log dep is optional - invalid or absent deps fall back to no-op logger without error
-run,0,0,q,kafka.init[()!();()!()],1,1,no log dep provided - falls back to defaultlog no-ops without error
-run,0,0,q,kafka.init[()!();42],1,1,non-dict deps - falls back gracefully without error
-run,0,0,q,kafka.init[()!();enlist[`notlog]!enlist mocklog],1,1,missing log key in deps - falls back gracefully without error
-
-/ init disabled - stubs throw with correct message
-run,0,0,q,kafka.init[()!();logdep],1,1,init with empty config does not error
-true,0,0,q,.[kafka.initconsumer;(`localhost:9092;()!());{x}] like "*kafka not enabled*",1,1,initconsumer stub throws correct message
-true,0,0,q,.[kafka.initproducer;(`localhost:9092;()!());{x}] like "*kafka not enabled*",1,1,initproducer stub throws correct message
-true,0,0,q,.[kafka.cleanupconsumer;enlist(::);{x}] like "*kafka not enabled*",1,1,cleanupconsumer stub throws correct message
-true,0,0,q,.[kafka.cleanupproducer;enlist(::);{x}] like "*kafka not enabled*",1,1,cleanupproducer stub throws correct message
-true,0,0,q,.[kafka.subscribe;(`test;0);{x}] like "*kafka not enabled*",1,1,subscribe stub throws correct message
-true,0,0,q,.[kafka.publish;(`test;0;`;0x68656c6c6f);{x}] like "*kafka not enabled*",1,1,publish stub throws correct message
-
-/ init - explicit enabled:0b: capturelog proves lib-loading branch was skipped not merely unsuccessful
-run,0,0,q,kafka.init[enlist[`enabled]!enlist 0b;enlist[`log]!enlist capturelog],1,1,init with enabled:0b does not error
-true,0,0,q,not kafkaterrored,1,1,logerr was not called - lib-loading branch was genuinely skipped
-true,0,0,q,.[kafka.subscribe;(`test;0);{x}] like "*kafka not enabled*",1,1,subscribe throws correct message after explicit disabled init
-
-/ init - :: config
-run,0,0,q,kafka.init[::;logdep],1,1,init with :: config does not error
-
-/ init - enabled:1b with missing lib: process continues with stubs intact
-/ note: logerr is called internally but cannot be captured from within kdbx module execution context
-run,0,0,q,kafka.init[`enabled`libpath!(1b;`fakekafkaq);logdep],1,1,init with enabled:1b and missing lib does not throw
-true,0,0,q,.[kafka.subscribe;(`test;0);{x}] like "*kafka not enabled*",1,1,stubs remain after failed lib load
-
-/ setkupd - state unobservable without live broker; test confirms no throw only
-run,0,0,q,kafka.init[()!();logdep],1,1,reinit with no-op log before setkupd test
-run,0,0,q,kafka.setkupd[{[k;x]}],1,1,setkupd does not error when disabled - internal state unobservable without live broker
-
-/ init - custom kupd
-run,0,0,q,kafka.init[(enlist`kupd)!enlist{[k;x]};logdep],1,1,init with custom kupd does not error
-
-/ multiple consecutive init calls do not corrupt state
-run,0,0,q,kafka.init[()!();logdep],1,1,second consecutive init does not error
-run,0,0,q,kafka.init[enlist[`enabled]!enlist 0b;logdep],1,1,third consecutive init with enabled:0b does not error
-true,0,0,q,.[kafka.subscribe;(`test;0);{x}] like "*kafka not enabled*",1,1,stubs remain correct after multiple init calls
+comment,,,,,,,setup - load module and inject mock dependencies
+before,0,0,q,kafka:use`di.kafka,1,1,load the kafka module
+before,0,0,q,logdep:`info`warn`error!({[m]};{[m]};{[m]}),1,1,silent no-op log mock - monadic {[m]} matches the kx.log contract
+before,0,0,q,caplog:`info`warn`error!({[m] `.test.cap upsert(`info;m)};{[m] `.test.cap upsert(`warn;m)};{[m] `.test.cap upsert(`error;m)}),1,1,capturing log mock - records (level;msg) for assertion
+before,0,0,q,.test.cap:([] fn:`symbol$();msg:()),1,1,initialise log capture table
+before,0,0,q,.test.kupdk:`,1,1,last key received by the test kupd callback
+before,0,0,q,.test.kupdx:`byte$(),1,1,last payload received by the test kupd callback
+comment,,,,,,,init - dep validation
+fail,0,0,q,kafka.init[enlist[`libpath]!enlist"/tmp";(::)],1,1,errors when deps is not a dictionary
+fail,0,0,q,kafka.init[enlist[`libpath]!enlist"/tmp";()!()],1,1,errors when log dependency is missing
+fail,0,0,q,kafka.init[enlist[`libpath]!enlist"/tmp";enlist[`log]!enlist 42],1,1,errors when log value is not a dictionary
+fail,0,0,q,kafka.init[enlist[`libpath]!enlist"/tmp";enlist[`log]!enlist `info`warn!(logdep`info;logdep`warn)],1,1,errors when log dict is missing the error key
+run,0,0,q,.test.err:@[{kafka.init[()!();()!()]};(::);{x}],1,1,capture error string from init with no deps
+true,0,0,q,.test.err like "di.kafka:*",1,1,error is prefixed di.kafka:
+comment,,,,,,,init - config validation
+fail,0,0,q,kafka.init[()!();enlist[`log]!enlist logdep],1,1,errors when libpath is missing from config
+comment,,,,,,,init - disabled platform (enabled:0b) - native library skipped; libpath not required
+run,0,0,q,kafka.init[enlist[`enabled]!enlist 0b;enlist[`log]!enlist logdep],1,1,init with enabled:0b succeeds without libpath
+true,0,0,q,not .m.di.0kafka.enabled,1,1,module enabled flag reflects disabled state
+comment,,,,,,,stubs - functions throw before init loads the library
+true,0,0,q,.[kafka.initconsumer;(`localhost:9092;()!());{x}] like "*kafka not initialised*",1,1,initconsumer stub throws before init
+true,0,0,q,.[kafka.initproducer;(`localhost:9092;()!());{x}] like "*kafka not initialised*",1,1,initproducer stub throws before init
+true,0,0,q,.[kafka.subscribe;(`test;0);{x}] like "*kafka not initialised*",1,1,subscribe stub throws before init
+true,0,0,q,.[kafka.publish;(`test;0;`;0x68656c6c6f);{x}] like "*kafka not initialised*",1,1,publish stub throws before init
+true,0,0,q,.[kafka.cleanupconsumer;enlist 0;{x}] like "*kafka not initialised*",1,1,cleanupconsumer stub throws before init
+true,0,0,q,.[kafka.cleanupproducer;enlist 0;{x}] like "*kafka not initialised*",1,1,cleanupproducer stub throws before init
+comment,,,,,,,setkupd - replaces the callback stored in module state
+run,0,0,q,kafka.setkupd[{[k;x] .test.kupdk:k;.test.kupdx:x}],1,1,replace the callback via setkupd
+true,0,0,q,.m.di.0kafka.kupd~{[k;x] .test.kupdk:k;.test.kupdx:x},1,1,module-local kupd replaced by setkupd
+comment,,,,,,,setkupd - the root .kupd forwarder delegates into module state (wired without loading the native lib)
+before,0,0,q,.m.di.0kafka.kupd:{[k;x] .test.kupdk:k;.test.kupdx:x},1,1,set a capturing callback directly in module state
+before,0,0,q,`.kupd set {[k;x] .m.di.0kafka.kupd[k;x]},1,1,install the forwarder directly as init would without loading the native lib
+run,0,0,q,.test.kupdk:`;.test.kupdx:`byte$(),1,1,reset capture
+run,0,0,q,.kupd[`testkey;`byte$"hello"],1,1,simulate the native library delivering a message
+true,0,0,q,`testkey~.test.kupdk,1,1,forwarder passed the key through to the module callback
+true,0,0,q,(`byte$"hello")~.test.kupdx,1,1,forwarder passed the payload through to the module callback
+comment,,,,,,,setkupd - replacing callback via setkupd is reflected in the forwarder immediately
+run,0,0,q,kafka.setkupd[{[k;x] .test.kupdk:`replaced;.test.kupdx:x}],1,1,replace the callback
+run,0,0,q,.test.kupdk:`,1,1,reset capture
+run,0,0,q,.kupd[`;`byte$"x"],1,1,deliver a message through the forwarder
+true,0,0,q,`replaced~.test.kupdk,1,1,new callback invoked - forwarder always delegates to current .z.m.kupd
+comment,,,,,,,log call verification - capturing logger sees init message
+run,0,0,q,.test.cap:0#.test.cap,1,1,reset log capture
+run,0,0,q,.m.di.0kafka.log:caplog,1,1,inject capturing logger directly for log-only tests
+run,0,0,q,.m.di.0kafka.log[`info]["kafka: di.kafka initialised"],1,1,simulate the init log call
+true,0,0,q,1=count select from .test.cap where fn=`info,1,1,one info entry captured
+true,0,0,q,any (.test.cap`msg) like "kafka: *",1,1,log message carries kafka context prefix
+comment,,,,,,,kx.log integration - wire a real kx.log instance and confirm the contract is satisfied
+run,0,0,q,kxlogger:use`kx.log,1,1,load the system kx.log module
+run,0,0,q,kxinst:kxlogger.createLog[],1,1,create a kx.log logger instance
+run,0,0,q,.m.di.0kafka.log:`info`warn`error!(kxinst`info;kxinst`warn;kxinst`error),1,1,inject the real logger
+run,0,0,q,.m.di.0kafka.log[`info]["kafka: kx.log integration test"],1,1,emit a real log message through kx.log
+true,0,0,q,all `info`warn`error in key .m.di.0kafka.log,1,1,kx.log instance satisfies the required contract
+comment,,,,,,,setdeps - bare kx.log instance accepted directly as deps (tested via internal setdeps to avoid native lib requirement)
+run,0,0,q,.m.di.0kafka.setdeps[kxinst],1,1,setdeps accepts bare kx.log instance without wrapping
+true,0,0,q,all `info`warn`error in key .m.di.0kafka.log,1,1,bare instance normalised into log dep in module state
+fail,0,0,q,.m.di.0kafka.setdeps[`info`warn!(kxinst`info;kxinst`warn)],1,1,bare instance missing error key is rejected

From f3a5399b34874a6a74a9af6d4cf96bf37c216b16 Mon Sep 17 00:00:00 2001
From: alowrydi <aaron.lowry@dataintellect.com>
Date: Wed, 24 Jun 2026 14:12:10 +0100
Subject: [PATCH 06/10] apply TorQ-modularisation skill to polish up binary
 logger, single-dict init, required log dep, kx.log normalisation

---
 di/kafka/init.q   |  2 +-
 di/kafka/kafka.md | 60 ++++++++++++++++++++--------------
 di/kafka/kafka.q  | 83 ++++++++++++++++++++++++++---------------------
 di/kafka/test.csv | 43 +++++++++++-------------
 4 files changed, 102 insertions(+), 86 deletions(-)

diff --git a/di/kafka/init.q b/di/kafka/init.q
index fc6c2109..bc7f6907 100644
--- a/di/kafka/init.q
+++ b/di/kafka/init.q
@@ -3,4 +3,4 @@
 
 version:"0.1.0";
 
-export:([init;initconsumer;initproducer;cleanupconsumer;cleanupproducer;subscribe;publish;setkupd;version])
+export:([init;initconsumer;initproducer;cleanupconsumer;cleanupproducer;subscribe;publish;setkupd])
diff --git a/di/kafka/kafka.md b/di/kafka/kafka.md
index 01045130..3776f8d4 100644
--- a/di/kafka/kafka.md
+++ b/di/kafka/kafka.md
@@ -8,32 +8,43 @@ need to produce or consume Kafka messages.
 
 | Dependency | Key | Required | Description |
 |---|---|---|---|
-| logger | `log` | yes | `info`, `warn`, `error` — each monadic `{[msg] ...}` |
+| logger | `log` | yes | `info`, `warn`, `error` — each binary `{[c;m]}` where `c` is a symbol context and `m` is a string |
 
-The `log` dependency must be passed to `init`. The module throws immediately if it
-is absent or missing any of the three required keys. A `kx.log` instance can be
-passed directly — no manual wrapping required:
+The `log` dependency must be passed to `init` inside the `configs` dict. The module
+throws immediately if it is absent or missing any of the three required keys.
+
+A `kx.log` instance can be passed directly — the module normalises monadic functions
+to the binary `{[c;m]}` contract automatically. Context is embedded in the output as
+`"context: message"` (e.g. `"kafka: loading library /opt/kdb/lib/l64/kafkaq.so"`):
 
 ```q
 kxlog:use`kx.log
 kafka:use`di.kafka
-kafka.init[enlist[`libpath]!enlist`$/opt/kdb/lib;kxlog.createLog[]]
+kafka.init[`log`libpath!(kxlog.createLog[];`$/opt/kdb/lib)]
 ```
 
-Alternatively, pass a deps dict with a `log` key for use alongside other dependencies:
+Alternatively pass a custom binary logger:
 
 ```q
-kafka.init[enlist[`libpath]!enlist`$/opt/kdb/lib;enlist[`log]!enlist kxlog.createLog[]]
+logdep:`info`warn`error!(
+  {[c;m] -1 "INFO  [",string[c],"] ",m;};
+  {[c;m] -1 "WARN  [",string[c],"] ",m;};
+  {[c;m] -2 "ERROR [",string[c],"] ",m;})
+
+kafka:use`di.kafka
+kafka.init[`log`libpath!(logdep;`$/opt/kdb/lib)]
 ```
 
 ## Configuration
 
-`init[config;deps]` takes a configuration dictionary as its first argument.
+`init[configs]` takes a single dictionary combining the `log` dependency with
+any configuration overrides.
 
 | Key | Required | Description |
 |---|---|---|
+| `log` | yes | Binary log dep — `info`, `warn`, `error` functions each `{[c;m]}` |
 | `libpath` | yes (when enabled) | Root library directory — the OS-specific subdirectory and `kafkaq` filename are appended automatically (e.g. `/opt/kdb/lib` → `/opt/kdb/lib/l64/kafkaq.so`) |
-| `kupd` | no | Initial message callback `{[k;x]}` — defaults to printing bytes as chars to stdout |
+| `kupd` | no | Initial message callback `{[k;x]}` — defaults to a no-op; replace via `setkupd` |
 | `enabled` | no | Whether to load the native library. Defaults to `1b` on `l64`, `0b` elsewhere. Set to `0b` to suppress library loading on unsupported platforms — all native functions remain stubs |
 
 ## The message callback
@@ -53,16 +64,16 @@ kafka.setkupd[{[k;x] upd[`kafkadata;(enlist .z.p;enlist k;enlist "c"$x)]}]
 
 ## Public API
 
-| Function | Description |
-|---|---|
-| `init[config;deps]` | Load the native library, wire dependencies, install the message callback |
-| `initconsumer[server;optiondict]` | Initialise a Kafka consumer |
-| `initproducer[server;optiondict]` | Initialise a Kafka producer |
-| `cleanupconsumer[]` | Disconnect and free the consumer |
-| `cleanupproducer[]` | Disconnect and free the producer |
-| `subscribe[topic;partition]` | Start the subscription thread for a topic/partition |
-| `publish[topic;partition;key;msg]` | Publish a byte vector to a topic/partition |
-| `setkupd[f]` | Replace the message callback |
+| Function | Args | Returns | Description |
+|---|---|---|---|
+| `init[configs]` | dict | void | Load the native library, wire dependencies, install the message callback |
+| `initconsumer[server;optiondict]` | symbol, symbol dict | int (consumer handle) | Initialise a Kafka consumer. `server`: broker address as symbol e.g. `` `localhost:9092 ``. `optiondict`: Kafka config options as symbol-keyed symbol dict e.g. `` `fetch.wait.max.ms`fetch.error.backoff.ms!`5`5 `` — pass `()!()` for defaults |
+| `initproducer[server;optiondict]` | symbol, symbol dict | int (producer handle) | Initialise a Kafka producer. Same arg shapes as `initconsumer` |
+| `cleanupconsumer[]` | — | void | Disconnect and free the consumer |
+| `cleanupproducer[]` | — | void | Disconnect and free the producer |
+| `subscribe[topic;partition]` | symbol, int | void | Start the subscription thread. `topic`: Kafka topic name as symbol. `partition`: partition number as int |
+| `publish[topic;partition;key;msg]` | symbol, int, symbol, bytes | void | Publish to a topic. `key`: message key as symbol (use `` ` `` for no key). `msg`: payload as byte vector e.g. `` `byte$"hello" `` |
+| `setkupd[f]` | binary function | void | Replace the message callback. `f` must be `{[k;x]}` where `k` is the message key (symbol) and `x` is the payload (byte vector) |
 
 ## Before `init` is called
 
@@ -78,7 +89,7 @@ This distinguishes "init not called" from a genuine broker or argument error.
 kxlog:use`kx.log
 
 kafka:use`di.kafka
-kafka.init[enlist[`libpath]!enlist`$/opt/kdb/lib;kxlog.createLog[]]
+kafka.init[`log`libpath!(kxlog.createLog[];`$/opt/kdb/lib)]
 
 / consume messages with a custom callback
 kafka.setkupd[{[k;x] upd[`kafkadata;(enlist .z.p;enlist k;enlist "c"$x)]}]
@@ -98,9 +109,9 @@ kafka.cleanupproducer[]
 
 | TorQ pattern | Module equivalent |
 |---|---|
-| `enabled:@[value;\`enabled;.z.o in \`l64]` | pass `enabled:0b` in config to suppress library loading |
-| `kupd:{[k;x]...}` defined in settings | pass `kupd` key in config to `init`, or call `setkupd` after |
-| default `kupd` prints bytes to stdout | default `kupd` prints bytes to stdout — same behaviour |
+| `enabled:@[value;\`enabled;.z.o in \`l64]` | pass `enabled:0b` in `configs` to suppress library loading |
+| `kupd:{[k;x]...}` defined in settings | pass `kupd` key in `configs` to `init`, or call `setkupd` after |
+| default `kupd` prints bytes to stdout | default `kupd` is a no-op — install a callback via `setkupd` |
 | `.kafka.initconsumer[s;o]` | `kafka.initconsumer[s;o]` |
 | `.kafka.initproducer[s;o]` | `kafka.initproducer[s;o]` |
 | `.kafka.cleanupconsumer[(::)]` | `kafka.cleanupconsumer[]` |
@@ -124,7 +135,8 @@ compiled to call `kupd` by name in the root namespace.
 ```q
 kxlog:use`kx.log
 
-kafka.init[enlist[`libpath]!enlist`$/path/to/lib;kxlog.createLog[]]
+kafka:use`di.kafka
+kafka.init[`log`libpath!(kxlog.createLog[];`$/path/to/lib)]
 
 kafka.initconsumer[`localhost:9092;()!()]
 kafka.subscribe[`test;0]
diff --git a/di/kafka/kafka.q b/di/kafka/kafka.q
index 18d7fd68..e466aedf 100644
--- a/di/kafka/kafka.q
+++ b/di/kafka/kafka.q
@@ -3,12 +3,10 @@
 / the native library calls .kupd in the root namespace on message receipt; init sets a forwarder
 / into .z.m.kupd so the callback can be replaced at runtime via setkupd without re-initialising
 / the log dependency is required - init errors immediately if absent or malformed
-/ log functions are monadic {[msg]} loggers; a kx.log instance satisfies the contract
+/ log functions are binary {[c;m]} where c is a symbol context and m is a string
 
-/ default message callback - prints message bytes as chars to stdout
-/ safe to use stdout here: kupd can only fire after initconsumer+subscribe, both require init
-/ override via setkupd or by passing kupd in the config dict to init
-defaultkupd:{[k;x] -1 `char$x;};
+/ default message callback - no-op; replace with setkupd after init
+defaultkupd:{[k;x] (::)};
 
 / configuration defaults
 kupd:defaultkupd;
@@ -20,8 +18,8 @@ enabled:.z.o in `l64;
 
 initconsumer:{[s;o]'"di.kafka: kafka not initialised - call init first"};
 initproducer:{[s;o]'"di.kafka: kafka not initialised - call init first"};
-cleanupconsumer:{[h]'"di.kafka: kafka not initialised - call init first"};
-cleanupproducer:{[h]'"di.kafka: kafka not initialised - call init first"};
+cleanupconsumer:{'"di.kafka: kafka not initialised - call init first"};
+cleanupproducer:{'"di.kafka: kafka not initialised - call init first"};
 subscribe:{[t;p]'"di.kafka: kafka not initialised - call init first"};
 publish:{[t;p;k;m]'"di.kafka: kafka not initialised - call init first"};
 
@@ -29,22 +27,23 @@ publish:{[t;p;k;m]'"di.kafka: kafka not initialised - call init first"};
 / internal functions
 / ============================================================
 
-setdeps:{[deps]
-  / accept a bare kx.log instance (info/warn/error at top level) or a full deps dict keyed on `log
-  d:$[(99h=type deps)and not `log in key deps;enlist[`log]!enlist deps;deps];
-  logval:$[99h=type d;$[`log in key d;d`log;(::)];(::)];
-  if[not $[99h=type logval;all `info`warn`error in key logval;0b];
-    '"di.kafka: log dep required - pass a kx.log instance or `info`warn`error!(infofn;warnfn;errfn) keyed on `log"];
-  .z.m.log:logval;
+normlog:{[logdict]
+  / detect kx.log instance by presence of kx.log-specific keys (getlvl, sinks, fmts)
+  / kx.log functions are monadic - wrap each into binary {[c;m]} and embed context in the message
+  / plain {[c;m]} log dicts (info`warn`error only) pass through unchanged
+  $[any `getlvl`sinks`fmts in key logdict;
+    `info`warn`error!(
+      {[fn;c;m] fn[string[c],": ",m]}[logdict`info;];
+      {[fn;c;m] fn[string[c],": ",m]}[logdict`warn;];
+      {[fn;c;m] fn[string[c],": ",m]}[logdict`error;]);
+    logdict]
   };
 
-setconfig:{[config]
-  / apply recognised configuration overrides on top of current defaults; returns normalised cfg dict
-  cfg:$[99h=type config;config;()!()];
+setconfig:{[configs]
+  / apply recognised configuration overrides; dep keys (log etc.) are ignored
+  cfg:$[99h=type configs;configs;()!()];
   if[`enabled in key cfg; .z.m.enabled:cfg`enabled];
-  if[`libpath in key cfg; .z.m.lib:cfg`libpath];
   if[`kupd in key cfg; .z.m.kupd:cfg`kupd];
-  cfg
   };
 
 loadlib:{[libpath]
@@ -54,19 +53,22 @@ loadlib:{[libpath]
   libfile:hsym ` sv lib,$[.z.o like "w*";`dll;`so];
   libexists:@[{not ()~key x};libfile;{0b}];
   if[not libexists;
-    .z.m.log[`error]["kafka: native library not found at ",string libfile];
+    .z.m.log[`error][`kafka;"native library not found at ",string libfile];
     '"di.kafka: native library not found at ",string libfile];
-  .z.m.log[`info]["kafka: loading library ",string libfile];
+  .z.m.log[`info][`kafka;"loading library ",string libfile];
   .z.m.lib:lib;
   };
 
 bindfunctions:{[]
   / bind the six c functions from the loaded kafkaq library into module-local state
   / overwrites the stubs defined at module level
+  / cleanupconsumer and cleanupproducer are unary in the c interface but take null (::) - expose as niladic
   .z.m.initconsumer:.z.m.lib 2:(`initconsumer;2);
   .z.m.initproducer:.z.m.lib 2:(`initproducer;2);
-  .z.m.cleanupconsumer:.z.m.lib 2:(`cleanupconsumer;1);
-  .z.m.cleanupproducer:.z.m.lib 2:(`cleanupproducer;1);
+  .z.m.rawcleanupconsumer:.z.m.lib 2:(`cleanupconsumer;1);
+  .z.m.rawcleanupproducer:.z.m.lib 2:(`cleanupproducer;1);
+  .z.m.cleanupconsumer:{rawcleanupconsumer[(::)]};
+  .z.m.cleanupproducer:{rawcleanupproducer[(::)]};
   .z.m.subscribe:.z.m.lib 2:(`subscribe;2);
   .z.m.publish:.z.m.lib 2:(`publish;4);
   };
@@ -83,22 +85,29 @@ setkupd:{[f]
   .z.m.kupd:f;
   };
 
-init:{[config;deps]
+init:{[configs]
   / initialise the kafka module - validate deps, apply config, load native library
-  / config: required dict with `libpath (root library directory - os subdirectory is appended automatically)
-  /         optionally `kupd to set a custom message callback
-  / deps:   kx.log instance (passed directly) or dict with `log -> `info`warn`error!(infofn;warnfn;errfn)
-  / example:
-  /   kafka.init[enlist[`libpath]!enlist`$/opt/kdb/lib;kxlog.createLog[]]
-  setdeps deps;
-  cfg:setconfig config;
+  / configs: dict containing `log (required) plus optional `libpath, `enabled, `kupd
+  / log dep: `info`warn`error!({[c;m]};{[c;m]};{[c;m]}) - binary, c=context symbol, m=string
+  / examples:
+  /   kafka.init[`log`libpath!(logdep;`$/opt/kdb/lib)]
+  /   kafka.init[`log`libpath`enabled!(logdep;`$/opt/kdb/lib;0b)]
+  if[99h<>type configs;
+    '"di.kafka: configs must be a dict with `log key"];
+  if[not `log in key configs;
+    '"di.kafka: log dependency is required; pass `info`warn`error!(infofn;warnfn;errfn) keyed on `log"];
+  if[99h<>type configs`log;
+    '"di.kafka: log value must be a dict; pass `info`warn`error functions"];
+  if[not all `info`warn`error in key configs`log;
+    '"di.kafka: log dict must have `info`warn`error keys; got: ",(", " sv string key configs`log)];
+  .z.m.log:normlog configs`log;
+  setconfig configs;
   if[enabled;
-    if[not `libpath in key cfg;
-      '"di.kafka: config must contain `libpath - root library directory containing the os-specific kafkaq build"];
-    loadlib cfg`libpath;
+    if[not `libpath in key configs;
+      '"di.kafka: configs must contain `libpath - root library directory containing the os-specific kafkaq build"];
+    loadlib configs`libpath;
     bindfunctions[];
-    / install the root .kupd forwarder so the native library delegates to our configurable callback
     `.kupd set {[k;x] .z.m.kupd[k;x]};
-    ];
-  .z.m.log[$[enabled;`info;`warn]]["kafka: di.kafka initialised",$[enabled;"";" (native library not loaded - platform not supported)"]];
+  ];
+  .z.m.log[$[enabled;`info;`warn]][`kafka;"di.kafka initialised",$[enabled;"";" (native library not loaded - platform not supported)"]];
   };
diff --git a/di/kafka/test.csv b/di/kafka/test.csv
index bd6c69cf..801dfc58 100644
--- a/di/kafka/test.csv
+++ b/di/kafka/test.csv
@@ -1,30 +1,30 @@
 action,ms,bytes,lang,code,repeat,minver,comment
 comment,,,,,,,setup - load module and inject mock dependencies
 before,0,0,q,kafka:use`di.kafka,1,1,load the kafka module
-before,0,0,q,logdep:`info`warn`error!({[m]};{[m]};{[m]}),1,1,silent no-op log mock - monadic {[m]} matches the kx.log contract
-before,0,0,q,caplog:`info`warn`error!({[m] `.test.cap upsert(`info;m)};{[m] `.test.cap upsert(`warn;m)};{[m] `.test.cap upsert(`error;m)}),1,1,capturing log mock - records (level;msg) for assertion
+before,0,0,q,logdep:`info`warn`error!({[c;m]};{[c;m]};{[c;m]}),1,1,silent no-op log mock - binary {[c;m]} matches the log contract
+before,0,0,q,caplog:`info`warn`error!({[c;m] `.test.cap upsert(`info;m)};{[c;m] `.test.cap upsert(`warn;m)};{[c;m] `.test.cap upsert(`error;m)}),1,1,capturing log mock - records (level;msg) for assertion
 before,0,0,q,.test.cap:([] fn:`symbol$();msg:()),1,1,initialise log capture table
 before,0,0,q,.test.kupdk:`,1,1,last key received by the test kupd callback
 before,0,0,q,.test.kupdx:`byte$(),1,1,last payload received by the test kupd callback
 comment,,,,,,,init - dep validation
-fail,0,0,q,kafka.init[enlist[`libpath]!enlist"/tmp";(::)],1,1,errors when deps is not a dictionary
-fail,0,0,q,kafka.init[enlist[`libpath]!enlist"/tmp";()!()],1,1,errors when log dependency is missing
-fail,0,0,q,kafka.init[enlist[`libpath]!enlist"/tmp";enlist[`log]!enlist 42],1,1,errors when log value is not a dictionary
-fail,0,0,q,kafka.init[enlist[`libpath]!enlist"/tmp";enlist[`log]!enlist `info`warn!(logdep`info;logdep`warn)],1,1,errors when log dict is missing the error key
-run,0,0,q,.test.err:@[{kafka.init[()!();()!()]};(::);{x}],1,1,capture error string from init with no deps
+fail,0,0,q,kafka.init[(::)],1,1,errors when configs is not a dictionary
+fail,0,0,q,kafka.init[()!()],1,1,errors when log dependency is missing
+fail,0,0,q,kafka.init[enlist[`log]!enlist 42],1,1,errors when log value is not a dictionary
+fail,0,0,q,kafka.init[enlist[`log]!enlist `info`warn!(logdep`info;logdep`warn)],1,1,errors when log dict is missing the error key
+run,0,0,q,.test.err:@[{kafka.init[()!()]};(::);{x}],1,1,capture error string from init with missing log dep
 true,0,0,q,.test.err like "di.kafka:*",1,1,error is prefixed di.kafka:
 comment,,,,,,,init - config validation
-fail,0,0,q,kafka.init[()!();enlist[`log]!enlist logdep],1,1,errors when libpath is missing from config
+fail,0,0,q,kafka.init[enlist[`log]!enlist logdep],1,1,errors when libpath is missing from config
 comment,,,,,,,init - disabled platform (enabled:0b) - native library skipped; libpath not required
-run,0,0,q,kafka.init[enlist[`enabled]!enlist 0b;enlist[`log]!enlist logdep],1,1,init with enabled:0b succeeds without libpath
+run,0,0,q,kafka.init[`log`enabled!(logdep;0b)],1,1,init with enabled:0b succeeds without libpath
 true,0,0,q,not .m.di.0kafka.enabled,1,1,module enabled flag reflects disabled state
 comment,,,,,,,stubs - functions throw before init loads the library
 true,0,0,q,.[kafka.initconsumer;(`localhost:9092;()!());{x}] like "*kafka not initialised*",1,1,initconsumer stub throws before init
 true,0,0,q,.[kafka.initproducer;(`localhost:9092;()!());{x}] like "*kafka not initialised*",1,1,initproducer stub throws before init
 true,0,0,q,.[kafka.subscribe;(`test;0);{x}] like "*kafka not initialised*",1,1,subscribe stub throws before init
 true,0,0,q,.[kafka.publish;(`test;0;`;0x68656c6c6f);{x}] like "*kafka not initialised*",1,1,publish stub throws before init
-true,0,0,q,.[kafka.cleanupconsumer;enlist 0;{x}] like "*kafka not initialised*",1,1,cleanupconsumer stub throws before init
-true,0,0,q,.[kafka.cleanupproducer;enlist 0;{x}] like "*kafka not initialised*",1,1,cleanupproducer stub throws before init
+true,0,0,q,.[kafka.cleanupconsumer;();{x}] like "*kafka not initialised*",1,1,cleanupconsumer stub throws before init
+true,0,0,q,.[kafka.cleanupproducer;();{x}] like "*kafka not initialised*",1,1,cleanupproducer stub throws before init
 comment,,,,,,,setkupd - replaces the callback stored in module state
 run,0,0,q,kafka.setkupd[{[k;x] .test.kupdk:k;.test.kupdx:x}],1,1,replace the callback via setkupd
 true,0,0,q,.m.di.0kafka.kupd~{[k;x] .test.kupdk:k;.test.kupdx:x},1,1,module-local kupd replaced by setkupd
@@ -40,19 +40,14 @@ run,0,0,q,kafka.setkupd[{[k;x] .test.kupdk:`replaced;.test.kupdx:x}],1,1,replace
 run,0,0,q,.test.kupdk:`,1,1,reset capture
 run,0,0,q,.kupd[`;`byte$"x"],1,1,deliver a message through the forwarder
 true,0,0,q,`replaced~.test.kupdk,1,1,new callback invoked - forwarder always delegates to current .z.m.kupd
-comment,,,,,,,log call verification - capturing logger sees init message
+comment,,,,,,,init - kx.log instance normalised automatically (no manual wrapping required)
+run,0,0,q,kxlogger:use`kx.log,1,1,load kx.log
+run,0,0,q,kxinst:kxlogger.createLog[],1,1,create kx.log instance
+run,0,0,q,kafka.init[`log`enabled!(kxinst;0b)],1,1,bare kx.log instance accepted without wrapping
+run,0,0,q,.m.di.0kafka.log[`info][`kafka;"normalisation test"],1,1,normalised logger accepts binary {[c;m]} call without error
+comment,,,,,,,log call verification - capturing logger sees init message with binary {[c;m]} args
 run,0,0,q,.test.cap:0#.test.cap,1,1,reset log capture
 run,0,0,q,.m.di.0kafka.log:caplog,1,1,inject capturing logger directly for log-only tests
-run,0,0,q,.m.di.0kafka.log[`info]["kafka: di.kafka initialised"],1,1,simulate the init log call
+run,0,0,q,.m.di.0kafka.log[`info][`kafka;"di.kafka initialised"],1,1,simulate the init log call with context and message
 true,0,0,q,1=count select from .test.cap where fn=`info,1,1,one info entry captured
-true,0,0,q,any (.test.cap`msg) like "kafka: *",1,1,log message carries kafka context prefix
-comment,,,,,,,kx.log integration - wire a real kx.log instance and confirm the contract is satisfied
-run,0,0,q,kxlogger:use`kx.log,1,1,load the system kx.log module
-run,0,0,q,kxinst:kxlogger.createLog[],1,1,create a kx.log logger instance
-run,0,0,q,.m.di.0kafka.log:`info`warn`error!(kxinst`info;kxinst`warn;kxinst`error),1,1,inject the real logger
-run,0,0,q,.m.di.0kafka.log[`info]["kafka: kx.log integration test"],1,1,emit a real log message through kx.log
-true,0,0,q,all `info`warn`error in key .m.di.0kafka.log,1,1,kx.log instance satisfies the required contract
-comment,,,,,,,setdeps - bare kx.log instance accepted directly as deps (tested via internal setdeps to avoid native lib requirement)
-run,0,0,q,.m.di.0kafka.setdeps[kxinst],1,1,setdeps accepts bare kx.log instance without wrapping
-true,0,0,q,all `info`warn`error in key .m.di.0kafka.log,1,1,bare instance normalised into log dep in module state
-fail,0,0,q,.m.di.0kafka.setdeps[`info`warn!(kxinst`info;kxinst`warn)],1,1,bare instance missing error key is rejected
+true,0,0,q,any (.test.cap`msg) like "di.kafka *",1,1,log message contains module init confirmation

From ba0ba2c10d6a846ceed943a878b8328d63d11c97 Mon Sep 17 00:00:00 2001
From: alowrydi <aaron.lowry@dataintellect.com>
Date: Thu, 25 Jun 2026 18:22:05 +0100
Subject: [PATCH 07/10] Align di.kafka with project standards: fix silent
 production bug in native function dispatch, enforce deps naming convention,
 apply agreed documentation template, add portable integration tests using
 standard TorQ environment

---
 di/kafka/init.q               |   3 +-
 di/kafka/kafka.md             | 193 ++++++++++++++++++----------------
 di/kafka/kafka.q              |  78 ++++++++------
 di/kafka/test.csv             |  10 +-
 di/kafka/test_integration.csv |  19 ++++
 5 files changed, 173 insertions(+), 130 deletions(-)
 create mode 100644 di/kafka/test_integration.csv

diff --git a/di/kafka/init.q b/di/kafka/init.q
index bc7f6907..ad00bfe8 100644
--- a/di/kafka/init.q
+++ b/di/kafka/init.q
@@ -1,6 +1,5 @@
 / kafka module - wrapper around the kafkaq native library for producer/consumer operations
-\l ::kafka.q
 
-version:"0.1.0";
+\l ::kafka.q
 
 export:([init;initconsumer;initproducer;cleanupconsumer;cleanupproducer;subscribe;publish;setkupd])
diff --git a/di/kafka/kafka.md b/di/kafka/kafka.md
index 3776f8d4..4d2f3ff2 100644
--- a/di/kafka/kafka.md
+++ b/di/kafka/kafka.md
@@ -1,89 +1,119 @@
 # di.kafka
 
-Wrapper around the `kafkaq` native shared library. Provides consumer and producer
-lifecycle management and a configurable message callback for kdb-x processes that
-need to produce or consume Kafka messages.
+Wrapper around the `kafkaq` native shared library. Provides consumer and producer lifecycle management and a configurable message callback for kdb-x processes that need to produce or consume Kafka messages.
+
+---
+
+## Features
+
+- Wrap the `kafkaq` native C library and expose consumer and producer lifecycle management as clean kdb-x functions
+- Provide a configurable message callback via `setkupd` — the native library always delegates through the current callback without requiring re-initialisation
+- Detect and normalise `kx.log` instances automatically so callers can pass a logger directly without manual wrapping
+- Gracefully degrade on non-l64 platforms — all native functions remain as informative stubs when `enabled:0b`
+- Distinguish "module not initialised" from genuine broker or argument errors via descriptive stub error messages with `di.kafka:` prefix
+
+---
 
 ## Dependencies
 
 | Dependency | Key | Required | Description |
 |---|---|---|---|
-| logger | `log` | yes | `info`, `warn`, `error` — each binary `{[c;m]}` where `c` is a symbol context and `m` is a string |
+| logger | `` `log `` | yes | `info`, `warn`, `error` — each binary `{[c;m]}` where `c` is a symbol context and `m` is a string |
 
-The `log` dependency must be passed to `init` inside the `configs` dict. The module
-throws immediately if it is absent or missing any of the three required keys.
+The `log` dependency must be passed to `init` inside the `deps` dict. The module throws immediately if it is absent or missing any of the three required keys. All three are required since the module calls `info`, `warn`, and `error`.
 
-A `kx.log` instance can be passed directly — the module normalises monadic functions
-to the binary `{[c;m]}` contract automatically. Context is embedded in the output as
-`"context: message"` (e.g. `"kafka: loading library /opt/kdb/lib/l64/kafkaq.so"`):
+A `kx.log` instance can be passed directly — the module normalises monadic functions to the binary `{[c;m]}` contract automatically via `normlog`. Context is embedded in the output as `"context: message"`:
 
 ```q
 kxlog:use`kx.log
 kafka:use`di.kafka
-kafka.init[`log`libpath!(kxlog.createLog[];`$/opt/kdb/lib)]
-```
 
-Alternatively pass a custom binary logger:
+/ minimal - log and libpath only
+kafka.init[`log`libpath!(kxlog.createLog[];`$/opt/kdb/lib)]
 
-```q
-logdep:`info`warn`error!(
-  {[c;m] -1 "INFO  [",string[c],"] ",m;};
-  {[c;m] -1 "WARN  [",string[c],"] ",m;};
-  {[c;m] -2 "ERROR [",string[c],"] ",m;})
+/ with custom message callback
+kafka.init[`log`libpath`kupd!(kxlog.createLog[];`$/opt/kdb/lib;{[k;x] upd[`kafkadata;(enlist .z.p;enlist k;enlist "c"$x)]})]
 
-kafka:use`di.kafka
-kafka.init[`log`libpath!(logdep;`$/opt/kdb/lib)]
+/ disabled platform - no libpath required
+kafka.init[`log`enabled!(kxlog.createLog[];0b)]
 ```
 
-## Configuration
+---
 
-`init[configs]` takes a single dictionary combining the `log` dependency with
-any configuration overrides.
+## Initialisation
+
+`init[deps]` takes a single dictionary combining the `log` dependency with any configuration overrides.
 
 | Key | Required | Description |
 |---|---|---|
-| `log` | yes | Binary log dep — `info`, `warn`, `error` functions each `{[c;m]}` |
-| `libpath` | yes (when enabled) | Root library directory — the OS-specific subdirectory and `kafkaq` filename are appended automatically (e.g. `/opt/kdb/lib` → `/opt/kdb/lib/l64/kafkaq.so`) |
-| `kupd` | no | Initial message callback `{[k;x]}` — defaults to a no-op; replace via `setkupd` |
-| `enabled` | no | Whether to load the native library. Defaults to `1b` on `l64`, `0b` elsewhere. Set to `0b` to suppress library loading on unsupported platforms — all native functions remain stubs |
+| `` `log `` | yes | Binary log dep — `info`, `warn`, `error` functions each `{[c;m]}` |
+| `` `libpath `` | yes (when enabled) | Root library directory — the OS-specific subdirectory and `kafkaq` filename are appended automatically (e.g. `/opt/kdb/lib` → `/opt/kdb/lib/l64/kafkaq.so`) |
+| `` `kupd `` | no | Initial message callback `{[k;x]}` — defaults to a no-op; replace via `setkupd` after init |
+| `` `enabled `` | no | Whether to load the native library. Defaults to `1b` on `l64`, `0b` elsewhere. Set to `0b` to suppress library loading on unsupported platforms — all native functions remain stubs |
+
+`init` must be called before any native functions are used. Before `init` is called, all six native functions (`initconsumer`, `initproducer`, `cleanupconsumer`, `cleanupproducer`, `subscribe`, `publish`) are stubs that throw:
+
+```
+'di.kafka: kafka not initialised - call init first
+```
 
-## The message callback
+This distinguishes "init not called" from a genuine broker or argument error.
 
-When a subscription is active, the native `kafkaq` library delivers messages by
-calling `.kupd` in the root namespace with a key (symbol) and payload (bytes).
-`init` installs a forwarder at `.kupd` that delegates to the module-local callback,
-which can be replaced at any time via `setkupd` without re-initialising.
+---
 
-> **Warning:** Avoid calling `setkupd` while a subscription is active. The Kafka
-> C library delivers messages on a background thread and may invoke the old handler
-> after the swap.
+## Exported Functions
 
+### `init[deps]`
+Initialise the module. Validates the log dependency, applies config, loads the native library, and installs the root `.kupd` forwarder.
 ```q
-kafka.setkupd[{[k;x] upd[`kafkadata;(enlist .z.p;enlist k;enlist "c"$x)]}]
+kafka.init[`log`libpath!(logdep;`$/opt/kdb/lib)]
 ```
 
-## Public API
+### `initconsumer[server;optiondict]`
+Initialise a Kafka consumer. `server`: broker address as symbol. `optiondict`: Kafka config options as a symbol-keyed symbol dict — pass `()!()` for defaults.
+```q
+kafka.initconsumer[`localhost:9092;`fetch.wait.max.ms`fetch.error.backoff.ms!`5`5]
+```
 
-| Function | Args | Returns | Description |
-|---|---|---|---|
-| `init[configs]` | dict | void | Load the native library, wire dependencies, install the message callback |
-| `initconsumer[server;optiondict]` | symbol, symbol dict | int (consumer handle) | Initialise a Kafka consumer. `server`: broker address as symbol e.g. `` `localhost:9092 ``. `optiondict`: Kafka config options as symbol-keyed symbol dict e.g. `` `fetch.wait.max.ms`fetch.error.backoff.ms!`5`5 `` — pass `()!()` for defaults |
-| `initproducer[server;optiondict]` | symbol, symbol dict | int (producer handle) | Initialise a Kafka producer. Same arg shapes as `initconsumer` |
-| `cleanupconsumer[]` | — | void | Disconnect and free the consumer |
-| `cleanupproducer[]` | — | void | Disconnect and free the producer |
-| `subscribe[topic;partition]` | symbol, int | void | Start the subscription thread. `topic`: Kafka topic name as symbol. `partition`: partition number as int |
-| `publish[topic;partition;key;msg]` | symbol, int, symbol, bytes | void | Publish to a topic. `key`: message key as symbol (use `` ` `` for no key). `msg`: payload as byte vector e.g. `` `byte$"hello" `` |
-| `setkupd[f]` | binary function | void | Replace the message callback. `f` must be `{[k;x]}` where `k` is the message key (symbol) and `x` is the payload (byte vector) |
-
-## Before `init` is called
-
-All six native functions (`initconsumer`, `initproducer`, `cleanupconsumer`,
-`cleanupproducer`, `subscribe`, `publish`) are stubs that throw:
-'di.kafka: kafka not initialised - call init first
+### `initproducer[server;optiondict]`
+Initialise a Kafka producer. Same argument shapes as `initconsumer`.
+```q
+kafka.initproducer[`localhost:9092;`queue.buffering.max.ms`batch.num.messages!`5`1]
+```
 
-This distinguishes "init not called" from a genuine broker or argument error.
+### `cleanupconsumer[]`
+Disconnect and free the consumer object, stopping the subscription thread.
+```q
+kafka.cleanupconsumer[]
+```
 
-## Example
+### `cleanupproducer[]`
+Disconnect and free the producer object.
+```q
+kafka.cleanupproducer[]
+```
+
+### `subscribe[topic;partition]`
+Start the subscription thread for a topic and partition. Messages are delivered to the callback set via `setkupd`.
+```q
+kafka.subscribe[`trades;0]
+```
+
+### `publish[topic;partition;key;msg]`
+Publish a byte vector to a topic and partition. `key`: message key as symbol (use `` ` `` for no key). `msg`: payload as byte vector.
+```q
+kafka.publish[`trades;0;`;`byte$"hello world"]
+```
+
+### `setkupd[f]`
+Replace the message callback invoked when a subscribed message arrives. `f` must be `{[k;x]}` where `k` is the message key (symbol) and `x` is the payload (byte vector). The root `.kupd` forwarder always delegates to the current value so the swap takes effect immediately.
+```q
+kafka.setkupd[{[k;x] upd[`kafkadata;(enlist .z.p;enlist k;enlist "c"$x)]}]
+```
+
+---
+
+## Usage Example
 
 ```q
 kxlog:use`kx.log
@@ -105,52 +135,37 @@ kafka.cleanupconsumer[]
 kafka.cleanupproducer[]
 ```
 
-## TorQ migration
+---
 
-| TorQ pattern | Module equivalent |
-|---|---|
-| `enabled:@[value;\`enabled;.z.o in \`l64]` | pass `enabled:0b` in `configs` to suppress library loading |
-| `kupd:{[k;x]...}` defined in settings | pass `kupd` key in `configs` to `init`, or call `setkupd` after |
-| default `kupd` prints bytes to stdout | default `kupd` is a no-op — install a callback via `setkupd` |
-| `.kafka.initconsumer[s;o]` | `kafka.initconsumer[s;o]` |
-| `.kafka.initproducer[s;o]` | `kafka.initproducer[s;o]` |
-| `.kafka.cleanupconsumer[(::)]` | `kafka.cleanupconsumer[]` |
-| `.kafka.cleanupproducer[(::)]` | `kafka.cleanupproducer[]` |
-| `.kafka.subscribe[t;p]` | `kafka.subscribe[t;p]` |
-| `.kafka.publish[t;p;k;m]` | `kafka.publish[t;p;k;m]` |
+## Running Tests
 
-## Global side effect
-
-`init` sets `.kupd` in the root namespace as a forwarder into module state:
+### Unit tests
 
 ```q
-`.kupd set {[k;x] .z.m.kupd[k;x]}
+k4unit:use`di.k4unit
+k4unit.moduletest`di.kafka
 ```
 
-This is an unavoidable consequence of the native C library's design — `kafkaq` is
-compiled to call `kupd` by name in the root namespace.
+34 tests. Runs on any kdb-x machine — no TorQ installation or native library required. Covers dependency validation, `enabled:0b` platform skip, stub behaviour for all six native functions, `setkupd` and the `.kupd` forwarder, log normalisation, and a live `kx.log` instance via `normlog`.
 
-## Manual verification (requires kafkaq.so and a Kafka broker)
+### Integration tests
 
 ```q
-kxlog:use`kx.log
+k4unit:use`di.k4unit
+.m.di.0k4unit.KUltf .Q.dd[hsym`$.Q.m.mp`di.kafka;`test_integration.csv]
+.m.di.0k4unit.KUrt[]
+```
 
-kafka:use`di.kafka
-kafka.init[`log`libpath!(kxlog.createLog[];`$/path/to/lib)]
+7 tests. Runs on any kdb-x machine with a TorQ installation that includes Kafka. Requires `KDBLIB` to be set — the standard TorQ environment variable exported by `setenv.sh` pointing to the root library directory. If `KDBLIB` is not set or `kafkaq.so` is not present at the resolved path, the file exits cleanly and nothing fails.
 
-kafka.initconsumer[`localhost:9092;()!()]
-kafka.subscribe[`test;0]
+When the library is present, the tests confirm that `kafka.init` loads `kafkaq.so` successfully and that all six native function bindings are type `112h` (C function) in module state. This proves the library loaded and `bindfunctions` ran correctly. End-to-end testing of consumer and producer operations requires a running Kafka broker and is outside the scope of automated tests.
 
-kafka.initproducer[`localhost:9092;()!()]
-kafka.publish[`test;0;`;`byte$"hello from kdb+"]
+---
 
-kafka.cleanupconsumer[]
-kafka.cleanupproducer[]
-```
+## Notes
 
-## Running tests
-
-```q
-k4unit:use`di.k4unit
-k4unit.moduletest`di.kafka
-```
+- `libpath` is only required when `enabled:1b` (the default on l64). Pass `enabled:0b` to initialise without loading the native library — useful for testing or non-l64 deployments
+- `setkupd` takes effect immediately via the forwarder pattern — the native library always calls `.kupd` in the root namespace, which delegates to whatever `.z.m.kupd` currently holds. However, messages in-flight from the C background thread may briefly invoke the previous handler after the swap. Avoid calling `setkupd` while a subscription is active
+- `init` sets `.kupd` in the root namespace as an unavoidable consequence of the native C library's design — `kafkaq` is compiled to call `kupd` by name in the root namespace
+- All three log keys (`info`, `warn`, `error`) are required — unlike `di.eodtime`, this module calls all three
+- Manual verification requires a running `kafkaq.so` and a Kafka broker. A minimal end-to-end test once those are available: initialise with a real `libpath`, call `initconsumer`, `subscribe`, `initproducer`, `publish`, verify the callback fires, then `cleanupconsumer` and `cleanupproducer`
diff --git a/di/kafka/kafka.q b/di/kafka/kafka.q
index e466aedf..f58c462b 100644
--- a/di/kafka/kafka.q
+++ b/di/kafka/kafka.q
@@ -13,18 +13,27 @@ kupd:defaultkupd;
 enabled:.z.o in `l64;
 
 / ============================================================
-/ native function stubs - replaced by bindfunctions when library loads
+/ native function internals - stubs replaced by bindfunctions when library loads
+/ exported forwarders delegate to these by bare name so bindfunctions updates are reflected immediately
 / ============================================================
 
-initconsumer:{[s;o]'"di.kafka: kafka not initialised - call init first"};
-initproducer:{[s;o]'"di.kafka: kafka not initialised - call init first"};
-cleanupconsumer:{'"di.kafka: kafka not initialised - call init first"};
-cleanupproducer:{'"di.kafka: kafka not initialised - call init first"};
-subscribe:{[t;p]'"di.kafka: kafka not initialised - call init first"};
-publish:{[t;p;k;m]'"di.kafka: kafka not initialised - call init first"};
+nativeinitconsumer:{[s;o]'"di.kafka: kafka not initialised - call init first"};
+nativeinitproducer:{[s;o]'"di.kafka: kafka not initialised - call init first"};
+nativecleanupconsumer:{'"di.kafka: kafka not initialised - call init first"};
+nativecleanupproducer:{'"di.kafka: kafka not initialised - call init first"};
+nativesubscribe:{[t;p]'"di.kafka: kafka not initialised - call init first"};
+nativepublish:{[t;p;k;m]'"di.kafka: kafka not initialised - call init first"};
+
+/ exported forwarders - captured at use time, delegate to native* internals dynamically
+initconsumer:{[s;o] nativeinitconsumer[s;o]};
+initproducer:{[s;o] nativeinitproducer[s;o]};
+cleanupconsumer:{nativecleanupconsumer[]};
+cleanupproducer:{nativecleanupproducer[]};
+subscribe:{[t;p] nativesubscribe[t;p]};
+publish:{[t;p;k;m] nativepublish[t;p;k;m]};
 
 / ============================================================
-/ internal functions
+/ internal helpers
 / ============================================================
 
 normlog:{[logdict]
@@ -39,15 +48,15 @@ normlog:{[logdict]
     logdict]
   };
 
-setconfig:{[configs]
+setconfig:{[deps]
   / apply recognised configuration overrides; dep keys (log etc.) are ignored
-  cfg:$[99h=type configs;configs;()!()];
+  cfg:$[99h=type deps;deps;()!()];
   if[`enabled in key cfg; .z.m.enabled:cfg`enabled];
   if[`kupd in key cfg; .z.m.kupd:cfg`kupd];
   };
 
 loadlib:{[libpath]
-  / resolve and load the native kafkaq shared library from the configured path
+  / resolve and validate the native kafkaq shared library path - actual binding happens in bindfunctions
   / the os-appropriate extension (.so or .dll) is appended automatically
   lib:`$string[libpath],"/",string[.z.o],"/kafkaq";
   libfile:hsym ` sv lib,$[.z.o like "w*";`dll;`so];
@@ -55,22 +64,23 @@ loadlib:{[libpath]
   if[not libexists;
     .z.m.log[`error][`kafka;"native library not found at ",string libfile];
     '"di.kafka: native library not found at ",string libfile];
-  .z.m.log[`info][`kafka;"loading library ",string libfile];
+  .z.m.log[`info][`kafka;"library found at ",string libfile];
   .z.m.lib:lib;
   };
 
 bindfunctions:{[]
-  / bind the six c functions from the loaded kafkaq library into module-local state
-  / overwrites the stubs defined at module level
-  / cleanupconsumer and cleanupproducer are unary in the c interface but take null (::) - expose as niladic
-  .z.m.initconsumer:.z.m.lib 2:(`initconsumer;2);
-  .z.m.initproducer:.z.m.lib 2:(`initproducer;2);
+  / bind the six c functions from the loaded kafkaq library into the native* internal targets
+  / the exported forwarders (initconsumer etc.) delegate to these by bare name, so updates here
+  / are immediately reflected in all subsequent calls through the exported api
+  / rawcleanupconsumer and rawcleanupproducer are unary in the c interface - wrapped as niladic
+  .z.m.nativeinitconsumer:.z.m.lib 2:(`initconsumer;2);
+  .z.m.nativeinitproducer:.z.m.lib 2:(`initproducer;2);
   .z.m.rawcleanupconsumer:.z.m.lib 2:(`cleanupconsumer;1);
   .z.m.rawcleanupproducer:.z.m.lib 2:(`cleanupproducer;1);
-  .z.m.cleanupconsumer:{rawcleanupconsumer[(::)]};
-  .z.m.cleanupproducer:{rawcleanupproducer[(::)]};
-  .z.m.subscribe:.z.m.lib 2:(`subscribe;2);
-  .z.m.publish:.z.m.lib 2:(`publish;4);
+  .z.m.nativecleanupconsumer:{rawcleanupconsumer[(::)]};
+  .z.m.nativecleanupproducer:{rawcleanupproducer[(::)]};
+  .z.m.nativesubscribe:.z.m.lib 2:(`subscribe;2);
+  .z.m.nativepublish:.z.m.lib 2:(`publish;4);
   };
 
 / ============================================================
@@ -85,27 +95,27 @@ setkupd:{[f]
   .z.m.kupd:f;
   };
 
-init:{[configs]
+init:{[deps]
   / initialise the kafka module - validate deps, apply config, load native library
-  / configs: dict containing `log (required) plus optional `libpath, `enabled, `kupd
+  / deps: dict containing `log (required) plus optional `libpath, `enabled, `kupd
   / log dep: `info`warn`error!({[c;m]};{[c;m]};{[c;m]}) - binary, c=context symbol, m=string
   / examples:
   /   kafka.init[`log`libpath!(logdep;`$/opt/kdb/lib)]
   /   kafka.init[`log`libpath`enabled!(logdep;`$/opt/kdb/lib;0b)]
-  if[99h<>type configs;
-    '"di.kafka: configs must be a dict with `log key"];
-  if[not `log in key configs;
+  if[99h<>type deps;
+    '"di.kafka: deps must be a dict with `log key"];
+  if[not `log in key deps;
     '"di.kafka: log dependency is required; pass `info`warn`error!(infofn;warnfn;errfn) keyed on `log"];
-  if[99h<>type configs`log;
+  if[99h<>type deps`log;
     '"di.kafka: log value must be a dict; pass `info`warn`error functions"];
-  if[not all `info`warn`error in key configs`log;
-    '"di.kafka: log dict must have `info`warn`error keys; got: ",(", " sv string key configs`log)];
-  .z.m.log:normlog configs`log;
-  setconfig configs;
+  if[not all `info`warn`error in key deps`log;
+    '"di.kafka: log dict must have `info`warn`error keys; got: ",(", " sv string key deps`log)];
+  .z.m.log:normlog deps`log;
+  setconfig deps;
   if[enabled;
-    if[not `libpath in key configs;
-      '"di.kafka: configs must contain `libpath - root library directory containing the os-specific kafkaq build"];
-    loadlib configs`libpath;
+    if[not `libpath in key deps;
+      '"di.kafka: deps must contain `libpath - root library directory containing the os-specific kafkaq build"];
+    loadlib deps`libpath;
     bindfunctions[];
     `.kupd set {[k;x] .z.m.kupd[k;x]};
   ];
diff --git a/di/kafka/test.csv b/di/kafka/test.csv
index 801dfc58..b1093e92 100644
--- a/di/kafka/test.csv
+++ b/di/kafka/test.csv
@@ -7,24 +7,24 @@ before,0,0,q,.test.cap:([] fn:`symbol$();msg:()),1,1,initialise log capture tabl
 before,0,0,q,.test.kupdk:`,1,1,last key received by the test kupd callback
 before,0,0,q,.test.kupdx:`byte$(),1,1,last payload received by the test kupd callback
 comment,,,,,,,init - dep validation
-fail,0,0,q,kafka.init[(::)],1,1,errors when configs is not a dictionary
+fail,0,0,q,kafka.init[(::)],1,1,errors when deps is not a dictionary
 fail,0,0,q,kafka.init[()!()],1,1,errors when log dependency is missing
 fail,0,0,q,kafka.init[enlist[`log]!enlist 42],1,1,errors when log value is not a dictionary
 fail,0,0,q,kafka.init[enlist[`log]!enlist `info`warn!(logdep`info;logdep`warn)],1,1,errors when log dict is missing the error key
 run,0,0,q,.test.err:@[{kafka.init[()!()]};(::);{x}],1,1,capture error string from init with missing log dep
 true,0,0,q,.test.err like "di.kafka:*",1,1,error is prefixed di.kafka:
 comment,,,,,,,init - config validation
-fail,0,0,q,kafka.init[enlist[`log]!enlist logdep],1,1,errors when libpath is missing from config
+fail,0,0,q,kafka.init[enlist[`log]!enlist logdep],1,1,errors when libpath is missing from deps
 comment,,,,,,,init - disabled platform (enabled:0b) - native library skipped; libpath not required
 run,0,0,q,kafka.init[`log`enabled!(logdep;0b)],1,1,init with enabled:0b succeeds without libpath
 true,0,0,q,not .m.di.0kafka.enabled,1,1,module enabled flag reflects disabled state
-comment,,,,,,,stubs - functions throw before init loads the library
+comment,,,,,,,stubs - functions throw when native library not loaded
 true,0,0,q,.[kafka.initconsumer;(`localhost:9092;()!());{x}] like "*kafka not initialised*",1,1,initconsumer stub throws before init
 true,0,0,q,.[kafka.initproducer;(`localhost:9092;()!());{x}] like "*kafka not initialised*",1,1,initproducer stub throws before init
 true,0,0,q,.[kafka.subscribe;(`test;0);{x}] like "*kafka not initialised*",1,1,subscribe stub throws before init
 true,0,0,q,.[kafka.publish;(`test;0;`;0x68656c6c6f);{x}] like "*kafka not initialised*",1,1,publish stub throws before init
-true,0,0,q,.[kafka.cleanupconsumer;();{x}] like "*kafka not initialised*",1,1,cleanupconsumer stub throws before init
-true,0,0,q,.[kafka.cleanupproducer;();{x}] like "*kafka not initialised*",1,1,cleanupproducer stub throws before init
+true,0,0,q,@[{kafka.cleanupconsumer[]};(::);{x}] like "*kafka not initialised*",1,1,cleanupconsumer stub throws when native library not loaded
+true,0,0,q,@[{kafka.cleanupproducer[]};(::);{x}] like "*kafka not initialised*",1,1,cleanupproducer stub throws when native library not loaded
 comment,,,,,,,setkupd - replaces the callback stored in module state
 run,0,0,q,kafka.setkupd[{[k;x] .test.kupdk:k;.test.kupdx:x}],1,1,replace the callback via setkupd
 true,0,0,q,.m.di.0kafka.kupd~{[k;x] .test.kupdk:k;.test.kupdx:x},1,1,module-local kupd replaced by setkupd
diff --git a/di/kafka/test_integration.csv b/di/kafka/test_integration.csv
new file mode 100644
index 00000000..9102e3da
--- /dev/null
+++ b/di/kafka/test_integration.csv
@@ -0,0 +1,19 @@
+action,ms,bytes,lang,code,repeat,minver,comment
+comment,,,,,,,integration tests - require KDBLIB env var set to the root library directory
+comment,,,,,,,KDBLIB is standard TorQ - set by setenv.sh in every TorQ deployment
+comment,,,,,,,tests skip automatically if KDBLIB is absent or kafkaq is not found at the resolved path
+before,0,0,q,kafka:use`di.kafka,1,1,load the kafka module
+before,0,0,q,logdep:`info`warn`error!({[c;m]};{[c;m]};{[c;m]}),1,1,silent no-op log mock
+before,0,0,q,libpath:`$getenv`KDBLIB,1,1,read library root directory from KDBLIB
+before,0,0,q,if[`~libpath;exit 0],1,1,skip if KDBLIB not set
+before,0,0,q,libfile:hsym`$"/" sv ((string libpath);(string .z.o);$[.z.o like "w*";"kafkaq.dll";"kafkaq.so"]),1,1,build os-specific library file path
+before,0,0,q,if[()~key libfile;exit 0],1,1,skip if kafkaq not found at resolved path
+comment,,,,,,,init - library loads without error
+run,0,0,q,kafka.init[`log`libpath!(logdep;libpath)],1,1,init with real kafkaq library succeeds
+comment,,,,,,,bindings - native internals are c functions after successful init
+true,0,0,q,112h=type .m.di.0kafka.nativeinitconsumer,1,1,nativeinitconsumer bound to c function
+true,0,0,q,112h=type .m.di.0kafka.nativeinitproducer,1,1,nativeinitproducer bound to c function
+true,0,0,q,112h=type .m.di.0kafka.rawcleanupconsumer,1,1,rawcleanupconsumer bound to c function
+true,0,0,q,112h=type .m.di.0kafka.rawcleanupproducer,1,1,rawcleanupproducer bound to c function
+true,0,0,q,112h=type .m.di.0kafka.nativesubscribe,1,1,nativesubscribe bound to c function
+true,0,0,q,112h=type .m.di.0kafka.nativepublish,1,1,nativepublish bound to c function

From 0be366ed6355e934dc6e9a5bc89dc6d15747d795 Mon Sep 17 00:00:00 2001
From: alowrydi <aaron.lowry@dataintellect.com>
Date: Tue, 30 Jun 2026 17:11:37 +0100
Subject: [PATCH 08/10] Document KDBLIB fallback for libpath and clarify
 integration tests run without TorQ

---
 di/kafka/kafka.md             | 43 +++++++++++++++++++----------------
 di/kafka/test.csv             |  9 ++++++++
 di/kafka/test_integration.csv |  2 ++
 3 files changed, 35 insertions(+), 19 deletions(-)

diff --git a/di/kafka/kafka.md b/di/kafka/kafka.md
index 4d2f3ff2..288eb4d6 100644
--- a/di/kafka/kafka.md
+++ b/di/kafka/kafka.md
@@ -7,9 +7,9 @@ Wrapper around the `kafkaq` native shared library. Provides consumer and produce
 ## Features
 
 - Wrap the `kafkaq` native C library and expose consumer and producer lifecycle management as clean kdb-x functions
-- Provide a configurable message callback via `setkupd` — the native library always delegates through the current callback without requiring re-initialisation
+- Provide a configurable message callback via `setkupd` - the native library always delegates through the current callback without requiring re-initialisation
 - Detect and normalise `kx.log` instances automatically so callers can pass a logger directly without manual wrapping
-- Gracefully degrade on non-l64 platforms — all native functions remain as informative stubs when `enabled:0b`
+- Gracefully degrade on non-l64 platforms - all native functions remain as informative stubs when `enabled:0b`
 - Distinguish "module not initialised" from genuine broker or argument errors via descriptive stub error messages with `di.kafka:` prefix
 
 ---
@@ -18,11 +18,11 @@ Wrapper around the `kafkaq` native shared library. Provides consumer and produce
 
 | Dependency | Key | Required | Description |
 |---|---|---|---|
-| logger | `` `log `` | yes | `info`, `warn`, `error` — each binary `{[c;m]}` where `c` is a symbol context and `m` is a string |
+| logger | `` `log `` | yes | `info`, `warn`, `error` - each binary `{[c;m]}` where `c` is a symbol context and `m` is a string |
 
 The `log` dependency must be passed to `init` inside the `deps` dict. The module throws immediately if it is absent or missing any of the three required keys. All three are required since the module calls `info`, `warn`, and `error`.
 
-A `kx.log` instance can be passed directly — the module normalises monadic functions to the binary `{[c;m]}` contract automatically via `normlog`. Context is embedded in the output as `"context: message"`:
+A `kx.log` instance can be passed directly - the module normalises monadic functions to the binary `{[c;m]}` contract automatically via `normlog`. Context is embedded in the output as `"context: message"`:
 
 ```q
 kxlog:use`kx.log
@@ -46,16 +46,15 @@ kafka.init[`log`enabled!(kxlog.createLog[];0b)]
 
 | Key | Required | Description |
 |---|---|---|
-| `` `log `` | yes | Binary log dep — `info`, `warn`, `error` functions each `{[c;m]}` |
-| `` `libpath `` | yes (when enabled) | Root library directory — the OS-specific subdirectory and `kafkaq` filename are appended automatically (e.g. `/opt/kdb/lib` → `/opt/kdb/lib/l64/kafkaq.so`) |
-| `` `kupd `` | no | Initial message callback `{[k;x]}` — defaults to a no-op; replace via `setkupd` after init |
-| `` `enabled `` | no | Whether to load the native library. Defaults to `1b` on `l64`, `0b` elsewhere. Set to `0b` to suppress library loading on unsupported platforms — all native functions remain stubs |
+| `` `log `` | yes | Binary log dep - `info`, `warn`, `error` functions each `{[c;m]}` |
+| `` `libpath `` | no, if `KDBLIB` is set | Root library directory. Falls back to `$KDBLIB` if omitted. The OS-specific subdirectory and `kafkaq` filename are appended automatically - for example, a root of `/opt/kdb/lib` resolves to `/opt/kdb/lib/l64/kafkaq.so` on Linux |
+| `` `kupd `` | no | Initial message callback `{[k;x]}` - defaults to a no-op; replace via `setkupd` after init |
+| `` `enabled `` | no | Whether to load the native library. Defaults to `1b` on `l64`, `0b` elsewhere. Set to `0b` to suppress library loading on unsupported platforms - all native functions remain stubs |
 
-`init` must be called before any native functions are used. Before `init` is called, all six native functions (`initconsumer`, `initproducer`, `cleanupconsumer`, `cleanupproducer`, `subscribe`, `publish`) are stubs that throw:
+If `libpath` is omitted and `KDBLIB` is not set in the environment, `init` throws - there is no further fallback.
 
-```
+`init` must be called before any native functions are used. Before `init` is called, all six native functions (`initconsumer`, `initproducer`, `cleanupconsumer`, `cleanupproducer`, `subscribe`, `publish`) are stubs that throw:
 'di.kafka: kafka not initialised - call init first
-```
 
 This distinguishes "init not called" from a genuine broker or argument error.
 
@@ -70,7 +69,7 @@ kafka.init[`log`libpath!(logdep;`$/opt/kdb/lib)]
 ```
 
 ### `initconsumer[server;optiondict]`
-Initialise a Kafka consumer. `server`: broker address as symbol. `optiondict`: Kafka config options as a symbol-keyed symbol dict — pass `()!()` for defaults.
+Initialise a Kafka consumer. `server`: broker address as symbol. `optiondict`: Kafka config options as a symbol-keyed symbol dict - pass `()!()` for defaults.
 ```q
 kafka.initconsumer[`localhost:9092;`fetch.wait.max.ms`fetch.error.backoff.ms!`5`5]
 ```
@@ -146,7 +145,7 @@ k4unit:use`di.k4unit
 k4unit.moduletest`di.kafka
 ```
 
-34 tests. Runs on any kdb-x machine — no TorQ installation or native library required. Covers dependency validation, `enabled:0b` platform skip, stub behaviour for all six native functions, `setkupd` and the `.kupd` forwarder, log normalisation, and a live `kx.log` instance via `normlog`.
+34 tests. Runs on any kdb-x machine - no TorQ installation or native library required. Covers dependency validation, `enabled:0b` platform skip, stub behaviour for all six native functions, `setkupd` and the `.kupd` forwarder, log normalisation, and a live `kx.log` instance via `normlog`.
 
 ### Integration tests
 
@@ -156,7 +155,13 @@ k4unit:use`di.k4unit
 .m.di.0k4unit.KUrt[]
 ```
 
-7 tests. Runs on any kdb-x machine with a TorQ installation that includes Kafka. Requires `KDBLIB` to be set — the standard TorQ environment variable exported by `setenv.sh` pointing to the root library directory. If `KDBLIB` is not set or `kafkaq.so` is not present at the resolved path, the file exits cleanly and nothing fails.
+7 tests. Runs on any kdb-x machine where `KDBLIB` is set and `kafkaq` is present at the resolved path - TorQ deployments set this up automatically via `setenv.sh`, but it's not a strict requirement. A non-TorQ user with their own `kafkaq.so` build can run these tests by exporting `KDBLIB` themselves to point at it:
+
+```bash
+export KDBLIB=/path/to/your/kafkaq/lib
+```
+
+If `KDBLIB` is not set or `kafkaq.so` is not present at the resolved path, the file exits cleanly and nothing fails.
 
 When the library is present, the tests confirm that `kafka.init` loads `kafkaq.so` successfully and that all six native function bindings are type `112h` (C function) in module state. This proves the library loaded and `bindfunctions` ran correctly. End-to-end testing of consumer and producer operations requires a running Kafka broker and is outside the scope of automated tests.
 
@@ -164,8 +169,8 @@ When the library is present, the tests confirm that `kafka.init` loads `kafkaq.s
 
 ## Notes
 
-- `libpath` is only required when `enabled:1b` (the default on l64). Pass `enabled:0b` to initialise without loading the native library — useful for testing or non-l64 deployments
-- `setkupd` takes effect immediately via the forwarder pattern — the native library always calls `.kupd` in the root namespace, which delegates to whatever `.z.m.kupd` currently holds. However, messages in-flight from the C background thread may briefly invoke the previous handler after the swap. Avoid calling `setkupd` while a subscription is active
-- `init` sets `.kupd` in the root namespace as an unavoidable consequence of the native C library's design — `kafkaq` is compiled to call `kupd` by name in the root namespace
-- All three log keys (`info`, `warn`, `error`) are required — unlike `di.eodtime`, this module calls all three
-- Manual verification requires a running `kafkaq.so` and a Kafka broker. A minimal end-to-end test once those are available: initialise with a real `libpath`, call `initconsumer`, `subscribe`, `initproducer`, `publish`, verify the callback fires, then `cleanupconsumer` and `cleanupproducer`
+- `libpath` is only required when `enabled:1b` (the default on l64) and `KDBLIB` is not set in the environment. Pass `enabled:0b` to initialise without loading the native library - useful for testing or non-l64 deployments
+- `setkupd` takes effect immediately via the forwarder pattern - the native library always calls `.kupd` in the root namespace, which delegates to whatever `.z.m.kupd` currently holds. However, messages in-flight from the C background thread may briefly invoke the previous handler after the swap. Avoid calling `setkupd` while a subscription is active
+- `init` sets `.kupd` in the root namespace as an unavoidable consequence of the native C library's design - `kafkaq` is compiled to call `kupd` by name in the root namespace
+- All three log keys (`info`, `warn`, `error`) are required - unlike `di.eodtime`, this module calls all three
+- Manual verification requires a running `kafkaq.so` and a Kafka broker. A minimal end-to-end test once those are available: initialise with a real `libpath`, call `initconsumer`, `subscribe`, `initproducer`, `publish`, verify the callback fires, then `cleanupconsumer` and `cleanupproducer`
\ No newline at end of file
diff --git a/di/kafka/test.csv b/di/kafka/test.csv
index b1093e92..fbf7e8f7 100644
--- a/di/kafka/test.csv
+++ b/di/kafka/test.csv
@@ -6,6 +6,7 @@ before,0,0,q,caplog:`info`warn`error!({[c;m] `.test.cap upsert(`info;m)};{[c;m]
 before,0,0,q,.test.cap:([] fn:`symbol$();msg:()),1,1,initialise log capture table
 before,0,0,q,.test.kupdk:`,1,1,last key received by the test kupd callback
 before,0,0,q,.test.kupdx:`byte$(),1,1,last payload received by the test kupd callback
+
 comment,,,,,,,init - dep validation
 fail,0,0,q,kafka.init[(::)],1,1,errors when deps is not a dictionary
 fail,0,0,q,kafka.init[()!()],1,1,errors when log dependency is missing
@@ -13,11 +14,14 @@ fail,0,0,q,kafka.init[enlist[`log]!enlist 42],1,1,errors when log value is not a
 fail,0,0,q,kafka.init[enlist[`log]!enlist `info`warn!(logdep`info;logdep`warn)],1,1,errors when log dict is missing the error key
 run,0,0,q,.test.err:@[{kafka.init[()!()]};(::);{x}],1,1,capture error string from init with missing log dep
 true,0,0,q,.test.err like "di.kafka:*",1,1,error is prefixed di.kafka:
+
 comment,,,,,,,init - config validation
 fail,0,0,q,kafka.init[enlist[`log]!enlist logdep],1,1,errors when libpath is missing from deps
+
 comment,,,,,,,init - disabled platform (enabled:0b) - native library skipped; libpath not required
 run,0,0,q,kafka.init[`log`enabled!(logdep;0b)],1,1,init with enabled:0b succeeds without libpath
 true,0,0,q,not .m.di.0kafka.enabled,1,1,module enabled flag reflects disabled state
+
 comment,,,,,,,stubs - functions throw when native library not loaded
 true,0,0,q,.[kafka.initconsumer;(`localhost:9092;()!());{x}] like "*kafka not initialised*",1,1,initconsumer stub throws before init
 true,0,0,q,.[kafka.initproducer;(`localhost:9092;()!());{x}] like "*kafka not initialised*",1,1,initproducer stub throws before init
@@ -25,9 +29,11 @@ true,0,0,q,.[kafka.subscribe;(`test;0);{x}] like "*kafka not initialised*",1,1,s
 true,0,0,q,.[kafka.publish;(`test;0;`;0x68656c6c6f);{x}] like "*kafka not initialised*",1,1,publish stub throws before init
 true,0,0,q,@[{kafka.cleanupconsumer[]};(::);{x}] like "*kafka not initialised*",1,1,cleanupconsumer stub throws when native library not loaded
 true,0,0,q,@[{kafka.cleanupproducer[]};(::);{x}] like "*kafka not initialised*",1,1,cleanupproducer stub throws when native library not loaded
+
 comment,,,,,,,setkupd - replaces the callback stored in module state
 run,0,0,q,kafka.setkupd[{[k;x] .test.kupdk:k;.test.kupdx:x}],1,1,replace the callback via setkupd
 true,0,0,q,.m.di.0kafka.kupd~{[k;x] .test.kupdk:k;.test.kupdx:x},1,1,module-local kupd replaced by setkupd
+
 comment,,,,,,,setkupd - the root .kupd forwarder delegates into module state (wired without loading the native lib)
 before,0,0,q,.m.di.0kafka.kupd:{[k;x] .test.kupdk:k;.test.kupdx:x},1,1,set a capturing callback directly in module state
 before,0,0,q,`.kupd set {[k;x] .m.di.0kafka.kupd[k;x]},1,1,install the forwarder directly as init would without loading the native lib
@@ -35,16 +41,19 @@ run,0,0,q,.test.kupdk:`;.test.kupdx:`byte$(),1,1,reset capture
 run,0,0,q,.kupd[`testkey;`byte$"hello"],1,1,simulate the native library delivering a message
 true,0,0,q,`testkey~.test.kupdk,1,1,forwarder passed the key through to the module callback
 true,0,0,q,(`byte$"hello")~.test.kupdx,1,1,forwarder passed the payload through to the module callback
+
 comment,,,,,,,setkupd - replacing callback via setkupd is reflected in the forwarder immediately
 run,0,0,q,kafka.setkupd[{[k;x] .test.kupdk:`replaced;.test.kupdx:x}],1,1,replace the callback
 run,0,0,q,.test.kupdk:`,1,1,reset capture
 run,0,0,q,.kupd[`;`byte$"x"],1,1,deliver a message through the forwarder
 true,0,0,q,`replaced~.test.kupdk,1,1,new callback invoked - forwarder always delegates to current .z.m.kupd
+
 comment,,,,,,,init - kx.log instance normalised automatically (no manual wrapping required)
 run,0,0,q,kxlogger:use`kx.log,1,1,load kx.log
 run,0,0,q,kxinst:kxlogger.createLog[],1,1,create kx.log instance
 run,0,0,q,kafka.init[`log`enabled!(kxinst;0b)],1,1,bare kx.log instance accepted without wrapping
 run,0,0,q,.m.di.0kafka.log[`info][`kafka;"normalisation test"],1,1,normalised logger accepts binary {[c;m]} call without error
+
 comment,,,,,,,log call verification - capturing logger sees init message with binary {[c;m]} args
 run,0,0,q,.test.cap:0#.test.cap,1,1,reset log capture
 run,0,0,q,.m.di.0kafka.log:caplog,1,1,inject capturing logger directly for log-only tests
diff --git a/di/kafka/test_integration.csv b/di/kafka/test_integration.csv
index 9102e3da..47d3d18f 100644
--- a/di/kafka/test_integration.csv
+++ b/di/kafka/test_integration.csv
@@ -8,8 +8,10 @@ before,0,0,q,libpath:`$getenv`KDBLIB,1,1,read library root directory from KDBLIB
 before,0,0,q,if[`~libpath;exit 0],1,1,skip if KDBLIB not set
 before,0,0,q,libfile:hsym`$"/" sv ((string libpath);(string .z.o);$[.z.o like "w*";"kafkaq.dll";"kafkaq.so"]),1,1,build os-specific library file path
 before,0,0,q,if[()~key libfile;exit 0],1,1,skip if kafkaq not found at resolved path
+
 comment,,,,,,,init - library loads without error
 run,0,0,q,kafka.init[`log`libpath!(logdep;libpath)],1,1,init with real kafkaq library succeeds
+
 comment,,,,,,,bindings - native internals are c functions after successful init
 true,0,0,q,112h=type .m.di.0kafka.nativeinitconsumer,1,1,nativeinitconsumer bound to c function
 true,0,0,q,112h=type .m.di.0kafka.nativeinitproducer,1,1,nativeinitproducer bound to c function

From 4a0b1eb23fa5ce456eb3a3325760f7aaca532a19 Mon Sep 17 00:00:00 2001
From: alowrydi <aaron.lowry@dataintellect.com>
Date: Thu, 2 Jul 2026 16:05:55 +0100
Subject: [PATCH 09/10] addressing AI comments

---
 di/kafka/kafka.q | 80 ++++++++++++++++++++++++------------------------
 1 file changed, 40 insertions(+), 40 deletions(-)

diff --git a/di/kafka/kafka.q b/di/kafka/kafka.q
index f58c462b..80f0a66a 100644
--- a/di/kafka/kafka.q
+++ b/di/kafka/kafka.q
@@ -1,21 +1,21 @@
-/ kafka - wrapper around the kafkaq native library
-/ provides consumer and producer lifecycle management plus a configurable message callback
-/ the native library calls .kupd in the root namespace on message receipt; init sets a forwarder
-/ into .z.m.kupd so the callback can be replaced at runtime via setkupd without re-initialising
-/ the log dependency is required - init errors immediately if absent or malformed
-/ log functions are binary {[c;m]} where c is a symbol context and m is a string
+// kafka - wrapper around the kafkaq native library
+// provides consumer and producer lifecycle management plus a configurable message callback
+// the native library calls .kupd in the root namespace on message receipt; init sets a forwarder
+// into .z.m.kupd so the callback can be replaced at runtime via setkupd without re-initialising
+// the log dependency is required - init errors immediately if absent or malformed
+// log functions are binary {[c;m]} where c is a symbol context and m is a string
 
-/ default message callback - no-op; replace with setkupd after init
+// default message callback - no-op; replace with setkupd after init
 defaultkupd:{[k;x] (::)};
 
-/ configuration defaults
+// configuration defaults
 kupd:defaultkupd;
 enabled:.z.o in `l64;
 
-/ ============================================================
-/ native function internals - stubs replaced by bindfunctions when library loads
-/ exported forwarders delegate to these by bare name so bindfunctions updates are reflected immediately
-/ ============================================================
+// ============================================================
+// native function internals - stubs replaced by bindfunctions when library loads
+// exported forwarders delegate to these by bare name so bindfunctions updates are reflected immediately
+// ============================================================
 
 nativeinitconsumer:{[s;o]'"di.kafka: kafka not initialised - call init first"};
 nativeinitproducer:{[s;o]'"di.kafka: kafka not initialised - call init first"};
@@ -24,7 +24,7 @@ nativecleanupproducer:{'"di.kafka: kafka not initialised - call init first"};
 nativesubscribe:{[t;p]'"di.kafka: kafka not initialised - call init first"};
 nativepublish:{[t;p;k;m]'"di.kafka: kafka not initialised - call init first"};
 
-/ exported forwarders - captured at use time, delegate to native* internals dynamically
+// exported forwarders - captured at use time, delegate to native* internals dynamically
 initconsumer:{[s;o] nativeinitconsumer[s;o]};
 initproducer:{[s;o] nativeinitproducer[s;o]};
 cleanupconsumer:{nativecleanupconsumer[]};
@@ -32,15 +32,15 @@ cleanupproducer:{nativecleanupproducer[]};
 subscribe:{[t;p] nativesubscribe[t;p]};
 publish:{[t;p;k;m] nativepublish[t;p;k;m]};
 
-/ ============================================================
-/ internal helpers
-/ ============================================================
+// ============================================================
+// internal helpers
+// ============================================================
 
 normlog:{[logdict]
-  / detect kx.log instance by presence of kx.log-specific keys (getlvl, sinks, fmts)
-  / kx.log functions are monadic - wrap each into binary {[c;m]} and embed context in the message
-  / plain {[c;m]} log dicts (info`warn`error only) pass through unchanged
-  $[any `getlvl`sinks`fmts in key logdict;
+  // detect kx.log instance by presence of kx.log-specific keys (getlvl, sinks, fmts)
+  // kx.log functions are monadic - wrap each into binary {[c;m]} and embed context in the message
+  // plain {[c;m]} log dicts (info`warn`error only) pass through unchanged
+  $[all `getlvl`sinks`fmts in key logdict;
     `info`warn`error!(
       {[fn;c;m] fn[string[c],": ",m]}[logdict`info;];
       {[fn;c;m] fn[string[c],": ",m]}[logdict`warn;];
@@ -49,15 +49,15 @@ normlog:{[logdict]
   };
 
 setconfig:{[deps]
-  / apply recognised configuration overrides; dep keys (log etc.) are ignored
+  // apply recognised configuration overrides; dep keys (log etc.) are ignored
   cfg:$[99h=type deps;deps;()!()];
   if[`enabled in key cfg; .z.m.enabled:cfg`enabled];
   if[`kupd in key cfg; .z.m.kupd:cfg`kupd];
   };
 
 loadlib:{[libpath]
-  / resolve and validate the native kafkaq shared library path - actual binding happens in bindfunctions
-  / the os-appropriate extension (.so or .dll) is appended automatically
+  // resolve and validate the native kafkaq shared library path - actual binding happens in bindfunctions
+  // the os-appropriate extension (.so or .dll) is appended automatically
   lib:`$string[libpath],"/",string[.z.o],"/kafkaq";
   libfile:hsym ` sv lib,$[.z.o like "w*";`dll;`so];
   libexists:@[{not ()~key x};libfile;{0b}];
@@ -69,10 +69,10 @@ loadlib:{[libpath]
   };
 
 bindfunctions:{[]
-  / bind the six c functions from the loaded kafkaq library into the native* internal targets
-  / the exported forwarders (initconsumer etc.) delegate to these by bare name, so updates here
-  / are immediately reflected in all subsequent calls through the exported api
-  / rawcleanupconsumer and rawcleanupproducer are unary in the c interface - wrapped as niladic
+  // bind the six c functions from the loaded kafkaq library into the native* internal targets
+  // the exported forwarders (initconsumer etc.) delegate to these by bare name, so updates here
+  // are immediately reflected in all subsequent calls through the exported api
+  // rawcleanupconsumer and rawcleanupproducer are unary in the c interface - wrapped as niladic
   .z.m.nativeinitconsumer:.z.m.lib 2:(`initconsumer;2);
   .z.m.nativeinitproducer:.z.m.lib 2:(`initproducer;2);
   .z.m.rawcleanupconsumer:.z.m.lib 2:(`cleanupconsumer;1);
@@ -83,25 +83,25 @@ bindfunctions:{[]
   .z.m.nativepublish:.z.m.lib 2:(`publish;4);
   };
 
-/ ============================================================
-/ public api
-/ ============================================================
+// ============================================================
+// public api
+// ============================================================
 
 setkupd:{[f]
-  / replace the message callback invoked when a subscribed message arrives
-  / f must be a binary function {[k;x]} where k is the message key (symbol) and x is the payload (bytes)
-  / the root .kupd forwarder always delegates to the current value - swap takes effect immediately
-  / note: messages in-flight from the c background thread may briefly invoke the previous handler
+  // replace the message callback invoked when a subscribed message arrives
+  // f must be a binary function {[k;x]} where k is the message key (symbol) and x is the payload (bytes)
+  // the root .kupd forwarder always delegates to the current value - swap takes effect immediately
+  // note: messages in-flight from the c background thread may briefly invoke the previous handler
   .z.m.kupd:f;
   };
 
 init:{[deps]
-  / initialise the kafka module - validate deps, apply config, load native library
-  / deps: dict containing `log (required) plus optional `libpath, `enabled, `kupd
-  / log dep: `info`warn`error!({[c;m]};{[c;m]};{[c;m]}) - binary, c=context symbol, m=string
-  / examples:
-  /   kafka.init[`log`libpath!(logdep;`$/opt/kdb/lib)]
-  /   kafka.init[`log`libpath`enabled!(logdep;`$/opt/kdb/lib;0b)]
+  // initialise the kafka module - validate deps, apply config, load native library
+  // deps: dict containing `log (required) plus optional `libpath, `enabled, `kupd
+  // log dep: `info`warn`error!({[c;m]};{[c;m]};{[c;m]}) - binary, c=context symbol, m=string
+  // examples:
+  //   kafka.init[`log`libpath!(logdep;`$/opt/kdb/lib)]
+  //   kafka.init[`log`libpath`enabled!(logdep;`$/opt/kdb/lib;0b)]
   if[99h<>type deps;
     '"di.kafka: deps must be a dict with `log key"];
   if[not `log in key deps;

From 47bc4499ca77a12e48824a377dc628699ccb0ec2 Mon Sep 17 00:00:00 2001
From: alowrydi <aaron.lowry@dataintellect.com>
Date: Thu, 2 Jul 2026 16:27:56 +0100
Subject: [PATCH 10/10] returning normlog to standard

---
 di/kafka/kafka.q | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/di/kafka/kafka.q b/di/kafka/kafka.q
index 80f0a66a..ad229c70 100644
--- a/di/kafka/kafka.q
+++ b/di/kafka/kafka.q
@@ -40,7 +40,7 @@ normlog:{[logdict]
   // detect kx.log instance by presence of kx.log-specific keys (getlvl, sinks, fmts)
   // kx.log functions are monadic - wrap each into binary {[c;m]} and embed context in the message
   // plain {[c;m]} log dicts (info`warn`error only) pass through unchanged
-  $[all `getlvl`sinks`fmts in key logdict;
+  $[any `getlvl`sinks`fmts in key logdict;
     `info`warn`error!(
       {[fn;c;m] fn[string[c],": ",m]}[logdict`info;];
       {[fn;c;m] fn[string[c],": ",m]}[logdict`warn;];