Skip to main content

Kafka Backup on Confluent Cloud: OAuth with Entra ID, identity pools and Azure storage

Kafka Backup Enterprise authenticates to Kafka with SASL/OAUTHBEARER and to Schema Registry with OAuth bearer tokens. Tokens come from Microsoft Entra ID (client credentials, Azure managed identity or AKS workload identity), from any other OIDC provider, or from a token file. Everything is Rust-native: no JVM, no librdkafka and no Azure SDK.

Licence featureoauth (included in the 14-day trial)
Commandsbackup, restore, three-phase-restore, validate-restore, snapshot-groups, validation run, the offset commands with --config <yaml> --cluster source|target, and migrate msk-kraft for MSK IAM
Kafkasasl_mechanism: OAUTHBEARER plus an enterprise.oauth.kafka.<cluster> binding
Schema Registry / DEK Registryauth: { type: oauth, provider: <name>, target_sr_cluster: lsrc-..., identity_pool_id: pool-... }
Released verification

Enterprise 0.7.0 is released. Its public image and downloaded Helm chart passed backup and restore on an Azure VM with Managed Identity and on AKS with Workload Identity. Both restored into a genuinely new Confluent Cloud Standard cluster and empty Schema Registry. Each source and fresh-target restore passed 44 independent checks for original schema IDs, references, deleted versions, records, headers, topic settings and consumer positions. Direct restores used a distinct restore identity without client secrets and wrote Azure checkpoints without Blob delete permission.

API-key and Entra Basic, POST and automatic-pool flows also passed live Cloud checks on an earlier candidate. These fixture checks do not qualify 24-hour operation, projected-token rotation, separate-tenant isolation, encrypted record recovery or representative throughput. Verify identity and role setup in your environment before a production rollout.

1. How it fits together​

Entra ID (or another IdP) ──token──▶ Kafka Backup ──SASL/OAUTHBEARER (+ logicalCluster, identityPoolId)──▶ Confluent Cloud Kafka
                                          └──────────Authorization: Bearer (+ target-sr-cluster)──────────▶ Confluent Cloud Schema Registry
Confluent Cloud: identity provider (issuer + JWKS) → identity pools (CEL filter on the token claims) → role bindings
  1. Kafka Backup gets an access token from the identity provider.
  2. Confluent Cloud validates it against the identity provider you registered (issuer and JWKS), picks the identity pool whose filter matches the token's claims, and authorises with that pool's role bindings.
  3. Kafka Backup caches the token per provider and shares it across every connection; KIP-368 re-authentication renews long-lived connections in place.

2. Microsoft Entra ID​

2.1 The API application (the token audience)​

Register one application that represents "Confluent Cloud for Kafka Backup":

  • Expose an API: set the Application ID URI, e.g. api://kafka-backup-confluent (or the default api://<application-id>).
  • Request v2 tokens. Set "requestedAccessTokenVersion": 2 under api in the app manifest (the older AAD Graph manifest calls it accessTokenAcceptedVersion). Without it Entra issues v1 tokens whose issuer is https://sts.windows.net/<tenant>/, which does not match the v2 issuer you register in Confluent Cloud.
  • Clients request the scope <Application ID URI>/.default. In a v2 token, aud is the API application's application (client) ID (a GUID), and oid identifies the calling service principal. Inspect the actual token claims before choosing a pool filter; do not assume sub equals an object ID.

2.2 The calling identity​

Pick one per environment:

Where Kafka Backup runsIdentityProvider type
Anywhere, with a secretApp registration + client secretoidc_client_credentials
Azure VM / VMSSSystem- or user-assigned managed identityazure_managed_identity
AKSWorkload identity: a user-assigned managed identity (or app registration) federated with the cluster's OIDC issuerazure_workload_identity

Note the identity's object ID (service principal object ID for an app, principal ID for a managed identity): check it against the token's oid claim before configuring the pool filter.

No API permissions or admin consent are needed for the client-credentials .default scope unless the API application defines app roles you want to require.

3. Confluent Cloud​

Standard Kafka cluster creation requires a payment method on the Confluent organisation, including when trial credits remain. In the OSO qualification lab, the API rejected Standard cluster creation with 402 Payment Required: No credit card on file. Confirm this prerequisite before applying Terraform.

3.1 Identity provider​

Accounts & access → Workload identities → Add provider → Microsoft Entra ID:

FieldValue
Issuer URIhttps://login.microsoftonline.com/<tenant-id>/v2.0
JWKS URIhttps://login.microsoftonline.com/<tenant-id>/discovery/v2.0/keys

3.2 Identity pools​

Use separate pools for backup and restore so each gets only its own role bindings. The identity claim is claims.sub; the filter is CEL over the token's claims:

# Backup pool: only the backup identity, only tokens for our API app
claims.aud == "<api-application-id>" && claims.oid == "<backup-identity-object-id>"

# Restore pool
claims.aud == "<api-application-id>" && claims.oid == "<restore-identity-object-id>"

Kafka Backup sends the pool ID as the identityPoolId SASL extension when you configure it (a comma-separated list is allowed). Leave it out to let Confluent Cloud match the token against your pools' filters (auto pool mapping; verify that your organisation has it).

3.3 Role bindings​

The following roles passed live Confluent Cloud tests with the public 0.7.0 image. Scope topic, group and subject CRN patterns to your recovery set.

PoolRoleResourceWhy
BackupDeveloperReadTopics in scope (prefix)Fetch, offsets, topic configs
BackupDeveloperReadConsumer groups in scope (prefix)consumer_group_snapshot (OffsetFetch)
BackupDeveloperReadSchema Registry subjects in scopeSchema backup
Backup and restoreDeveloperReadSchema Registry subject __GLOBALRead global compatibility and mode
RestoreDeveloperWriteTarget topics (prefix)Produce
RestoreDeveloperManageTarget topics (prefix)Create topics
Restore configuration identityCloudClusterAdminRestore Kafka clusterApply topic configuration changes
RestoreDeveloperReadTarget topics and consumer groups (prefix)Phase 3 offset commits (OffsetCommit needs Read on both)
RestoreResourceOwnerSchema Registry subjects in scopeSubject-level IMPORT and schema restore

The backup identity has read-only Kafka and registry permissions. Cloud requires CloudClusterAdmin for topic configuration writes; developer roles can create topics but cannot apply archived settings. This grants wider control on the restore cluster. To keep provisioning separate from record restoration, pre-create topics and settings through a configuration identity, then run the restore with scoped topic, group and subject roles.

The fixture producer also needs DeveloperWrite on its transactional-ID prefix when seeding committed and aborted transactions.

4. Kafka Backup configuration​

OAuth settings live under enterprise.oauth, never under source.security / target.security (the configuration is rejected with a hint if they are). Every key in enterprise.oauth is checked: a misspelt key is an error. ${ENV_VAR} references are expanded, so keep secrets in the environment.

4.1 Kafka with Entra client credentials​

source:
  bootstrap_servers: ["pkc-xxxxx.westeurope.azure.confluent.cloud:9092"]
  security:
    security_protocol: SASL_SSL
    sasl_mechanism: OAUTHBEARER

enterprise:
  oauth:
    providers:
      entra-backup:
        type: oidc_client_credentials
        token_endpoint_url: https://login.microsoftonline.com/<tenant-id>/oauth2/v2.0/token
        # or: issuer: https://login.microsoftonline.com/<tenant-id>/v2.0  (OIDC discovery)
        client_id: <backup app client id>
        client_secret: ${ENTRA_BACKUP_CLIENT_SECRET}
        client_auth: basic          # client_secret_basic (default) or post
        scope: api://kafka-backup-confluent/.default
    kafka:
      source:
        provider: entra-backup
        extensions:
          logicalCluster: lkc-abc123
          identityPoolId: pool-Backup1      # optional; "pool-a,pool-b" also works

A cluster with sasl_mechanism: OAUTHBEARER and no binding, a binding to an undefined provider, a binding on a cluster that does not use OAUTHBEARER, or a token endpoint over plain http:// to anything but localhost is a configuration error before anything connects.

Other OIDC providers (Okta, Auth0, Keycloak) use the same type; audience sends the audience parameter and extra_params adds any other form field. ca_cert adds a CA for a private identity provider.

4.2 Azure managed identity (VM, VMSS)​

enterprise:
  oauth:
    providers:
      mi:
        type: azure_managed_identity
        scope: api://kafka-backup-confluent/.default   # or resource: api://kafka-backup-confluent
        client_id: <user-assigned identity client id>  # omit for the system-assigned identity
    kafka:
      source: { provider: mi, extensions: { logicalCluster: lkc-abc123 } }

Kafka Backup calls IMDS (http://169.254.169.254/metadata/identity/oauth2/token, header Metadata: true); endpoint overrides it for proxies and tests.

4.3 AKS workload identity​

enterprise:
  oauth:
    providers:
      wi:
        type: azure_workload_identity
        scope: api://kafka-backup-confluent/.default
        # tenant_id, client_id, token_file and authority_host default to the
        # AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_FEDERATED_TOKEN_FILE and
        # AZURE_AUTHORITY_HOST variables the workload identity webhook injects.
    kafka:
      source: { provider: wi, extensions: { logicalCluster: lkc-abc123 } }

Label the pod azure.workload.identity/use: "true" and annotate its service account with azure.workload.identity/client-id. Kafka Backup re-reads the projected token file on every refresh and exchanges it for an Entra token with a client assertion.

Use a separate managed identity and federated service account for restoration. Bind it to the restore identity pool and enterprise.oauth.kafka.target; give it the restore topic, group and Schema Registry roles from section 3.3. Annotate the restore service account with that identity's client ID and retain the same workload-identity pod label. The qualification lab used kbq:backup and kbq:restore service accounts with distinct identities.

4.4 Token file and static token​

providers:
  vault:
    type: file
    path: /var/run/secrets/kafka/token      # re-read on every refresh, trimmed
    expires_at_file: /var/run/secrets/kafka/token.expires_at   # optional: RFC 3339 or Unix seconds
  test:
    type: static          # tests only: never refreshed
    token: ${TEST_TOKEN}

Without an expiry from the provider, the JWT exp claim is read (unverified); an opaque token without one is refreshed every five minutes.

4.5 Schema Registry (and the DEK Registry)​

enterprise:
  schema_registry:
    url: https://psrc-xxxxx.westeurope.azure.confluent.cloud
    auth:
      type: oauth
      provider: entra-backup            # shares the Kafka token cache
      target_sr_cluster: lsrc-abc123    # sent as the target-sr-cluster header
      identity_pool_id: pool-Backup1    # sent as Confluent-Identity-Pool-Id; optional

enterprise.encryption falls back to this auth for the DEK Registry, or takes its own auth. The inline form of earlier releases (token_url, client_id, client_secret, scope; client_secret_post) still works, also with target_sr_cluster / identity_pool_id, and does not need the oauth licence; prefer a named provider.

4.6 Restore, offsets and validation​

A restore binds enterprise.oauth.kafka.target (and source when a three-phase restore reads source groups). The offset commands take the cluster from a configuration file:

kafka-backup offset-reset-bulk --path s3://bucket --backup-id daily \
  --groups orders-service --config restore.yaml --cluster target

validation run reads enterprise.oauth from its own file and binds target.

5. Token cache​

One cache per provider name, shared by every connection and by Schema Registry:

Setting (enterprise.oauth.cache)DefaultMeaning
refresh_fraction0.8Refresh after this share of the lifetime...
refresh_before_expiry5m...or this long before expiry, whichever is earlier
clock_skew30sA token is valid while exp - clock_skew > now
min_refresh_interval30sNever refresh sooner than this after a fetch; first backoff step
  • One request at a time per provider; concurrent callers wait for it.
  • Past the refresh time a valid token keeps being used while a background refresh runs. If the identity provider fails, the valid token stays in use, retries back off exponentially (30 s, 60 s, ...) and always try once more before the token stops being valid.
  • Every SASL handshake (initial and re-authentication) asks for a token with at least 60 s left (bounded by 80 % of the token lifetime), so the session is long enough to re-authenticate in place. A slow identity provider holds a handshake up by at most 5 s when a valid token is available.
  • A broker invalid_token rejection or a Schema Registry 401 drops the cached token unless it is younger than min_refresh_interval.

Identity-provider requests time out after 15 s (5 s to connect) and are retried three times on 429, 5xx and connection errors, honouring Retry-After. Redirects are never followed.

6. Licence and grace period​

OAuth needs the oauth feature (trials include it). When a commercial licence expires, OAuth keeps working for 30 days: every run logs a warning with the days left, license info shows them, and the metrics endpoint exposes kafka_backup_enterprise_licence_grace_days_remaining. After the grace period OAuth is refused: the command exits with status 2 and says why. A continuous backup whose licence or grace period ends while it is running stops after its current cycle and exits with status 2. Trials have no grace period. Other enterprise features stop at expiry as before.

7. Observability and security​

  • kafka_backup_enterprise_oauth_token_requests_total{provider,kind,outcome} counts identity-provider requests (on the metrics endpoint whenever OAuth is in use).
  • Tokens and secrets never reach the logs: secrets are redacted in Debug and serialisation, identity-provider error text is scrubbed of secrets and JWT-shaped strings, and a test captures every log line at TRACE to check.
  • Use SASL_SSL. SASL_PLAINTEXT works (for local testing) with a warning, because bearer tokens then cross the network unencrypted.

8. Troubleshooting​

SymptomLikely cause
SASL OAUTHBEARER rejected by broker ... invalid credentials and a WARN with status=invalid_tokenWrong issuer (v1 token: set requestedAccessTokenVersion: 2), wrong audience, no identity pool matches the claims, or JWKS not refreshed yet
token endpoint returned HTTP 401 invalid_clientWrong client secret, or client_auth does not match what the app allows
Cannot change principals during re-authentication (then a reconnect)The token's sub changed (rotated identity or file); harmless, the connection re-authenticates from scratch
Exit status 2Licence refusal: no oauth feature, trial over, or the grace period ended
... needs a token provider from enterprise.oauth--sasl-mechanism OAUTHBEARER without --config/--cluster

9. Terraform​

Resource names for automating the setup (providers azuread, azurerm, confluent):

PurposeResource
API application, v2 tokensazuread_application (api { requested_access_token_version = 2 }, identifier_uris) and azuread_service_principal
Backup / restore client appsazuread_application, azuread_service_principal, azuread_application_password
Managed identityazurerm_user_assigned_identity (attach to the VM or VMSS)
AKS workload identityazurerm_user_assigned_identity + azurerm_federated_identity_credential (or azuread_application_federated_identity_credential for an app registration)
Confluent identity providerconfluent_identity_provider (issuer, jwks_uri)
Poolsconfluent_identity_pool (identity_claim = "claims.sub", filter)
Role bindingsconfluent_role_binding (principal = "User:${confluent_identity_pool.backup.id}", role_name, crn_pattern)
resource "confluent_identity_provider" "entra" {
  display_name = "entra"
  description  = "Microsoft Entra ID"
  issuer       = "https://login.microsoftonline.com/${var.tenant_id}/v2.0"
  jwks_uri     = "https://login.microsoftonline.com/${var.tenant_id}/discovery/v2.0/keys"
}

resource "confluent_identity_pool" "backup" {
  identity_provider { id = confluent_identity_provider.entra.id }
  display_name   = "kafka-backup"
  description    = "Kafka Backup (read)"
  identity_claim = "claims.sub"
  filter         = "claims.aud == \"${azuread_application.confluent_api.client_id}\" && claims.oid == \"${azuread_service_principal.backup.object_id}\""
}

resource "confluent_role_binding" "backup_topics" {
  principal   = "User:${confluent_identity_pool.backup.id}"
  role_name   = "DeveloperRead"
  crn_pattern = "${data.confluent_kafka_cluster.prod.rbac_crn}/kafka=${data.confluent_kafka_cluster.prod.id}/topic=orders-*"
}

10. Azure backup storage​

Use a storage account in a separate tenant where your recovery requirements include compromise of the Kafka platform's administrative identity. A separate subscription gives an additional access boundary within one tenant; it does not isolate tenant-wide administrators. The qualification harness currently uses OSO Development in the same tenant, so it does not prove cross-tenant isolation.

Enable blob versioning, blob soft delete and container soft delete. Choose a retention period that covers your recovery window. These settings support recovery of previous blob versions; they do not provide WORM protection and do not prevent an authorised administrator from changing retention or access.

resource "azurerm_storage_account" "backup" {
  name                     = var.backup_storage_account
  resource_group_name      = azurerm_resource_group.backup.name
  location                 = azurerm_resource_group.backup.location
  account_tier             = "Standard"
  account_replication_type = "LRS"
  min_tls_version          = "TLS1_2"
  blob_properties {
    versioning_enabled = true
    delete_retention_policy { days = 14 }
    container_delete_retention_policy { days = 14 }
  }
}

resource "azurerm_storage_container" "backup" {
  name                  = "kafka-backups"
  storage_account_id    = azurerm_storage_account.backup.id
  container_access_type = "private"
}

Scope storage role assignments to the container. Backup needs read access to its existing manifests and checkpoints as well as write access, so it is not a write-only process. A restore reads the archive and writes its registry reports and durable restore checkpoints; give it container-scoped read/write permission without deletion of backup objects. Pruning is a separate operation that requires delete permission.

For cross-tenant storage, register the storage identity in the storage tenant and use that tenant's credentials. Kafka OAuth and storage authentication are separate: the storage tenant_id, client_id and client_secret need not match the Kafka provider. This cross-tenant arrangement remains a live qualification item.

storage:
  backend: azure
  account_name: recoveryaccount
  container_name: kafka-backups
  prefix: production/cluster-a
  tenant_id: ${STORAGE_TENANT_ID}
  client_id: ${STORAGE_CLIENT_ID}
  client_secret: ${STORAGE_CLIENT_SECRET}

For AKS storage authentication, replace client_secret with use_workload_identity: true, and use the tenant/client ID of the federated storage identity. For a VM, omit secret/key/SAS settings and set client_id to the attached user-assigned identity. SAS tokens and account keys are also supported. A connection string and a logged-in Azure CLI session are not supported storage credentials.

Use a distinct backup ID for each job. Two jobs must not write to the same backup ID. Keep non-production backups separate from production.

11. Terraform setup sequence​

The qualification harness has a validated configuration for these resources:

StepTerraform resourcesCheck before continuing
Ownershipdata.confluent_organization, data.azurerm_client_config, terraform_data.ownership_guardCloud API key belongs to the confirmed OSO organisation; Azure tenant/subscription match
Source and recovery environmentsconfluent_environment with Essentials governance, confluent_kafka_cluster with standardBoth environments have independent Kafka and Schema Registry IDs
Schema Registry discoverydata.confluent_schema_registry_clusterRegistry endpoint belongs to its intended environment
Entra audience and callersazuread_application, azuread_application_identifier_uri, azuread_service_principal, azuread_application_passwordActual tokens contain the v2 issuer, expected audience and calling object
Confluent OAuthconfluent_identity_provider, confluent_identity_pool, confluent_role_bindingEach provider/role succeeds, and access outside its scope fails
API-key baselineconfluent_service_account, confluent_api_key, confluent_role_bindingBackup identity can read; restore identity can create and write target topics
Azure storageazurerm_storage_account, azurerm_storage_container, azurerm_role_assignmentPrevious versions and deleted blobs can be recovered
Workload identityazurerm_user_assigned_identity, azurerm_kubernetes_cluster, azurerm_federated_identity_credentialProjected token is refreshed and exchanged during a real AKS run

Keep Terraform state, Cloud API keys, client secrets, saved plans and generated configuration outside source control. The state contains secrets even when Terraform marks its outputs as sensitive. Review the plan before applying it. Destroy the qualification resources after each run and confirm their removal in both provider consoles.

Environment A and B in one organisation prove an independent environment restore. A recovery into another organisation needs a second, separately confirmed organisation and its own credentials; the current harness does not prove that scenario.

12. Recover into a fresh environment​

  1. Recreate the Kafka cluster, identity provider, pools and role bindings. Keep the backup storage outside the compromised environment's administration.
  2. Configure the destination Schema Registry under enterprise.schema_registry.url. Use into_existing for a shared environment registry and subject-level IMPORT. Use preserve_ids only for an empty registry when the identity has permission to change global and context modes.
  3. Preview the registry with validate-restore --schema-only --format json. Resolve ID/version conflicts before replay. An ID that belongs to different content must fail in strict mode; enable rewrite_ids only when the intended record framing and ID mapping have been checked.
  4. Restore the registry, then topics and records, with restore.target_profile: confluent_cloud. This profile adapts topic settings and replication defaults; actual cluster permissions and rejected settings still need verification.
  5. Enable backup.consumer_group_snapshot: true when capturing the source. Use restore.auto_consumer_groups: true and a suitable consumer_group_strategy, such as header-based, to translate the saved group offsets after replay.
  6. Verify counts and checksums with an independent consumer. Check committed-only transactional records, original schema IDs or the declared rewrite mapping, deserialisation using only the destination registry, and the next record each recovered consumer group reads. Confirm the registry was not left in IMPORT.
  7. Stop the restore identity and retain the measured recovery evidence.
mode: restore
backup_id: production-daily
target:
  bootstrap_servers: ["pkc-target.westeurope.azure.confluent.cloud:9092"]
  topics:
    include: ["orders-*"]
  security:
    security_protocol: SASL_SSL
    sasl_mechanism: OAUTHBEARER
restore:
  target_profile: confluent_cloud
  create_topics: true
  produce_acks: -1
  auto_consumer_groups: true
  consumer_group_strategy: header-based
enterprise:
  oauth:
    providers:
      entra-restore:
        type: oidc_client_credentials
        token_endpoint_url: https://login.microsoftonline.com/${ENTRA_TENANT_ID}/oauth2/v2.0/token
        client_id: ${ENTRA_RESTORE_CLIENT_ID}
        client_secret: ${ENTRA_RESTORE_CLIENT_SECRET}
        scope: ${ENTRA_API_SCOPE}
    kafka:
      target:
        provider: entra-restore
        extensions:
          logicalCluster: ${TARGET_KAFKA_CLUSTER_ID}
          identityPoolId: ${RESTORE_POOL_ID}
  schema_registry:
    url: ${TARGET_SCHEMA_REGISTRY_URL}
    auth:
      type: oauth
      provider: entra-restore
      target_sr_cluster: ${TARGET_SCHEMA_REGISTRY_CLUSTER_ID}
      identity_pool_id: ${RESTORE_POOL_ID}
    restore:
      strategy: into_existing
      verify: true

Add the storage section from section 10 to make this configuration complete. Do not combine the Confluent-specific restore options above with common enterprise.registry_restore serializer options unless that path has been qualified for the target. The latter retains the separately qualified self-hosted Confluent and Apicurio workflow from 0.6.1.

13. Qualification results​

No measured Confluent Cloud throughput, transfer cost or recovery time is available yet. Local results are not Cloud performance figures.

VerificationResult
Workspace and HTTP-provider tests882 tests passed after integration with 0.6.1 and the safe-read retry fix
Local Confluent Platform backup and restoreS01 passed, including records and group translation
Local Schema Registry restoreS03 passed, 36 assertions including IDs, deleted versions and mode recovery
Local 30-minute OAuth soakPassed: 1821 seconds, 190 re-authentications, no reconnections after the first minute
Real Entra Basic, POST and OIDC discovery token exchangePassed with the production provider; temporary applications removed and verified absent
Confluent identity pool authorisation and automatic mappingPassed on the candidate with Basic, POST and automatic pool mapping
Public Enterprise 0.7.0 with Azure managed identity and AKS workload identityBackup, source restore and fresh-cluster restore passed; 44 checks per restore. Projected-token rotation untested
OAuth over SASL_SSL, Schema Registry and restore permissions on CloudPublic VM and AKS image/chart passed
DEK OAuth headersEarlier candidate accepted; empty DEK list does not qualify encrypted-record recovery
Public operator and chart 1.4.5Source and fresh-target data restore passed 44 checks each, including committed transactions. API-key Kafka auth and Workload Identity storage; registry recovered separately by Enterprise
Restore kill/resumePassed on Cloud using durable Azure progress; no missing records or duplicates
24-hour Cloud soak, incremental compaction, point-in-time restore and hard-killed backupPending
Cross-tenant storage isolation, real SAS deletes, versioning and soft deletePending
Measured Cloud throughput, transfer, storage operations and costPending

Publish performance figures only after the qualification report records the released binary version and checksum, environment and cluster types, scenario results, measurement conditions, cost and confirmed teardown.