ockam_kafka

Beta

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

# Common configuration fields, showing default values
input:
  label: ""
  ockam_kafka:
    kafka:
      seed_brokers: [] # No default (optional)
      topic: "" # No default (required)
      key: "" # No default (optional)
      partition: ${! meta("partition") } # No default (optional)
      metadata:
        include_prefixes: []
        include_patterns: []
      max_in_flight: 10
      batching:
        count: 0
        byte_size: 0
        period: ""
        check: ""
    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: []
# All configuration fields, showing default values
input:
  label: ""
  ockam_kafka:
    kafka:
      seed_brokers: [] # No default (optional)
      topic: "" # No default (required)
      key: "" # No default (optional)
      partitioner: "" # No default (optional)
      partition: ${! meta("partition") } # No default (optional)
      idempotent_write: true
      metadata:
        include_prefixes: []
        include_patterns: []
      max_in_flight: 10
      timeout: 10s
      batching:
        count: 0
        byte_size: 0
        period: ""
        check: ""
        processors: [] # No default (optional)
      max_message_bytes: 1MiB
      broker_write_max_bytes: 100MiB
      compression: "" # No default (optional)
      tls:
        enabled: false
        skip_cert_verify: false
        enable_renegotiation: false
        root_cas: ""
        root_cas_file: ""
        client_certs: []
      timestamp_ms: ${! timestamp_unix_milli() } # 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

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

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

Type: string

kafka.key

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

Type: bool

Default: false

kafka.partitioner

Override the default murmur2 hashing partitioner (optional).

Type: string

Option

Summary

least_backup

Choose the partition with the fewest buffered records. Partitions are selected per batch.

manual

Manually select a partition for each message. To use this option, specify a value in the partition field.

murmur2_hash

Kafka’s default hash algorithm that uses a 32-bit murmur2 hash of the key to compute which partition the record is written to.

round_robin

Does a round-robin of all available partitions for messages. This algorithm has lower throughput and causes higher CPU load on brokers, but is useful if you want to ensure an even distribution of records to partitions.

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

Determines which metadata values are added to messages as headers.

Type: object

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

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

Type: int

Default: 10

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

The period of time after which an incomplete batch is flushed regardless of its size.

Type: string

Default: ""

# Examples
period: 1s
period: 1m
period: 500ms

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.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: array

# Examples
processors:
  - archive:
      format: concatenate
processors:
  - archive:
      format: lines
processors:
  - archive:
      format: json_array

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: 100MiB
max_message_bytes: 50MiB

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: 128MiB

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

Override system defaults with custom TLS settings.

Type: object

kafka.tls.enabled

Whether custom TLS settings are enabled.

Type: bool

Default: false

kafka.tls.skip_cert_verify

Whether to skip server-side certificate verification.

Type: bool

Default: false

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.

Type: bool

Default: false

Requires version 3.45.0 or newer

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.

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.

Type: string

Default: ""

# Examples
root_cas_file: ./root_cas.pem

kafka.tls.client_certs

A list of client certificates to use. For each certificate, specify either the fields cert and key or cert_file and key_file.

Type: array

Default: []

# Examples
client_certs:
  - cert: foo
    key: bar
client_certs:
  - cert_file: ./example.pem
    key_file: ./example.key

kafka.tls.client_certs[].cert

The plain text certificate to use.

Type: string

Default: ""

kafka.tls.client_certs[].key

The 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[].cert_file

The path of a certificate to use.

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

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

The pbeWithMD5AndDES-CBC algorithm does not authenticate ciphertext, and is vulnerable to padding oracle attacks that may allow an attacker to recover the plain text password.

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.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") }

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

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

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

route_to_kafka_outlet

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

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

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

kafka.encrypted_fields

Fields to encrypt in the Kafka messages when the record is a valid JSON map. By default, the whole record is encrypted.

Type: array

Default: []