[WIP][1000% vibed] Add RestateKafkaIntegration CRD (restate.dev/v1alpha1)

[slinkydeveloper]

Runs the Restate Kafka ingress integration github.com/restatedev/ingress-integration-kafka as an operator-managed Deployment: a container that consumes Kafka topics and turns each record into a Restate invocation.

Usage example:

apiVersion: restate.dev/v1alpha1
kind: RestateKafkaIntegration
metadata:
  name: orders-to-restate
spec:
  replicas: 2
  restate:
    ingress:
      cluster: restate          # -> http://restate.restate.svc.cluster.local:8080
  config: |
    bootstrap.servers=kafka.kafka.svc.cluster.local:9092
    group.id=orders-to-restate
    topics=orders
    auto.offset.reset=earliest
    restate.record.mapper.service=OrderService
    restate.record.mapper.handler=onKafkaEvent

Another example with secrets:

apiVersion: v1
kind: Secret
metadata:
  name: kafka-config
stringData:
  config.properties: |
    bootstrap.servers=broker.example:9093
    group.id=orders-to-restate
    topics=orders
    security.protocol=SASL_SSL
    sasl.mechanism=PLAIN
    sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="user" password="hunter2";
    restate.record.mapper.service=OrderService
    restate.record.mapper.handler=onKafkaEvent
---
apiVersion: restate.dev/v1alpha1
kind: RestateKafkaIntegration
metadata:
  name: orders-to-restate
spec:
  restate:
    ingress:
      cluster: restate
  configFrom:
    secretRef:
      name: kafka-config     # `key` defaults to config.properties

--- Slop slop slop ---

Only the Restate destination is modelled as CRD fields:

• spec.restate.ingress is the same cluster/cloud/service/url union as RestateDeployment's spec.restate.register, resolved to the ingress port (8080) rather than the admin port.
• spec.restate.authToken references a Secret key in the custom resource's own namespace, passed through as RESTATE_AUTH_TOKEN. Required with ingress.cloud, since the RestateCloudEnvironment's own credentials live in the operator's namespace and copying them across namespaces would mean granting the operator cluster-wide read access to Secrets.

Everything Kafka-side -- connection, consumer settings, record mapper, metrics, retry policy -- is the container's own configuration surface and is passed through verbatim as a Java .properties file, inline via spec.config (held in a ConfigMap the operator owns, digested into the pod template so an edit rolls the pods) or from a Secret/ConfigMap via spec.configFrom. Mirroring those options as typed fields would mean a CRD change for every upstream addition, for no validation worth having.

spec.template is a partial pod template strategic-merged over the generated one: objects merge key by key, and lists whose Kubernetes merge key we know merge entry by entry, so overriding a container's image does not drop its ports and adding an env var does not drop the operator's. The labels backing the Deployment's immutable selector are re-applied after the merge.

The generated pod has no probes on purpose: the image exposes no health endpoint and exits non-zero on unrecoverable errors, so the restart policy is the real liveness mechanism, and a probe on the metrics port would break for anyone setting restate.metrics.enabled=false. Rollouts use maxSurge: 0 to avoid an extra Kafka rebalance.

No finalizer: there is nothing to deregister, so owner references and garbage collection are enough.

RBAC gains cluster-wide apps/deployments and delete on configmaps, since unlike the RestateCloudEnvironment tunnel these children land in the user's namespace.

Note: reaching a RestateCluster's ingress requires spec.security.networkPeers.ingress on that cluster -- documented in the README and examples, since the failure otherwise presents as a hang.

Claude-Session: https://claude.ai/code/session_01PNVbxD7hyaHFyCQTowMdXi