grpc-kafka proxy

The grpc-kafka proxy binding adapts grpc request-response streams to kafka topic streams.

grpc_kafka_proxy:
  type: grpc-kafka
  kind: proxy
  options:
    idempotency:
      metadata: idempotency-key
    reliability:
      field: 32767
      metadata: last-message-id
    correlation:
      headers:
        service: zilla:service
        method: zilla:method
        correlation-id: zilla:correlation-id
        reply-to: zilla:reply-to
  routes:
    - when:
        - method: example.FanoutService/*
          metadata:
            custom-text: custom value
            custom-binary:
              base64: Y3VzdG9tIHZhbHVl
      exit: kafka_cache_client
      with:
        capability: fetch
        topic: messages
        filters:
          key: custom-key
          headers:
            custom-text: custom-value
    - when:
        - method: example.EchoService/*
          metadata:
            custom-text: custom value
            custom-binary:
              base64: Y3VzdG9tIHZhbHVl
      exit: kafka_cache_client
      with:
        capability: produce
        topic: requests
        acks: leader_only
        key: custom-key
        overrides:
          custom-text: custom-value
        reply-to: responses

Configuration (* required)

options

object

The grpc-kafka specific options.

options:
  idempotency:
    metadata: idempotency-key
  reliability:
    field: 32767
    metadata: last-message-id
  correlation:
    headers:
      service: zilla:service
      method: zilla:method
      correlation-id: zilla:correlation-id
      reply-to: zilla:reply-to

options.idempotency

object

Metadata header used to specify the idempotency key when adapting grpc request-response streams to kafka topic streams.

idempotency.metadata

string | Default: idempotency-key

The grpc metadata header name for idempotency key.

options.reliability

object

Properties used when handling stream recovery.

reliability.field

integer | Default: 32767 Minimum: 1 Maximum: 536870911

The grpc unknown field number to send the message-id.

reliability.metadata

string | Default: last-message-id

The grpc metadata header name for the last message-id seen when resuming a stream.

options.correlation

object

Kafka request message headers injected when adapting grpc request-response streams to kafka topic streams.

correlation.headers*

object

Kafka request message reply to and correlation id header names injected when adapting grpc request-response streams to kafka topic streams.

headers.service

string | Default: zilla:service

Kafka header name for grpc service.

headers.method

string | Default: zilla:method

Kafka header name for grpc method.

headers.correlation-id

string | Default: zilla:correlation-id

Kafka header name for request-response correlation identifier.

headers.reply-to

string | Default: zilla:reply-to

Kafka header name for reply-to topic.

routes*

array of object

Conditional grpc-kafka specific routes.

routes:
  - guarded:
      my_guard:
        - read:messages
    when:
      - service: example.FanoutService
        metadata:
          custom-text: custom value
          custom-binary:
            base64: Y3VzdG9tIHZhbHVl
    exit: kafka_cache_client
    with:
      capability: fetch
      topic: messages
      filters:
        key: custom-key
        headers:
          custom-text: custom-value
  - guarded:
      my_guard:
        - echo:messages
    when:
      - method: example.EchoService/*
        metadata:
          custom-text: custom value
          custom-binary:
            base64: Y3VzdG9tIHZhbHVl
    exit: kafka_cache_client
    with:
      capability: produce
      topic: requests
      acks: leader_only
      key: custom-key
      overrides:
        custom-text: custom-value
      reply-to: responses

routes[].guarded

object as map of named array of string

Roles required by named guard.

routes:
  - guarded:
      my_guard:
        - read:messages

routes[].when

array of object

List of conditions (any match) to match this route when adapting grpc request-response streams to kafka topic streams. Read more: When a route matches

routes:
  - when:
      - method: example.EchoService/*
        metadata:
          custom-text: custom value
          custom-binary:
            base64: Y3VzdG9tIHZhbHVl

when[].method

string | Pattern: ^(?<Service>[^/]+)/(?<Method>[^/]+)

Pattern matching the fully qualified name of a grpc service method, in the format <service>/<method> allowing wildcard * for the method to indicate any method.

when[].metadata

object as map of named string or object properties

Metadata header name value pairs (all match).

Each metadata header value can be string or object with base64 property.

metadata.base64

string

Base64 encoded value for binary metadata header.

routes[].exit

string

Next binding when following this route.

routes:
  - when:
    ...
    exit: kafka_cache_client

routes[].with*

object

Defines the route with the fetch capability.

with:
  capability: fetch

Defines the route with the produce capability.

with:
  capability: produce

with.capability: fetch*

const

Kafka parameters for matched route when adapting grpc request-response streams to kafka topic fetch streams.

Routes with fetch capability map grpc Empty requests to a kafka topic, supporting filtered retrieval of messages with a specific key or headers, or unfiltered retrieval of all messages in the topic merged into a unified response.

Filtering can be performed by kafka message key, message headers, or a combination of both message key and headers.

Reliable message delivery is achieved by capturing the value of the reliability field injected into each response stream message at the grpc client, and replaying the value via the reliability metadata header when reestablishing the stream with a new grpc request.

with:
  capability: fetch
  topic: messages
  filters:
    key: custom-key
    headers:
      custom-text: custom-value

with.topic*

string

The name of a Kafka topic.

with.filters

array of object

List of criteria (any match) to this filter. Kafka filters for matched route when adapting grpc request-response streams to kafka topic fetch streams. All specified headers and key must match for the combined criteria to match.

filters[].key

string

The filter criteria for the Kafka message key.

Supports dynamic parameter substitution using the following pattern:

${guarded['jwt'].identity} – Substitutes a guarded identity value.

filters[].headers

object as map of named string properties

The filter criteria for the Kafka message headers.

Supports dynamic parameter substitution using the following pattern:

${guarded['jwt'].identity} – Substitutes a guarded identity value.

with.capability: produce*

const

Kafka parameters for matched route when adapting grpc request-response streams to kafka topic produce streams.

Routes with produce capability map any grpc request-response to a correlated stream of kafka messages. The grpc request message(s) are sent to a requests topic, with a zilla:correlation-id header. When the request message(s) are received and processed by the kafka requests topic consumer, it produces response message(s) to the responses topic, with the same zilla:correlation-id header to correlate the response.

Requests including an idempotency-key grpc metadata header can be replayed and safely receive the same response. This requires the kafka consumer to detect and ignore the duplicate request with the same idempotency-key and zilla:correlation-id.

with:
  capability: produce
  topic: requests
  acks: leader_only
  key: custom-key
  overrides:
    custom-text: custom-value
  reply-to: responses

with.topic*

string

The name of a Kafka topic for requests.

with.acks

enum [ none, leader_only, in_sync_replicas ] | Default: in_sync_replicas

Kafka acknowledgment mode

with.key

string

The Kafka message key to include with each message.

Supports dynamic parameter substitution using the following pattern:

${guarded['jwt'].identity} – Substitutes a guarded identity value.

with.overrides

object as map of named string properties

The Kafka message headers to inject with each message.

Supports dynamic parameter substitution using the following pattern:

${guarded['jwt'].identity} – Substitutes a guarded identity value.

with.reply-to*

string

The name of the Kafka topic for correlated responses.

telemetry

object

Defines the desired telemetry for the binding.

telemetry.metrics

array

Telemetry metrics to track

telemetry:
  metrics:
    - stream.*
    - grpc.*