Connect

ockam_kafka

Type:

Available in: Self-Managed

Uses Ockam 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 using Ockam Command.

Ockam 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.

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

Introduced in version 4.33.0.

  • Common

  • Advanced

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: []
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

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

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

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[]

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

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.

Type: string

identity_name

The name of the Ockam identity to use. If this value is not provided, the default Ockam identity is automatically generated and used (optional).

Type: string

kafka

Type: object

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

Configure a batching policy.

Type: object

# 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

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

A Bloblang query that returns a boolean value indicating whether a message should end a batch.

Type: string

Default: ""

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

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

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: ""

# Examples:
period: 1s

# ---

period: 1m

# ---

period: 500ms

kafka.batching.processors[]

For aggregating and archiving message batches, you can add a list of processors 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

# Examples:
processors:
  - archive:
      format: concatenate


# ---

processors:
  - archive:
      format: lines


# ---

processors:
  - archive:
      format: json_array

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

# Examples:
broker_write_max_bytes: 128MB

# ---

broker_write_max_bytes: 50mib

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

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

Populates a key for each message (optional). This field supports interpolation functions.

Type: string

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

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

# Examples:
max_message_bytes: 100MB

# ---

max_message_bytes: 50mib

kafka.metadata

Determines which metadata values are added to messages as headers.

Type: object

kafka.metadata.include_patterns[]

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

Type: array

Default: []

# Examples:
include_patterns:
  - .*

# ---

include_patterns:
  - _timestamp_unix$

kafka.metadata.include_prefixes[]

A list of explicit metadata key prefixes to match against.

Type: array

Default: []

# Examples:
include_prefixes:
  - foo_
  - bar_

# ---

include_prefixes:
  - kafka_

# ---

include_prefixes:
  - content-

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.

Type: string

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

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[]

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

Type: array

# Examples:
seed_brokers:
  - "localhost:9092"

# ---

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

# ---

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

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

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

This field supports interpolation functions.

Type: string

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

# ---

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

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[]

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: []

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


# ---

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

kafka.tls.client_certs[].cert

A plain text certificate to use.

Type: string

Default: ""

kafka.tls.client_certs[].cert_file

The path of a certificate to use.

Type: string

Default: ""

kafka.tls.client_certs[].key

A plain text certificate key to use.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Secrets.

Type: string

Default: ""

kafka.tls.client_certs[].key_file

The path of a certificate key to use.

Type: string

Default: ""

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.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Secrets.

Type: string

Default: ""

# Examples:
password: foo

# ---

password: ${KEY_PASSWORD}

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

Whether custom TLS settings are enabled.

Type: bool

Default: false

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.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Secrets.

Type: string

Default: ""

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

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: ""

# Examples:
root_cas_file: ./root_cas.pem

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

The Kafka topic to write messages to (required). This field supports interpolation functions.

Type: string

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

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

Type: string

Default: self