Feature di.kafka

di/kafka/init.q

@@ -0,0 +1,5 @@
+/ kafka module - wrapper around the kafkaq native library for producer/consumer operations
+
+\l ::kafka.q
+
+export:([init;initconsumer;initproducer;cleanupconsumer;cleanupproducer;subscribe;publish;setkupd])

di/kafka/kafka.md

@@ -0,0 +1,176 @@
+# 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.
+
+---
+
+## 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 |
+
+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"`:
+
+```q
+kxlog:use`kx.log
+kafka:use`di.kafka
+
+/ minimal - log and libpath only
+kafka.init[`log`libpath!(kxlog.createLog[];`$/opt/kdb/lib)]
+
+/ 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)]})]
+
+/ disabled platform - no libpath required
+kafka.init[`log`enabled!(kxlog.createLog[];0b)]
+```
+
+---
+
+## 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 `` | 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 |
+
+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.
+
+---
+
+## 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.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.
+```q
+kafka.initconsumer[`localhost:9092;`fetch.wait.max.ms`fetch.error.backoff.ms!`5`5]
+```
+
+### `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]
+```
+
+### `cleanupconsumer[]`
+Disconnect and free the consumer object, stopping the subscription thread.
+```q
+kafka.cleanupconsumer[]
+```
+
+### `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
+
+kafka:use`di.kafka
+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)]}]
+kafka.initconsumer[`localhost:9092;`fetch.wait.max.ms`fetch.error.backoff.ms!`5`5]
+kafka.subscribe[`trades;0]
+
+/ 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[]
+```
+
+---
+
+## Running Tests
+
+### Unit tests
+
+```q
+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`.
+
+### Integration tests
+
+```q
+k4unit:use`di.k4unit
+.m.di.0k4unit.KUltf .Q.dd[hsym`$.Q.m.mp`di.kafka;`test_integration.csv]
+.m.di.0k4unit.KUrt[]
+```
+
+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.
+
+---
+
+## Notes
+
+- `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`

di/kafka/kafka.q

@@ -0,0 +1,123 @@
+// 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
+defaultkupd:{[k;x] (::)};
+
+// 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
+// ============================================================
+
+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 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;
+    `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:{[deps]
+  // 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
+  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;"library found at ",string libfile];
+  .z.m.lib:lib;
+  };
+
+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
+  .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.nativecleanupconsumer:{rawcleanupconsumer[(::)]};
+  .z.m.nativecleanupproducer:{rawcleanupproducer[(::)]};
+  .z.m.nativesubscribe:.z.m.lib 2:(`subscribe;2);
+  .z.m.nativepublish:.z.m.lib 2:(`publish;4);
+  };
+
+// ============================================================
+// 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
+  .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)]
+  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 deps`log;
+    '"di.kafka: log value must be a dict; pass `info`warn`error functions"];
+  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 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]};
+  ];
+  .z.m.log[$[enabled;`info;`warn]][`kafka;"di.kafka initialised",$[enabled;"";" (native library not loaded - platform not supported)"]];
+  };

di/kafka/test.csv

@@ -0,0 +1,62 @@
+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!({[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[(::)],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 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
+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 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
+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
+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 "di.kafka *",1,1,log message contains module init confirmation