kafka-proxy proxy

The kafka-proxy proxy binding for adapting kafka topic streams to kafka topic streams.

kafka_proxy:
  type: kafka-proxy
  kind: proxy
  options:
    external:
      host: kafka-#.external.net
      port: 9093
    internal:
      host: b-#.kafka.internal.net
      port: 9094
  exit: tls_client

Template used to build the bearer token presented to the internal broker. Required when mechanism is oauthbearer. Supports ${guarded['my_guard'].credentials} to reference the raw credential string an external guard authorized the session with.

internal:
  authorization:
    credentials:
      mechanism: oauthbearer
      credentials: "Bearer ${guarded['cognito0'].credentials}"

options.topics

array of object

Topic configuration list.

topics[].name

string

Topic name, as observed by the external client.

topics[].alias

string

Template for the internal topic name. Supports ${topic} to reference the topic's own name, along with identity and attribute placeholders such as ${guarded['my_guard'].identity} and ${guarded['my_guard'].attributes.my_attribute}. When omitted, the existing rules apply, either cluster-id prefixed or match the external name.

topics:
  - name: messages
    alias: "${topic}-${guarded['my_guard'].identity}"

topics[].key

enum [ avro, boolean, double, float, int32, int64, json, string ], object

Enforce validation for key.

key.model*

enum [ avro, boolean, double, float, int32, int64, json, string ]

A schema or type to validate the topic's key. Refer to the individual model docs for type specific implementation.

topics[].value

enum [ avro, boolean, double, float, int32, int64, json, string ], object

Enforce validation for value.

value.model*

enum [ avro, boolean, double, float, int32, int64, json, string ]

A schema or type to validate the topic's value. Refer to the individual model docs for type specific implementation.

catalog

object as map of named array

Catalog for schema retrieval. Required when model is avro, json, or protobuf. Applies to both key and value models.

catalog[].id*

integer

Define specific schema id to refer from catalog.

catalog[].strategy*

enum [ topic ]

To determine the subject based on the specified strategy.

catalog[].version

string | Default: latest

Specific iteration or version of a registered schema in the defined catalog.

catalog[].subject*

string

Unique identifier for schema categorization in the catalog.

topics[].limit

string

Topic limit.

topics[].constraints

array of object

Constraints applied to this topic.

constraints[].mode

enum [ enforce, warn ]

Constraint enforcement mode.

constraints[].regex

string

Constraint regex pattern.

constraints[].partitions

object

Partition constraints.

partitions.min

integer | Minimum: 1

Minimum number of partitions.

partitions.max

integer | Minimum: 1

Maximum number of partitions.

constraints[].replicas

object

Replica constraints.

replicas.min

integer | Minimum: 1

Minimum number of replicas.

replicas.max

integer | Minimum: 1

Maximum number of replicas.

constraints[].configs

object as map of named string properties, object as map of named object properties

Topic configuration constraints.

configs.min

integer

Minimum configs.

configs.max

integer

Maximum configs.

configs.enum

array of string

Allowed config values.

constraints[].deletion

enum [ protected ]

Topic deletion constraint.

options.limits

object as map of named object properties

Rate limits by named group.

limits.produce

object

Produce rate limit.

produce.bytes

string | Pattern: ^[0-9]+(\\.[0-9]+)?\\s*(B|kB|KB|MB|mB|GB|gB|TB|tB|b|kb|Kb|mb|Mb|gb|Gb|tb|Tb)/s$

Maximum produce data rate.

options.constraints

array of object

Global connection constraints.

constraints[].client_id

string | Pattern: ^[a-zA-Z0-9_-]*\\*?$

Client ID pattern to match.

constraints[].api_versions

object

API version constraints.

api_versions.min

object as map of named integer properties

Minimum API versions by key.

api_versions.max

object as map of named integer properties

Maximum API versions by key.

constraints[].cooldown

string

Cooldown duration.

constraints[].producer

enum [ idempotent ]

Producer constraint.

constraints[].compression

array of enum [ none, gzip, snappy, lz4, zstd ]

Allowed compression types.

routes

array of object

Conditional kafka-proxy specific routes.

routes[].when

array of object

List of conditions (any match) to match this route.

when[].topic

string

Topic name to match.

when[].api

array of enum [ produce, fetch, metadata, update_metadata, offset_commit, offset_fetch, offset_delete, find_coordinator, list_offsets, sasl_handshake, api_versions, join_group, heartbeat, leave_group, sync_group, init_producer_id, offset_for_leader_epoch, add_partitions_to_transaction, add_offsets_to_transaction, end_transaction, write_transaction_markers, transaction_offset_commit, sasl_authenticate, create_topics, describe_configs, alter_configs, incremental_alter_configs, create_partitions, delete_topics, delete_records, consumer_group_heartbeat, share_group_heartbeat, share_group_describe, leader_and_isr, elect_leaders, alter_partition, alter_partition_reassignments, list_partition_reassignments, describe_topic_partitions, describe_log_directories, alter_replica_log_directories, describe_producers ]