risingwave proxy

The risingwave proxy binding for adapting risingwave request-response streams.

risingwave_proxy:
  type: risingwave
  kind: proxy
  options:
    udf:
      - server: http://localhost:8815
      - server: http://localhost:8816
        language: python
    kafka:
      properties:
        bootstrap.server: localhost:9092
      format:
        model: json
        catalog:
          catalog0:
            - strategy: topic
  routes:
    - exit: south_pgsql_client
      when:
        - commands:
            - "CREATE TOPIC"
            - "ALTER TOPIC"
            - "DROP TOPIC"
  exit: pgsql_client

Configuration (* required)

catalog

object as map of named array

To map defined catalog for schema retrieval based on catalog specific parameters. Any of the possible combination can be configured.

id

catalog:
  my_catalog:
    - id: unique-catalog-id

---

strategy

catalog:
  my_catalog:
    - strategy: topic

---

subject

catalog:
  my_catalog:
    - subject: http

catalog[].strategy*

enum [ topic ]

To determine the subject based on the specified strategy

catalog[].subject*

string

Unique identifier for schema categorization in the catalog.

catalog[].version

string | Default: latest

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

catalog[].id*

integer

Define specific schema id to refer from catalog.

options

object

risingwave specific options.

options:
  udf:
    - server: http://localhost:8815
    - server: http://localhost:8816
      language: python
  kafka:
    properties:
      bootstrap.server: localhost:9092
    format:
      model: json
      catalog:
        catalog0:
          - strategy: topic

options.udf

array of object

Define UDF server

udf[].server

string | Pattern: ^([a-zA-Z0-9\\\\.-]+)(:(\\\\{[a-zA-Z_]+\\\\}|[0-9]+))?$

UDF Server location

udf[].language

enum [ java, python ] | Default: java

UDF language

options.kafka

object

Kafka properties and message formatting

kafka.properties

object

properties.bootstrap.server

string

Kafka bootstrap server

kafka.format

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

format.model*

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

A schema or type to validate the request content. Refer to the individual model docs for type specific implementation.

routes

array

Conditional risingwave specific routes.

Correlated Request-Response route:

routes:
  - exit: south_pgsql_client
    when:
    - commands:
        - "CREATE TOPIC"
        - "ALTER TOPIC"
        - "DROP TOPIC"

routes[].when

array

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

routes:
  - when:
      - commands:
          - "CREATE TOPIC"

when[].commands

array of enum [ CREATE TOPIC, ALTER TOPIC, DROP TOPIC ]

routes[].guarded

object as map of named array of string

List of roles required by each named guard to authorize this route.

routes[].exit

string

Next binding when following this route.

routes:
  - when:
    ...
    exit: south_pgsql_client

exit*

string

Default exit binding when no conditional routes are viable.

telemetry

object

Defines the desired telemetry for the binding.

telemetry.metrics

array

Telemetry metrics to track

telemetry:
  metrics:
    - stream.*

telemetry.attributes

object as map of named string properties

Telemetry attributes to add to the binding metrics.