# ockam_kafka

> For the complete documentation index, see [llms.txt](https://docs.redpanda.com/llms.txt). Component-specific: [connect-full.txt](https://docs.redpanda.com/connect-full.txt)

---
title: ockam_kafka
latest-connect-version: 4.93.0
latest-operator-version: v26.1.4
latest-console-tag: v3.7.3
latest-redpanda-tag: v26.1.9
docname: outputs/ockam_kafka
page-component-name: connect
page-version: master
page-component-version: master
page-component-title: Connect
page-relative-src-path: outputs/ockam_kafka.adoc
page-edit-url: https://github.com/redpanda-data/rp-connect-docs/edit/main/modules/components/pages/outputs/ockam_kafka.adoc
page-git-created-date: "2024-11-25"
page-git-modified-date: "2026-05-26"
---

<!-- Source: https://docs.redpanda.com/connect/components/outputs/ockam_kafka.md -->

**Type:** Output ▼

[Output](https://docs.redpanda.com/connect/components/outputs/ockam_kafka/)[Input](https://docs.redpanda.com/connect/components/inputs/ockam_kafka/)

**Available in:** Self-Managed

Uses [Ockam](https://docs.ockam.io/) to encrypt and write end-to-end encrypted messages to a Kafka topic. You can read encrypted messages using the `ockam_kafka` input or by [creating a Kafka Portal Inlet](https://command.ockam.io/manual/ockam-kafka-inlet-create.html) using Ockam Command.

[Ockam Secure Channels](https://docs.ockam.io/reference/command/secure-channels) guarantee that consumers of topics only receive data from authenticated and authorized producers, and that the data has not been leaked or tampered with in-flight. Neither Kafka brokers, service providers nor other components can see or manipulate the messages.

> 📝 **NOTE**
>
> You can use Ockam to encrypt whole messages, or specific fields in a message.

Introduced in version 4.33.0.

#### Common

```yml
outputs:
  label: ""
  ockam_kafka:
    kafka:
      seed_brokers: [] # No default (optional)
      tls:
        enabled: false
        skip_cert_verify: false
        enable_renegotiation: false
        root_cas: ""
        root_cas_file: ""
        client_certs: []
      max_in_flight: 10
      batching:
        count: 0
        byte_size: 0
        period: ""
        check: ""
        processors: [] # No default (optional)
      partitioner: "" # No default (optional)
      idempotent_write: true
      compression: "" # No default (optional)
      allow_auto_topic_creation: true
      timeout: 10s
      max_message_bytes: 1MiB
      broker_write_max_bytes: 100MiB
      topic: "" # No default (required)
      key: "" # No default (optional)
      partition: "" # No default (optional)
      metadata:
        include_prefixes: []
        include_patterns: []
      timestamp_ms: "" # No default (optional)
    disable_content_encryption: false
    enrollment_ticket: "" # No default (optional)
    identity_name: "" # No default (optional)
    allow: self
    route_to_kafka_outlet: self
    allow_consumer: self
    route_to_consumer: /ip4/127.0.0.1/tcp/6262
    encrypted_fields: []
```

#### Advanced

```yml
outputs:
  label: ""
  ockam_kafka:
    kafka:
      seed_brokers: [] # No default (optional)
      tls:
        enabled: false
        skip_cert_verify: false
        enable_renegotiation: false
        root_cas: ""
        root_cas_file: ""
        client_certs: []
      max_in_flight: 10
      batching:
        count: 0
        byte_size: 0
        period: ""
        check: ""
        processors: [] # No default (optional)
      partitioner: "" # No default (optional)
      idempotent_write: true
      compression: "" # No default (optional)
      allow_auto_topic_creation: true
      timeout: 10s
      max_message_bytes: 1MiB
      broker_write_max_bytes: 100MiB
      topic: "" # No default (required)
      key: "" # No default (optional)
      partition: "" # No default (optional)
      metadata:
        include_prefixes: []
        include_patterns: []
      timestamp_ms: "" # No default (optional)
    disable_content_encryption: false
    enrollment_ticket: "" # No default (optional)
    identity_name: "" # No default (optional)
    allow: self
    route_to_kafka_outlet: self
    allow_consumer: self
    route_to_consumer: /ip4/127.0.0.1/tcp/6262
    encrypted_fields: []
```

## [](#fields)Fields

### [](#allow)`allow`

Use in conjunction with the `route_to_kafka_outlet` field to specify an access control policy for the Kafka Portal Outlet.

For example, setting this value to `kafka_us_east` forces the Kafka Outlet to present an Ockam credential, which confirms that the Outlet has the attribute `kafka_us_east=true`.

**Type**: `string`

**Default**: `self`

### [](#allow_consumer)`allow_consumer`

Specify an access control policy for consumers.

For example, setting this value to `orders_consumer` forces the consumer to present an Ockam credential, which confirms that the consumer has the attribute `orders_consumer=true`.

**Type**: `string`

**Default**: `self`

### [](#disable_content_encryption)`disable_content_encryption`

If this value is set to `true`:

-   Only message payloads remain unencrypted. This setting does not disable TLS or any other transport-layer encryption that may also be enabled.

-   All other `ockam_kafka` inlets and outlets must also have their settings set to `true`.


**Type**: `bool`

**Default**: `false`

### [](#encrypted_fields)`encrypted_fields[]`

The fields to encrypt in the kafka messages, assuming the record is a valid JSON map. By default, the whole record is encrypted.

**Type**: `array`

**Default**: `[]`

### [](#enrollment_ticket)`enrollment_ticket`

The path to a file or a URL where the enrollment ticket value is stored, or an inline hex-encoded value of the enrollment ticket (optional).

You can generate a new ticket using the [`ockam project ticket` command](https://command.ockam.io/manual/ockam-project-ticket.html).

**Type**: `string`

### [](#identity_name)`identity_name`

The name of the [Ockam identity](https://command.ockam.io/manual/ockam-identity.html) to use. If this value is not provided, the default Ockam identity is automatically generated and used (optional).

**Type**: `string`

### [](#kafka)`kafka`

**Type**: `object`

### [](#kafka-allow_auto_topic_creation)`kafka.allow_auto_topic_creation`

Enables topics to be auto created if they do not exist when fetching their metadata.

**Type**: `bool`

**Default**: `true`

### [](#kafka-batching)`kafka.batching`

Configure a [batching policy](https://docs.redpanda.com/connect/configuration/batching/).

**Type**: `object`

```yaml
# Examples:
batching:
  byte_size: 5000
  count: 0
  period: 1s

# ---

batching:
  count: 10
  period: 1s

# ---

batching:
  check: this.contains("END BATCH")
  count: 0
  period: 1m
```

### [](#kafka-batching-byte_size)`kafka.batching.byte_size`

The number of bytes at which the batch is flushed. Set to `0` to disable size-based batching.

**Type**: `int`

**Default**: `0`

### [](#kafka-batching-check)`kafka.batching.check`

A [Bloblang query](https://docs.redpanda.com/connect/guides/bloblang/about/) that returns a boolean value indicating whether a message should end a batch.

**Type**: `string`

**Default**: `""`

```yaml
# Examples:
check: this.type == "end_of_transaction"
```

### [](#kafka-batching-count)`kafka.batching.count`

The number of messages after which the batch is flushed. Set to `0` to disable count-based batching.

**Type**: `int`

**Default**: `0`

### [](#kafka-batching-period)`kafka.batching.period`

The period of time after which an incomplete batch is flushed regardless of its size. This field accepts Go duration format strings such as `100ms`, `1s`, or `5s`.

**Type**: `string`

**Default**: `""`

```yaml
# Examples:
period: 1s

# ---

period: 1m

# ---

period: 500ms
```

### [](#kafka-batching-processors)`kafka.batching.processors[]`

For aggregating and archiving message batches, you can add a list of [processors](https://docs.redpanda.com/connect/components/processors/about/) to apply to a batch as it is flushed (optional). All resulting messages are flushed as a single batch even when you configure processors to split the batch into smaller batches.

**Type**: `processor`

```yaml
# Examples:
processors:
  - archive:
      format: concatenate

# ---

processors:
  - archive:
      format: lines

# ---

processors:
  - archive:
      format: json_array
```

### [](#kafka-broker_write_max_bytes)`kafka.broker_write_max_bytes`

The maximum number of bytes this output can write to a broker connection in a single write. This field corresponds to Kafka’s `socket.request.max.bytes`.

**Type**: `string`

**Default**: `100MiB`

```yaml
# Examples:
broker_write_max_bytes: 128MB

# ---

broker_write_max_bytes: 50mib
```

### [](#kafka-compression)`kafka.compression`

Set an explicit compression type (optional). The preferred default is `snappy` when the broker supports it, with a fall back to `none`.

**Type**: `string`

**Options**: `lz4`, `snappy`, `gzip`, `none`, `zstd`

### [](#kafka-idempotent_write)`kafka.idempotent_write`

Enables the `idempotent_write` producer option. This requires `IDEMPOTENT_WRITE` permission on `CLUSTER`. Disable this option if the correct permission is not available.

**Type**: `bool`

**Default**: `true`

### [](#kafka-key)`kafka.key`

Populates a key for each message (optional). This field supports [interpolation functions](https://docs.redpanda.com/connect/configuration/interpolation/#bloblang-queries).

**Type**: `string`

### [](#kafka-max_in_flight)`kafka.max_in_flight`

The maximum number of message batches to send in parallel at any given time.

**Type**: `int`

**Default**: `10`

### [](#kafka-max_message_bytes)`kafka.max_message_bytes`

The maximum size of an individual message in bytes. Messages larger than this value are rejected. This field is equivalent to Kafka’s `max.message.bytes`.

**Type**: `string`

**Default**: `1MiB`

```yaml
# Examples:
max_message_bytes: 100MB

# ---

max_message_bytes: 50mib
```

### [](#kafka-metadata)`kafka.metadata`

Determines which metadata values are added to messages as headers.

**Type**: `object`

### [](#kafka-metadata-include_patterns)`kafka.metadata.include_patterns[]`

A list of explicit metadata key regular expression (re2) patterns to match against.

**Type**: `array`

**Default**: `[]`

```yaml
# Examples:
include_patterns:
  - .*

# ---

include_patterns:
  - _timestamp_unix$
```

### [](#kafka-metadata-include_prefixes)`kafka.metadata.include_prefixes[]`

A list of explicit metadata key prefixes to match against.

**Type**: `array`

**Default**: `[]`

```yaml
# Examples:
include_prefixes:
  - foo_
  - bar_

# ---

include_prefixes:
  - kafka_

# ---

include_prefixes:
  - content-
```

### [](#kafka-partition)`kafka.partition`

Set an explicit partition for each message (optional). To use this field, set the `partitioner` to `manual`. You must provide an interpolation string that is a valid integer.

This field supports [interpolation functions](https://docs.redpanda.com/connect/configuration/interpolation/#bloblang-queries).

**Type**: `string`

```yaml
# Examples:
partition: ${! meta("partition") }
```

### [](#kafka-partitioner)`kafka.partitioner`

Override the default murmur2 hashing partitioner (optional).

**Type**: `string`

| Option | Summary |
| --- | --- |
| least_backup | Chooses the least backed up partition (the partition with the fewest amount of buffered records). Partitions are selected per batch. |
| manual | Manually select a partition for each message, requires the field partition to be specified. |
| murmur2_hash | Kafka’s default hash algorithm that uses a 32-bit murmur2 hash of the key to compute which partition the record will be on. |
| round_robin | Round-robin’s messages through all available partitions. This algorithm has lower throughput and causes higher CPU load on brokers, but can be useful if you want to ensure an even distribution of records to partitions. |

### [](#kafka-seed_brokers)`kafka.seed_brokers[]`

A list of broker addresses to connect to (optional). List items that contain commas are expanded into multiple addresses.

**Type**: `array`

```yaml
# Examples:
seed_brokers:
  - "localhost:9092"

# ---

seed_brokers:
  - "foo:9092"
  - "bar:9092"

# ---

seed_brokers:
  - "foo:9092,bar:9092"
```

### [](#kafka-timeout)`kafka.timeout`

The maximum period of time allowed for sending messages before a request is abandoned and a retry attempted.

**Type**: `string`

**Default**: `10s`

### [](#kafka-timestamp_ms)`kafka.timestamp_ms`

Set a timestamp (in milliseconds) for each message (optional). Leave this field empty to use the current timestamp.

This field supports [interpolation functions](https://docs.redpanda.com/connect/configuration/interpolation/#bloblang-queries).

**Type**: `string`

```yaml
# Examples:
timestamp_ms: ${! timestamp_unix_milli() }

# ---

timestamp_ms: ${! metadata("kafka_timestamp_ms") }
```

### [](#kafka-tls)`kafka.tls`

Configure Transport Layer Security (TLS) settings to secure network connections. This includes options for standard TLS as well as mutual TLS (mTLS) authentication where both client and server authenticate each other using certificates. Key configuration options include `enabled` to enable TLS, `client_certs` for mTLS authentication, `root_cas`/`root_cas_file` for custom certificate authorities, and `skip_cert_verify` for development environments.

**Type**: `object`

### [](#kafka-tls-client_certs)`kafka.tls.client_certs[]`

A list of client certificates for mutual TLS (mTLS) authentication. Configure this field to enable mTLS, authenticating the client to the server with these certificates.

You must set `tls.enabled: true` for the client certificates to take effect.

**Certificate pairing rules**: For each certificate item, provide either:

-   Inline PEM data using both `cert` **and** `key` or

-   File paths using both `cert_file` **and** `key_file`.


Mixing inline and file-based values within the same item is not supported.

**Type**: `object`

**Default**: `[]`

```yaml
# Examples:
client_certs:
  - cert: foo
    key: bar

# ---

client_certs:
  - cert_file: ./example.pem
    key_file: ./example.key
```

### [](#kafka-tls-client_certs-cert)`kafka.tls.client_certs[].cert`

A plain text certificate to use.

**Type**: `string`

**Default**: `""`

### [](#kafka-tls-client_certs-cert_file)`kafka.tls.client_certs[].cert_file`

The path of a certificate to use.

**Type**: `string`

**Default**: `""`

### [](#kafka-tls-client_certs-key)`kafka.tls.client_certs[].key`

A plain text certificate key to use.

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Secrets](https://docs.redpanda.com/connect/configuration/secrets/).

**Type**: `string`

**Default**: `""`

### [](#kafka-tls-client_certs-key_file)`kafka.tls.client_certs[].key_file`

The path of a certificate key to use.

**Type**: `string`

**Default**: `""`

### [](#kafka-tls-client_certs-password)`kafka.tls.client_certs[].password`

A plain text password for when the private key is password encrypted in PKCS#1 or PKCS#8 format. The obsolete `pbeWithMD5AndDES-CBC` algorithm is not supported for the PKCS#8 format.

Because the obsolete pbeWithMD5AndDES-CBC algorithm does not authenticate the ciphertext, it is vulnerable to padding oracle attacks that can let an attacker recover the plaintext.

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Secrets](https://docs.redpanda.com/connect/configuration/secrets/).

**Type**: `string`

**Default**: `""`

```yaml
# Examples:
password: foo

# ---

password: ${KEY_PASSWORD}
```

### [](#kafka-tls-enable_renegotiation)`kafka.tls.enable_renegotiation`

Whether to allow the remote server to request renegotiation. Enable this option if you’re seeing the error message `local error: tls: no renegotiation`.

Requires version 3.45.0 or later.

**Type**: `bool`

**Default**: `false`

### [](#kafka-tls-enabled)`kafka.tls.enabled`

Whether custom TLS settings are enabled.

**Type**: `bool`

**Default**: `false`

### [](#kafka-tls-root_cas)`kafka.tls.root_cas`

Specify a root certificate authority to use (optional). This is a string that represents a certificate chain from the parent-trusted root certificate, through possible intermediate signing certificates, to the host certificate. Use either this field for inline certificate data or `root_cas_file` for file-based certificate loading.

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Secrets](https://docs.redpanda.com/connect/configuration/secrets/).

**Type**: `string`

**Default**: `""`

```yaml
# Examples:
root_cas: |-
  -----BEGIN CERTIFICATE-----
  ...
  -----END CERTIFICATE-----
```

### [](#kafka-tls-root_cas_file)`kafka.tls.root_cas_file`

Specify the path to a root certificate authority file (optional). This is a file, often with a `.pem` extension, which contains a certificate chain from the parent-trusted root certificate, through possible intermediate signing certificates, to the host certificate. Use either this field for file-based certificate loading or `root_cas` for inline certificate data.

**Type**: `string`

**Default**: `""`

```yaml
# Examples:
root_cas_file: ./root_cas.pem
```

### [](#kafka-tls-skip_cert_verify)`kafka.tls.skip_cert_verify`

Whether to skip server-side certificate verification. Set to `true` only for testing environments as this reduces security by disabling certificate validation. When using self-signed certificates or in development, this may be necessary, but should never be used in production. Consider using `root_cas` or `root_cas_file` to specify trusted certificates instead of disabling verification entirely.

**Type**: `bool`

**Default**: `false`

### [](#kafka-topic)`kafka.topic`

The Kafka topic to write messages to (required). This field supports [interpolation functions](https://docs.redpanda.com/connect/configuration/interpolation/#bloblang-queries).

**Type**: `string`

### [](#route_to_consumer)`route_to_consumer`

The route to the Kafka consumer. For example, `/project/default/service/forward_to_orders_consumer/secure/api` would connect to a consumer exposed through a relay named `orders_consumer`.

**Type**: `string`

**Default**: `/ip4/127.0.0.1/tcp/6262`

### [](#route_to_kafka_outlet)`route_to_kafka_outlet`

The route to reach the Kafka Outlet of your Ockam portal. For example, `/project/default`.

**Type**: `string`

**Default**: `self`