Conduktor Gateway

apiVersion: gateway/v2
kind: VirtualCluster
metadata:
  name: payments-team
spec:
  type: Standard
  aclEnabled: true
  superUsers:
    - payments-admin

apiVersion: gateway/v2
kind: VirtualCluster
metadata:
  name: orders-team
spec:
  type: Standard
  aclEnabled: true
  superUsers:
    - orders-admin

apiVersion: gateway/v2
kind: AliasTopic
metadata:
  name: customers
  vCluster: partner-team
spec:
  physicalName: internal-crm-customers

apiVersion: gateway/v2
kind: AliasTopic
metadata:
  name: orders
  vCluster: partner-team
spec:
  physicalName: internal-billing-orders

# docker-compose.yml
schema-registry-proxy:
  ports:
    - "8080:8080"
    - "9464:9464"  # Prometheus metrics
  environment:
    # Schema Registry backend
    CONFLUENT_SCHEMA_REGISTRY_URL: http://schema-registry:8081

    # JWT authentication via OIDC provider
    AUTH_PROVIDER: jwt
    JWT_JWKS_URL: http://keycloak:8080/realms/kafka/protocol/openid-connect/certs
    JWT_SUBJECT_CLAIM_NAME: sub

    # Dynamic permissions via Kafka
    AUTH_USE_REACTIVE_CONFIG: true
    KAFKA_BOOTSTRAP_SERVERS: kafka:9092

    # OpenTelemetry tracing
    OTEL_EXPORTER_OTLP_ENDPOINT: http://otel-collector:4317

# Permission format (Kafka topic: _conduktor_srp_commands)
# Key: service account name
# Value: permission entries

# Admin — full access to all subjects
{ "permissions": [
    { "subject": "*",
      "permissionType": "WRITE" }
]}

# Payments team — prefix matching
{ "permissions": [
    { "subject": "payments-*",
      "permissionType": "WRITE" },
    { "subject": "orders-*",
      "permissionType": "READ" }
]}

# Analytics — read-only, all subjects
{ "permissions": [
    { "subject": "*",
      "permissionType": "READ" }
]}

# Get a JWT token (client credentials flow)
TOKEN=$(curl -s -X POST \
  http://keycloak:8080/realms/kafka/protocol/openid-connect/token \
  -d "grant_type=client_credentials" \
  -d "client_id=payments-service" \
  -d "client_secret=secret" | jq -r '.access_token')

# Register a schema through the proxy
curl -X POST http://localhost:8080/subjects/payments-value/versions \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"schema": "{\"type\":\"record\",\"name\":\"Payment\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"}]}"}'

# Response: {"id": 1}
# Metrics: schema_registry_requests_total +1
# Trace: full span in Jaeger/OTLP

# Gateway cluster configuration
config:
  main:
    bootstrap.servers: kafka-primary:9092
    security.protocol: SASL_SSL
    sasl.mechanism: PLAIN

  failover:
    bootstrap.servers: kafka-secondary:9092
    gateway.roles: failover

# Switch from main → failover
curl -X POST 'http://localhost:8888/gateway/v2/cluster-switching' \
  -H 'Content-Type: application/json' \
  -d '{"fromPhysicalCluster": "main", "toPhysicalCluster": "failover"}'

# Chaos testing interceptor
apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: chaos-broken-broker
spec:
  pluginClass: io.conduktor.gateway.interceptor.chaos.SimulateBrokenBrokersPlugin
  priority: 100
  config:
    rateInPercent: 100
    errorMap:
      FETCH: UNKNOWN_SERVER_ERROR
      PRODUCE: CORRUPT_MESSAGE

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: encrypt-pii-fields
spec:
  pluginClass: io.conduktor.gateway.interceptor.EncryptPlugin
  priority: 100
  config:
    topic: "customers.*"
    recordValue:
      fields:
        - fieldName: ssn
          keySecretId: "vault-kms://vault:8200/transit/keys/pii-key"
          algorithm: AES256_GCM
        - fieldName: creditCard.number
          keySecretId: "vault-kms://vault:8200/transit/keys/payment-key"
          algorithm: AES256_GCM
        - fieldName: email
          keySecretId: "in-memory-kms://email-key"
          algorithm: AES128_GCM
    kmsConfig:
      vault:
        uri: http://vault:8200
        type: TOKEN
        token: ${VAULT_TOKEN}

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: encrypt-with-kms
spec:
  pluginClass: io.conduktor.gateway.interceptor.EncryptPlugin
  priority: 100
  config:
    topic: ".*"
    recordValue:
      fields:
        - fieldName: ssn
          keySecretId: "vault-kms://..."
        - fieldName: payment.cardNumber
          keySecretId: "aws-kms://..."
        - fieldName: healthRecord
          keySecretId: "azure-kms://..."
    kmsConfig:
      vault:
        uri: http://vault:8200
        type: APP_ROLE
      aws:
        basicCredentials:
          accessKey: ${AWS_ACCESS_KEY}
      azure:
        tokenCredential:
          tenantId: ${AZURE_TENANT_ID}

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: inject-tracking-headers
spec:
  pluginClass: io.conduktor.gateway.interceptor.DynamicHeaderInjectionPlugin
  priority: 100
  config:
    topic: "orders.*"
    headers:
      X-Source-Region: "${GATEWAY_REGION}"
      X-Gateway-Timestamp: "${timestamp}"
    overrideIfExists: true

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: remove-internal-headers
spec:
  pluginClass: io.conduktor.gateway.interceptor.safeguard.MessageHeaderRemovalPlugin
  priority: 100
  config:
    topic: ".*"
    headerKeyRegex: "^X-Internal-.*"