Add kafka config to settings

CHANGELOG.md

@@ -12,6 +12,41 @@ beyond a package's own generated workflows). Because every package's
 make a package a component of the change — list only packages whose code, tests,
 config or docs changed.
 
+## 0.8.3
+
+### loopqqlz — Kafka client config passthrough [tkati-core, tkati-node-dedup, tkati-node-el, repo]
+
+- `[input.config]` and `[output.config]` (and `[dlq.config]`) set any librdkafka
+  property on the consumer or producer: compression, `linger.ms`, `acks`,
+  `client.id`, `SASL_PLAINTEXT`/`PLAIN` auth. Values may be strings, ints or
+  bools, and the dict is merged over the properties derived from the typed
+  fields. No Rust change was needed — `config_entries` already stringified
+  values and lowered bools
+- Property names and values are librdkafka's to validate, at client
+  construction: `compresion.type` fails node startup with `No such configuration
+  property`, `compression.type = "brotli"` with `Invalid value "brotli" for
+  configuration property "compression.codec"`. So there is no property list in
+  Python to go stale
+- The properties tkati sets itself — `PRODUCER_RESERVED`, `CONSUMER_RESERVED`,
+  both public — are rejected at settings-parse time rather than merged over.
+  `enable.auto.commit` is the one that matters: librdkafka accepts `true`
+  silently, and the consumer's explicit-commit protocol depends on it being
+  false, so a passthrough that could override it would turn a node's
+  rewind-on-failure path into data loss. The other four are reserved only
+  because a typed field already sets them
+- Compression is set this way rather than through a typed field: librdkafka's
+  own error names the bad codec at the same startup a pydantic error would, so a
+  `Literal` would have bought one better message at the cost of a second
+  mechanism for one property. It pays off far more under `format = "json"`, where
+  the codec sees many small near-identical messages, than under `"arrow-batch"`
+- `KafkaProducer` and `KafkaConsumer` now agree on
+  `kafka_config: dict[str, str | int | bool]`
+- No defaults changed: compression stays off and `enable.idempotence` stays
+  false unless `config` says otherwise. Audit finding F5's configurability half
+  is closed, the reordering half is still open and is the next change
+- New `design-docs/2026-10-01-kafka-config-passthrough.md`; F5 marked partly
+  fixed in the audit
+
 ## 0.8.2
 
 ### tzwvropv — ClickHouse insert failures: data errors versus outages [tkati-core, tkati-node-el, repo]

design-docs/2026-09-27-tkati-core-audit.md

@@ -39,7 +39,7 @@ Severity:
 | F2  | Critical | A failed write to a Kafka DLQ loses rows without an error      | Open   |
 | F3  | High     | A ClickHouse outage sends the whole batch to the DLQ           | Fixed 0.8.2 |
 | F4  | Medium   | Commit errors are never reported                               | Open   |
-| F5  | Medium   | Kafka settings can't be configured; producer defaults can reorder retried messages | Open |
+| F5  | Medium   | Kafka settings can't be configured; producer defaults can reorder retried messages | Partly fixed 0.8.3 |
 | F6  | Low      | No handling of revoked partitions: more duplicates after a rebalance | Open |
 | F7  | Low      | Offset ranges are wrong if a partition is re-assigned mid-poll | Open   |
 | F8  | Low      | Large tables in the `arrow-batch` format fail at enqueue       | Open   |
@@ -198,6 +198,20 @@ requests and keep order across retries. Also default to
 separate change: it is excluded in `Cargo.toml` for build reasons that are
 documented there.
 
+**Status.** The configurability half is fixed in 0.8.3 —
+`design-docs/2026-10-01-kafka-config-passthrough.md`. It differs from the
+recommendation above in two ways, both argued there: the dict sits on
+`KafkaInputSettings` / `KafkaOutputSettings` rather than
+`KafkaConnectionSettings`, because a producer-only property under a shared
+`connection` block would be silently ignored on an input; and it is *not*
+merged last — the properties tkati sets itself are rejected at parse time,
+because librdkafka accepts `enable.auto.commit=true` silently and that would
+turn a node's rewind-on-failure path into data loss.
+
+The defaults are unchanged, so **the reordering half of this finding is still
+open**: `enable.idempotence` is still false unless an operator sets it. That is
+the next change.
+
 ### F6. No handling of revoked partitions: more duplicates after a rebalance
 
 **Severity:** Low

packages/tkati-core/CHANGELOG.md

@@ -1,3 +1,17 @@
+# 0.8.3
+
+* `[input.config]` and `[output.config]` pass arbitrary librdkafka properties
+  through to the consumer and producer — compression, `linger.ms`, `acks`,
+  `client.id`, `SASL_PLAINTEXT`/`PLAIN` auth. Values may be strings, ints or
+  bools, and librdkafka validates names and values when the client is built.
+* The properties `tkati-core` sets itself (`PRODUCER_RESERVED`,
+  `CONSUMER_RESERVED`) are rejected in `config` at settings-parse time instead
+  of being silently overridden. `enable.auto.commit` is the one that matters:
+  librdkafka accepts `true` silently, and it would turn a node's rewind-on-
+  failure path into data loss.
+* No defaults changed: compression is still off and `enable.idempotence` still
+  false unless `config` says otherwise.
+
 # 0.8.2
 
 * Fix: a ClickHouse outage no longer empties a batch into the DLQ.

packages/tkati-core/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "tkati_core_native"
 # Kept equal to the workspace's shared Python version (see CLAUDE.md).
-version = "0.8.2"
+version = "0.8.3"
 edition = "2024"
 publish = false
 

packages/tkati-core/README.md

@@ -530,6 +530,51 @@ same way against `ClickHouseOutputSettings`.
 - `"json"` *(default)*: each row becomes a separate Kafka message serialized with orjson.
 - `"arrow-batch"`: the entire table is serialized as a single Arrow IPC stream message.
 
+**Client configuration** — any librdkafka property, via `[output.config]` for a producer
+or `[input.config]` for a consumer:
+
+```toml
+[output.config]
+"compression.type"   = "zstd"   # librdkafka's default is "none"
+"linger.ms"          = 50
+"batch.num.messages" = 50000
+"client.id"          = "node-el-prod-1"
+```
+
+Values may be strings, ints or bools; each reaches librdkafka in the string form it
+expects. The dict is merged over the properties derived from the typed settings, and
+**property names and values are validated by librdkafka when the client is built** — a
+typo'd `compresion.type` fails node startup with `No such configuration property`, and
+`compression.type = "brotli"` with `Invalid value "brotli" for configuration property
+"compression.codec"`. There is no list of properties here to go stale.
+
+`[dlq.config]` works the same way when the DLQ sink is Kafka.
+
+Properties `tkati-core` sets itself are rejected at settings-parse time rather than
+silently overridden — `PRODUCER_RESERVED` and `CONSUMER_RESERVED` in
+`tkati_core.kafka.settings`:
+
+| Property | Role | Why |
+| --- | --- | --- |
+| `enable.auto.commit` | consumer | Must stay false: a node commits each batch explicitly once it is durable, and rewinds it on failure. |
+| `bootstrap.servers` | both | Set from `connection.broker`. |
+| `group.id` | consumer | Set from `consumer.group_id`. |
+| `auto.offset.reset` | consumer | Set from `consumer.auto_offset_reset`. |
+
+Two notes on properties worth setting:
+
+- **Compression** pays off far more under `format = "json"`, where the codec sees many
+  small, near-identical messages in one message set, than under `"arrow-batch"`, which
+  is a single already-compact Arrow IPC message per table with no cross-message
+  redundancy to exploit. Both are compressed; only one is transformed. Consumers need
+  no setting — decompression is automatic.
+- **Authentication** is settable but only partly usable in the published wheel:
+  `security.protocol = "SASL_PLAINTEXT"` with `sasl.mechanism = "PLAIN"` works, while
+  `SCRAM-SHA-256`, `GSSAPI` and `OAUTHBEARER` fail with *"No provider for SASL
+  mechanism … recompile librdkafka with libsasl2"*, and any `_SSL` protocol with
+  *"OpenSSL not available at build time"*. See the `rdkafka` dependency comment in
+  `Cargo.toml` for what re-enabling them takes.
+
 **Message keys** — controlled by `output.topic.key_column` in `settings.toml`:
 
 ```toml

packages/tkati-core/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "tkati-core"
-version = "0.8.2"
+version = "0.8.3"
 description = "Core abstractions for Kafka/ClickHouse streaming pipeline nodes"
 readme = "README.md"
 requires-python = ">=3.13"

packages/tkati-core/tests/test_consumer.py

@@ -9,6 +9,7 @@
 from confluent_kafka import Consumer as RawConsumer
 from confluent_kafka import Producer, TopicPartition
 from confluent_kafka.admin import AdminClient, NewTopic
+from pydantic import ValidationError
 from tkati_core import (
     CONSUMER_PHASES,
     Batch,
@@ -19,7 +20,16 @@
 )
 from tkati_core.kafka.consumer import KafkaConsumer
 from tkati_core.kafka.producer import KafkaProducer
-from tkati_core.kafka.settings import KafkaInputSettings, KafkaOutputSettings
+from tkati_core.kafka.settings import (
+    _RESERVED_REASONS,
+    CONSUMER_RESERVED,
+    PRODUCER_RESERVED,
+    KafkaConnectionSettings,
+    KafkaConsumerSettings,
+    KafkaInputSettings,
+    KafkaOutputSettings,
+    KafkaTopicSettings,
+)
 
 
 def test_from_input_settings_sets_attributes(input_settings: KafkaInputSettings):
@@ -31,6 +41,57 @@ def test_from_input_settings_sets_attributes(input_settings: KafkaInputSettings)
         consumer.close()
 
 
+def test_input_config_is_merged_into_the_client_config(
+    input_settings: KafkaInputSettings, monkeypatch: pytest.MonkeyPatch
+):
+    captured_config: dict[str, str | int | bool] = {}
+
+    class CapturingNativeConsumer:
+        def __init__(self, config: dict[str, str | int | bool], topic: str) -> None:
+            captured_config.update(config)
+
+    monkeypatch.setattr(
+        "tkati_core.kafka.consumer.NativeConsumer", CapturingNativeConsumer
+    )
+    input_settings.config = {"fetch.min.bytes": 65536, "client.id": "reader-1"}
+    KafkaConsumer.from_input_settings(input_settings)
+    # Equality, not containment: the four derived properties are pinned too, so
+    # a passthrough that displaced one would fail here as well as at parse time.
+    assert captured_config == {
+        "bootstrap.servers": input_settings.connection.broker,
+        "group.id": input_settings.consumer.group_id,
+        "auto.offset.reset": input_settings.consumer.auto_offset_reset,
+        "enable.auto.commit": False,
+        "fetch.min.bytes": 65536,
+        "client.id": "reader-1",
+    }
+
+
+@pytest.mark.parametrize(
+    ("key", "message"),
+    [
+        ("enable.auto.commit", "is managed by tkati"),
+        ("group.id", r"is set from `consumer\.group_id`"),
+    ],
+)
+def test_input_config_rejects_a_reserved_property(key: str, message: str):
+    """One case per reason the set exists: a property owned for correctness, and
+    one owned by a typed field."""
+    with pytest.raises(ValidationError, match=message):
+        KafkaInputSettings(
+            connection=KafkaConnectionSettings(broker="broker:9092"),
+            topic=KafkaTopicSettings(name="t"),
+            consumer=KafkaConsumerSettings(group_id="g"),
+            config={key: True},
+        )
+
+
+def test_every_reserved_property_has_a_reason():
+    """The sets are the policy; the reasons are what an operator reads. A key
+    added to one without the other would raise `KeyError` from the validator."""
+    assert (PRODUCER_RESERVED | CONSUMER_RESERVED) <= _RESERVED_REASONS.keys()
+
+
 def test_from_input_settings_builds_arrow_schemas(input_settings: KafkaInputSettings):
     consumer = KafkaConsumer.from_input_settings(input_settings)
     try:

packages/tkati-core/tests/test_producer.py

@@ -5,7 +5,9 @@
 import pyarrow as pa
 import pytest
 from confluent_kafka import Consumer
+from pydantic import ValidationError
 from tkati_core import PRODUCER_PHASES, DeliveryError, LoopStats
+from tkati_core._native import NativeProducer
 from tkati_core.kafka.producer import KafkaProducer
 from tkati_core.kafka.settings import (
     KafkaConnectionSettings,
@@ -30,14 +32,87 @@ def _consume_all(
     return messages
 
 
-def test_from_output_settings_sets_attributes(output_settings: KafkaOutputSettings):
+def _capture_config(monkeypatch: pytest.MonkeyPatch) -> dict[str, str | int | bool]:
+    """Replace `NativeProducer` with a stub recording the config it is built
+    with, so a test can assert on it without a broker."""
+    captured: dict[str, str | int | bool] = {}
+
+    class CapturingNativeProducer:
+        def __init__(self, config: dict[str, str | int | bool], topic: str) -> None:
+            captured.update(config)
+
+    monkeypatch.setattr(
+        "tkati_core.kafka.producer.NativeProducer", CapturingNativeProducer
+    )
+    return captured
+
+
+def test_from_output_settings_sets_attributes(
+    output_settings: KafkaOutputSettings, monkeypatch: pytest.MonkeyPatch
+):
+    captured_config = _capture_config(monkeypatch)
     producer = KafkaProducer.from_output_settings(output_settings)
-    try:
-        assert producer.topic_name == output_settings.topic.name
-        assert producer.format == output_settings.topic.format
-        assert producer.key_column == output_settings.topic.key_column
-    finally:
-        producer.close()
+    assert producer.topic_name == output_settings.topic.name
+    assert producer.format == output_settings.topic.format
+    assert producer.key_column == output_settings.topic.key_column
+    assert captured_config == {"bootstrap.servers": output_settings.connection.broker}
+
+
+def test_output_config_is_merged_into_the_client_config(
+    output_settings: KafkaOutputSettings, monkeypatch: pytest.MonkeyPatch
+):
+    captured_config = _capture_config(monkeypatch)
+    output_settings.config = {
+        "compression.type": "zstd",
+        "linger.ms": 50,
+        "enable.idempotence": True,
+    }
+    KafkaProducer.from_output_settings(output_settings)
+    # Equality, not containment: an accidental extra or renamed property fails.
+    assert captured_config == {
+        "bootstrap.servers": output_settings.connection.broker,
+        "compression.type": "zstd",
+        "linger.ms": 50,
+        "enable.idempotence": True,
+    }
+
+
+def test_output_config_rejects_a_reserved_property():
+    with pytest.raises(ValidationError, match=r"is set from `connection\.broker`"):
+        KafkaOutputSettings(
+            connection=KafkaConnectionSettings(broker="broker:9092"),
+            topic=KafkaTopicSettings(name="t"),
+            config={"bootstrap.servers": "elsewhere:9092"},
+        )
+
+
+def test_librdkafka_accepts_a_passthrough_of_each_value_type(
+    output_settings: KafkaOutputSettings,
+):
+    """The stub above only checks our own merge. This builds a real client, so
+    the three properties are validated by librdkafka, and a str, an int and a
+    bool are all shown to survive the native layer's lowering."""
+    NativeProducer(
+        {
+            "bootstrap.servers": output_settings.connection.broker,
+            "compression.type": "zstd",
+            "linger.ms": 50,
+            "enable.idempotence": True,
+        },
+        "config-passthrough-probe",
+    )
+
+
+def test_librdkafka_rejects_an_unknown_or_invalid_property(
+    output_settings: KafkaOutputSettings,
+):
+    """Why no property-name validation in pydantic: librdkafka does it, at
+    client construction, naming the property."""
+    broker = output_settings.connection.broker
+    with pytest.raises(Exception, match="No such configuration property"):
+        NativeProducer({"bootstrap.servers": broker, "compresion.type": "zstd"}, "t")
+    with pytest.raises(Exception, match='Invalid value "brotli"'):
+        NativeProducer({"bootstrap.servers": broker, "compression.type": "brotli"}, "t")
 
 
 def test_produce_json_format(
@@ -72,7 +147,9 @@ def test_produce_json_format_preserves_timestamp(
     before JSON serialization, instead of orjson emitting an ISO-8601 string."""
     settings = KafkaOutputSettings(
         connection=KafkaConnectionSettings(broker="localhost:9092"),
-        topic=KafkaTopicSettings(name=kafka_output_topic, schema={"ts": "timestamp[ms]"}),
+        topic=KafkaTopicSettings(
+            name=kafka_output_topic, schema={"ts": "timestamp[ms]"}
+        ),
     )
     table = pa.table({"ts": pa.array([1_700_000_000_000], type=pa.int64())}).cast(
         pa.schema([pa.field("ts", pa.timestamp("ms"))])

packages/tkati-core/tkati_core/kafka/consumer.py

@@ -61,12 +61,17 @@ def from_input_settings(cls, settings: "KafkaInputSettings") -> "KafkaConsumer":
 
         Sets enable.auto.commit=False — each batch's offsets are committed
         explicitly via .commit(batch).
+
+        `settings.config` holds extra librdkafka properties, merged over the
+        four derived here. All four are in `CONSUMER_RESERVED` and rejected
+        when the settings are parsed, so the merge cannot overwrite them.
         """
-        kafka_config: dict[str, str | bool] = {
+        kafka_config: dict[str, str | int | bool] = {
             "bootstrap.servers": settings.connection.broker,
             "group.id": settings.consumer.group_id,
             "auto.offset.reset": settings.consumer.auto_offset_reset,
             "enable.auto.commit": False,
+            **settings.config,
         }
         return cls(
             kafka_config=kafka_config,
@@ -76,7 +81,7 @@ def from_input_settings(cls, settings: "KafkaInputSettings") -> "KafkaConsumer":
 
     def __init__(
         self,
-        kafka_config: dict[str, str | bool],
+        kafka_config: dict[str, str | int | bool],
         topic_name: str,
         input_schema: dict[str, str],
     ) -> None: