# yaml-language-server: $schema=https://raw.githubusercontent.com/confluentinc/kcp/main/internal/manifest/gatewaymigration.schema.json
#
# A STATIC migration (all-at-once, AAO): every selected topic cuts over together
# across a STATIC Gateway route.
#
#   kcp migration cutover execute --migration-yaml gateway-static-migration.yaml
#   kcp migration cutover lag-check --migration-yaml gateway-static-migration.yaml
#
# The migration type is declared by which section of spec is present: exactly one
# of staticMigration, dynamicMigration or dynamicToStaticConversion. kcp checks
# the declaration against the live Gateway CR and refuses to run if the route's
# mode disagrees (a staticMigration needs a static route).
#
# Paths in this file resolve relative to the PROCESS working directory, not to
# the file's own location. The one exception is spec.gateway.kubeconfig, where a
# leading ~/ is expanded.
#
# Every credentials field is a PATH to a credentials file — inline secret blocks
# are not accepted — so this manifest carries no secret material itself. Keep the
# referenced credentials files 0600.

apiVersion: kcp.confluent.io/v1alpha1
kind: GatewayMigration

metadata:
  # The migration identity, used as its migration_id label in logs and output.
  name: msk-prod-to-cc-batch-1

spec:
  # kcp only ever READS from the source.
  source:
    type: msk # msk | apache-kafka. Gates auth: iam is msk-only.
    bootstrapServers:
      - b-1.msk-prod.abc123.c2.kafka.us-east-1.amazonaws.com:9096
      - b-2.msk-prod.abc123.c2.kafka.us-east-1.amazonaws.com:9096
    # Path to the source Kafka credentials file (auth only). Its top-level shape
    # for MSK, e.g.:
    #   sasl_scram: { username: ..., password: ..., mechanism: SHA512 }
    # (MSK is SHA512-only; iam is also valid for an MSK source.)
    credentials: /etc/kcp/source-creds.yaml

  target:
    type: confluent-cloud # confluent-cloud | confluent-platform
    clusterId: lkc-abc123 # required for BOTH destination types
    kafka:
      bootstrapServers:
        - pkc-xxxxx.us-east-1.aws.confluent.cloud:9092
      restEndpoint: https://pkc-xxxxx.us-east-1.aws.confluent.cloud:443
      # Path to the destination KAFKA credentials file — the leg dialled directly
      # to bootstrapServers to read destination-side offsets. Its top-level shape
      # accepts sasl_plain, sasl_scram, mtls, unauthenticated_tls, or
      # unauthenticated_plaintext — iam is rejected (the destination is Confluent
      # Cloud/Platform, never MSK). Examples:
      #   sasl_plain: { username: ..., password: ..., tls: true }   # Confluent Cloud
      #   sasl_scram: { username: ..., password: ..., mechanism: SHA256, ca_cert: ... }
      #   mtls:       { client_cert: ..., client_key: ..., ca_cert: ... }
      clusterCredentials: /etc/kcp/dest-kafka-creds.yaml

  clusterLink:
    name: msk-to-cc # the link must ALREADY EXIST on the destination
    # Repeats spec.target.kafka.bootstrapServers, for self-documentation.
    bootstrapServers:
      - pkc-xxxxx.us-east-1.aws.confluent.cloud:9092
    # Path to the cluster-link REST credentials file — the credential that calls
    # restEndpoint's Admin REST API, which is what actually manages the cluster
    # link (status, list/promote mirror topics, alter configs). Always required;
    # never derived. Its top-level shape is one of: api_key/api_secret, basic,
    # bearer, or mtls. Examples:
    #   api_key: ...            # + api_secret: ...   (Confluent Cloud)
    #   basic:  { username: ..., password: ... }      (CP Basic realm)
    #   bearer: { token: ... }                        (MDS/RBAC)
    #   mtls:   { client_cert: ..., client_key: ... }
    linkCredentials: /etc/kcp/link-creds.yaml

  gateway:
    namespace: confluent
    # kubeconfig: ~/.kube/config          # optional (unset: in-cluster in a pod, else ~/.kube/config); a leading ~/ IS expanded
    cr-name: gateway-initial # a Kubernetes object NAME, read live

  staticMigration:
    # route names the route to fence/switch and the streaming domain it switches
    # to. kcp reads the live initial CR, injects fence: {scope: ALL, errorCode:
    # BROKER_NOT_AVAILABLE} onto the named route, and applies the patched CR —
    # there is no separate fenced-CR file. At cutover kcp derives the switched CR
    # the same way, flipping the route's streamingDomain to the target declared
    # here. This is safe with NO secret or auth change at cutover only because
    # the route's security.cluster already carries pre-staged ("redundant") swap
    # auth for the target domain; kcp checks that staging on every run, before
    # anything is fenced.
    route:
      name: migration-route # a pre-existing STATIC route in the initial CR
      targetStreamingDomain: confluent-cloud

    # topicGroup pairs the topic selection that migrates. Exactly one entry.
    #
    # topics is a flat list of LITERAL topic names — not globs. topicPatterns is
    # a list of anchored full-match regular expressions (RE2), matched against the
    # topic names on the source cluster. At least one of the two is REQUIRED; both
    # may be set (their union is used). Every selected topic must be a mirror
    # topic on the cluster link, or the run is refused. To cut over EVERY source
    # topic, use the match-all pattern: topicPatterns: ['.*'].
    #
    # Note the selection has NO effect on `lag-check`, which always watches every
    # mirror.
    topicGroup:
      - topics: ['t1.order', 't1.inventory', 't2.order']
        # topicPatterns: ['orders\..*']

    # OPTIONAL. Disable the link's consumer.offset.sync.enable right after the
    # fence, and set it back to `baseline` after the switch. A run that stops
    # before that restore leaves it owed, and re-running execute performs it.
    consumerOffsetSync:
      pause: true
      # The state consumer.offset.sync.enable was in BEFORE this migration
      # started ("enabled" or "disabled"). Required when pause is set, since a
      # resumed run cannot observe the pre-migration value itself.
      baseline: enabled

    # OPTIONAL, and read FRESH on every execute, so a change takes effect on the
    # next run. Each value is a DEFAULT: `kcp migration cutover execute` can override any
    # of them for a single run with the matching flag (e.g.
    # --detect-unrouted-producers-duration 60s), no edit needed.
    policies:
      lagThreshold: 0
      promoteBatchSize: 100 # 0 = promote all at once
      rolloutTimeout: 10m # 0 = no deadline
      detectUnroutedProducersDuration: 30s # 0 = CHECK SKIPPED; min 10s when set
      consumerOffsetSyncDrainDuration: 15s # 0 = no wait; needs consumerOffsetSync.pause
      hotReloadTimeout: 90s # 0 = built-in 90s budget; never unbounded
      gatewayConfigPort: 0 # 0 = the gateway default (9180)