# Declare a Kafka virtual cluster > Open a virtual cluster to Kafka clients: AsyncAPI contract, TLS listener or mesh fallback, upstreams, clients and chart. Guide - Since version 1.6.0 - Flexway v1.7.1 - https://flexway.io/en/docs/operations-kafka-virtual-clusters/ > Operations guide: configure, deploy and observe the gateway. What the feature does, product side: [product guide](https://flexway.io/en/docs/product-overview/); calling it: [developer guide](https://flexway.io/en/docs/developer-overview/). The AsyncAPI contract declares the topics served, `flexway.yaml` the listener and the upstream, the clients point at the gateway. What each tenant sees: [Native Kafka clients](https://flexway.io/en/docs/operations-kafka-clients/). ## Steps ### Declare the cluster in the contract An AsyncAPI operation marked `x-flexway-kafka-proxy` is served to the Kafka clients of the cluster it names; `flexway.yaml` declares the `listeners.kafka` listener and the upstream. The full example, here inside a mesh (a port per broker): ```yaml # Kafka clients behind the gateway (1.6.0): producers, consumers and admin tools of any # library reach the virtual cluster "shop" the AsyncAPI contract declares - each tenant inside its own # names, under the same identities as HTTP. Here the fallback inside a mesh, a port per broker; the TLS # listener, brokers named by SNI, is in the guide. version: 1 contracts: asyncapi: ./asyncapi.yaml upstreams: events: protocol: kafka brokers: [kafka-0.kafka:9093, kafka-1.kafka:9093, kafka-2.kafka:9093] tls: { server_name: kafka.internal } sasl: { mechanism: SCRAM-SHA-512, username: flexway, password: "file:/etc/flexway-secrets/kafka/password" } timeout: 5s # a connection to a broker, TLS and SASL included auth: issuers: - name: customers discovery: https://login.example.net/.well-known/openid-configuration audience: kafka tenant_claim: org # every topic, group and transaction is the tenant's api_keys: store: "file:./api-keys.yaml" # PLAIN: the key's id as the user, the key as the password listeners: kafka: plaintext: mesh_only: true # clear text only where the mesh encrypts between pods advertised_host: flexway-kafka.flexway.svc.cluster.local clusters: shop: { ports: { from: 19092, to: 19095 }, first_node_id: 0 } # bootstrap 19092, node n at 19093+n ``` ```yaml title="asyncapi.yaml" # examples/kafka-proxy/asyncapi.yaml asyncapi: 3.0.0 info: { title: Shop events, version: "1.0.0" } channels: orders: address: "{tenant}.orders" parameters: tenant: { x-flexway-bind: tenant } messages: order: { payload: { type: object, required: [id], properties: { id: { type: string } } } } operations: produceOrders: action: send channel: { $ref: "#/channels/orders" } x-flexway-upstream: events x-flexway-kafka-proxy: { cluster: shop, transactions: true } security: - $ref: "#/components/securitySchemes/apiKey" - $ref: "#/components/securitySchemes/ordersWrite" consumeOrders: action: receive channel: { $ref: "#/channels/orders" } x-flexway-upstream: events x-flexway-kafka-proxy: { cluster: shop, admin_scope: "orders:admin" } security: - $ref: "#/components/securitySchemes/apiKey" - $ref: "#/components/securitySchemes/ordersRead" components: securitySchemes: apiKey: { type: plain } # SASL PLAIN: an API key bound to the operation ordersWrite: # SASL OAUTHBEARER: a token carrying the scope type: oauth2 flows: { clientCredentials: { tokenUrl: "https://login.example.net/token", availableScopes: { "orders:write": Produce orders } } } scopes: ["orders:write"] ordersRead: type: oauth2 flows: { clientCredentials: { tokenUrl: "https://login.example.net/token", availableScopes: { "orders:read": Read orders } } } scopes: ["orders:read"] ``` ### The TLS listener, brokers named by SNI One port, the SNI name says which broker is meant - `bootstrap..` to bootstrap, `broker-..` for each; `Metadata` and `FindCoordinator` answer these names. The certificate covers `*..` for each virtual cluster (a wildcard covers one label); every such name resolves to the gateway (a wildcard DNS record, or the chart's `LoadBalancer` `Service`). ```yaml title="flexway.yaml" # flexway.yaml - the TLS listener; the rest as in the example above listeners: kafka: address: ":9093" tls: cert_file: /etc/flexway-kafka-tls/tls.crt # *.shop.kafka.example.com key_file: /etc/flexway-kafka-tls/tls.key client_ca_file: /etc/flexway-kafka-tls/ca.crt # optional: client certificates (X509) sni_domain: kafka.example.com advertised_port: 9093 # the port the client sees (the LoadBalancer's) max_request_bytes: 16MiB # an authenticated request; 64 KiB before authentication max_response_bytes: 64MiB # a broker's answer; a Fetch is held below half of it max_connections: 4096 auth_timeout: 10s idle_timeout: 10m auth: mtls: clients: - { id: billing, subject: "CN=billing-app", tenant: acme, operations: [consumeOrders] } ``` ### The plaintext fallback, a port per broker Without TLS there is no SNI; each virtual cluster gets a range - `from` to bootstrap, `from + 1 + n` for node `first_node_id + n` -, announced under `advertised_host`. Refused at start without `mesh_only: true`: clear text only crosses a mesh that encrypts (Linkerd, Istio in strict mTLS). The range covers **every** node id of the cluster; otherwise the client gets an unreachable address and the log says so. ### Upstreams Apache Kafka (`apache/kafka`, Confluent Platform, MSK, Strimzi) and Redpanda are declared alike; the gateway reads each broker's versions and announces to the client only their intersection with what it judges: ```yaml upstreams: events: # Apache Kafka: SCRAM over TLS protocol: kafka brokers: [kafka-0.kafka:9093, kafka-1.kafka:9093, kafka-2.kafka:9093] tls: { ca_file: /etc/flexway-secrets/kafka/ca.crt, server_name: kafka.internal } sasl: { mechanism: SCRAM-SHA-512, username: flexway, password: "file:/etc/flexway-secrets/kafka/password" } events-redpanda: # Redpanda: SASL SCRAM-SHA-256, or the gateway's mTLS protocol: kafka brokers: [redpanda-0.redpanda:9093] tls: { ca_file: /etc/flexway-secrets/redpanda/ca.crt, cert_file: /etc/flexway-secrets/redpanda/tls.crt, key_file: /etc/flexway-secrets/redpanda/tls.key } timeout: 5s # a connection to a broker, TLS and SASL included ``` ### Clients Standard Kafka properties, pointed at the gateway: ```go // franz-go: an API key over SASL PLAIN on TLS cl, err := kgo.NewClient( kgo.SeedBrokers("bootstrap.shop.kafka.example.com:9093"), kgo.DialTLSConfig(&tls.Config{MinVersion: tls.VersionTLS12}), kgo.SASL(plain.Auth{User: "acme-orders-producer", Pass: os.Getenv("FLEXWAY_KAFKA_KEY")}.AsMechanism()), kgo.TransactionalID("tx-orders"), // becomes acme.~t.tx-orders at the broker ) // a token: kgo.SASL(oauth.Oauth(func(ctx context.Context) (oauth.Auth, error) { return oauth.Auth{Token: tok}, nil })) ``` ```properties # librdkafka (confluent-kafka-go, Python, .NET): an OAUTHBEARER token of the declared issuer bootstrap.servers=bootstrap.shop.kafka.example.com:9093 security.protocol=SASL_SSL sasl.mechanisms=OAUTHBEARER sasl.oauthbearer.method=oidc sasl.oauthbearer.client.id=billing sasl.oauthbearer.client.secret=${BILLING_SECRET} sasl.oauthbearer.token.endpoint.url=https://login.example.net/token sasl.oauthbearer.scope=orders:read group.id=billing # becomes acme.~g.billing ``` ```properties # Java client (producer, consumer, AdminClient, Kafka Connect): an API key over PLAIN bootstrap.servers=bootstrap.shop.kafka.example.com:9093 security.protocol=SASL_SSL sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="acme-orders-reader" password="${FLEXWAY_KAFKA_KEY}"; # a client certificate instead: security.protocol=SSL, ssl.keystore.location=..., ssl.truststore.location=... # group.protocol=classic (the default): KIP-848's group protocol is not served ``` ### The chart: clear text inside the mesh only The plaintext ports go into a `Service` of their own, `-flexway-kafka-mesh`, **always** `ClusterIP`; the render fails without `kafka.networkPolicy.from` (the mesh's pods or namespaces). ```yaml title="values.yaml" # renders: the TLS port on a LoadBalancer, the plaintext range kept for the mesh config: listeners: kafka: address: ":9093" tls: { cert_file: /etc/flexway-kafka-tls/tls.crt, key_file: /etc/flexway-kafka-tls/tls.key } sni_domain: kafka.example.com plaintext: { mesh_only: true, advertised_host: orders-flexway-kafka-mesh.apps.svc.cluster.local, clusters: { shop: { ports: { from: 19092, to: 19095 } } } } kafka: service: { type: LoadBalancer } tls: { clusters: [shop], issuerRef: { name: kafka-ca, kind: ClusterIssuer } } networkPolicy: from: [ { namespaceSelector: { matchLabels: { kubernetes.io/metadata.name: apps } } } ] ``` ```yaml title="values.yaml" # refused: clear text with no declared peers kafka: networkPolicy: { from: [] } # Error: execution error at (flexway/templates/kafka.yaml): listeners.kafka.plaintext opens ports in # clear: kafka.networkPolicy.enabled and kafka.networkPolicy.from (the mesh's pods or namespaces) are # required - clear text is for the mesh only ``` The chart also renders the `-flexway-kafka` `Service`, its `NetworkPolicy` and, with cert-manager, each cluster's wildcard certificate (`kafka.tls.clusters`). ## Verify 1. `flexway validate` accepts the configuration; a refusal names the faulty extension or key ([Refused at load](https://flexway.io/en/docs/operations-kafka-reference/#refused-at-load)). 2. A client lists its topics under their short names, and only those its operations open: ```bash kafka-topics --bootstrap-server bootstrap.shop.kafka.example.com:9093 --command-config client.properties --list # orders ``` 3. The broker holds `acme.orders`, and `acme.~g.billing` for the group `billing` ([Names of groups and transactions](https://flexway.io/en/docs/operations-kafka-reference/#names-of-groups-and-transactions)). 4. The admin `/config` publishes the virtual clusters under `kafka`. 5. A refused client shows in the log `kafka authentication refused`, with its cause and never the key; the other cases: [Troubleshooting](https://flexway.io/en/docs/operations-kafka-reference/#troubleshooting). ## Good to know ### Reload The address, TLS, `sni_domain`, `advertised_port`, the fallback range and `max_connections`, `max_total_in_flight_bytes` and `max_tenant_in_flight_bytes` are read at start (`listeners.kafka` is refused on reload if they change); the contract, identities, bounds and timeouts apply to the next request - a connection keeps its identity as long as its credential stays the configuration's in force ([Sessions: revocation, lifetime, certificates](https://flexway.io/en/docs/operations-kafka-reference/#sessions-revocation-lifetime-certificates)), and each request is judged on it. ### Broker ACLs The gateway's principal only needs the topics, groups and transactional ids of the tenants served ([One upstream connection per client connection and broker](https://flexway.io/en/docs/operations-kafka-clients/#one-upstream-connection-per-client-connection-and-broker)), under the names of [Names of groups and transactions](https://flexway.io/en/docs/operations-kafka-reference/#names-of-groups-and-transactions).