Confluent Cloud Schema "<confluent_env>/<schema_registry_cluster_id>/<schema_name>/latest": error loading the latest Schema: error loading the latest Schema: 404

[tusharshahrs]

What happened?

When running pulumi up with a Confluent Cloud schema creation or update the GET url used to check the data for the latest schema is incorrect

Running pulumi up reults in the following:

GET https://<schema_registry_endpoint>/subjects/<schema_registry_cluster_id>%2F<schema_name>/versions/latest
    error: update failed

Schema "<confluent_env>/<schema_registry_cluster_id>/<schema_name>/latest": error loading the latest Schema: error loading the latest Schema: 404 Not Found: Subject '<schema_registry_cluster_id>/<schema_name>' not found.

the URL is supposed to be

https://<schema_registry_endpoint>/subjects/<schema_name>/versions/latest

Example

PENDING

Output of pulumi about

CLI
Version 3.116.1
Go Version go1.22.3
Go Compiler gc

Plugins
KIND NAME VERSION
resource aws 6.25.0
resource aws-native 0.98.0
resource awsx 2.5.0
resource confluentcloud 1.46.0
resource docker 4.5.1
language python unknown

Host
OS darwin
Version 14.5
Arch arm64

This project is written in python: executable='/usr/local/var/pyenv/shims/python3' version='3.11.1'

Found no pending operations associated with shared_resources_dev

Backend
Name pulumi.com
URL https://app.pulumi.com/thepulumiuser
User thepulumiuser
Token type personal

Dependencies:
NAME VERSION
kafka-python 2.0.2
pip 24.0
pulumi_confluentcloud 1.46.0
wheel 0.42.0

Additional context

No response

Contributing

Vote on this issue by adding a 👍 reaction.
To contribute a fix for this issue, leave a comment (and link to your pull request, if you've opened one already).

[iwahbe]

Hey @tusharshahrs. Can you include a self-contained Pulumi program that reproduce the error?

This looks like an upstream error. Have you checked upstream for a similar error?

[tusharshahrs]

Here is some code we got:

SANDBOX_ENV_NAME = "sandbox"
SANDBOX_ENV_ID = "env-*"
SANDBOX_SCHEMA_REGISTRY_ID = "lsrc-*"
CONFLUENT_NETWORK_CIDR_BLOCK="#.#.#.#/#"

sandbox_env = confluent.Environment.get(
            resource_name=SANDBOX_ENV_NAME, id=SANDBOX_ENV_ID
        )

sandbox_schema_registry = confluent.SchemaRegistryCluster.get(
            resource_name="sandbox-schema-registry",
            id=f"{SANDBOX_ENV_ID}/{SANDBOX_SCHEMA_REGISTRY_ID}",
        )

confluent_network = confluent.Network(
            "network-aws-transit-gateway",
            display_name="Work Delivery Dev AWS Peering Network",
            cloud="AWS",
            region="us-east-1",
            connection_types=["PEERING"],
            cidr=CONFLUENT_NETWORK_CIDR_BLOCK,
            environment=confluent.NetworkEnvironmentArgs(id=sandbox_env.id),
            opts=pulumi.ResourceOptions(protect=True),
        )

dev_cluster = connfluent.KafkaCluster(
            "dev-cluster",
            display_name="Shared Dev Cluster",
            availability="SINGLE_ZONE",
            cloud="AWS",
            region="us-east-1",
            dedicated=confluent.KafkaClusterDedicatedArgs(cku=1),
            network=confluent.KafkaClusterNetworkArgs(id=confluent_network.id),
            environment=confluent.KafkaClusterEnvironmentArgs(id=sandbox_env.id),
            opts=pulumi.ResourceOptions(protect=True),
        )

cluster_managed_resource = confluent.ApiKeyManagedResourceArgs(
            id=dev_cluster.id,
            api_version=dev_cluster.api_version,
            kind=dev_cluster.kind,
            environment=confluent.ApiKeyManagedResourceEnvironmentArgs(
                id=sandbox_env.id
            ),
        )

cluster_manager = confluent.ServiceAccount(
            "cluster-manager", description="Service account to manage Dev Kafka cluster"
        )

cluster_admin_role_binding = confluent.RoleBinding(
            "cluster-manager-kafka-cluster-admin",
            principal=pulumi.Output.concat("User:", cluster_manager.id),
            role_name="CloudClusterAdmin",
            crn_pattern=dev_cluster.rbac_crn,
            opts=pulumi.ResourceOptions(protect=True),
        )

cluster_manager_api_key_owner = confluent.ApiKeyOwnerArgs(
            id=cluster_manager.id,
            api_version=cluster_manager.api_version,
            kind=cluster_manager.kind,
        )

cluster_manager_api_key = confluent.ApiKey(
            "cluster-manager-kafka-api-key",
            description="Kafka API Key that is owned by 'cluster-manager' "
            "service account",
            owner=cluster_manager_api_key_owner,
            managed_resource=cluster_managed_resource,
            opts=pulumi.ResourceOptions(depends_on=[cluster_admin_role_binding]),
        )

kafka_topic = confluent.KafkaTopic(
            "test",
            kafka_cluster=confluent.KafkaTopicKafkaClusterArgs(id=dev_cluster.id),
            topic_name="test",
            rest_endpoint=dev_cluster.rest_endpoint,
            credentials=confluent.KafkaTopicCredentialsArgs(
                key=cluster_manager_api_key.id,
                secret=cluster_manager_api_key.secret,
            ),
            config={"retention.ms": "604800000"},
        )

pulumi.Output.all(kafka_topic_name=kafka_topic.topic_name).apply(
            lambda args: confluent.Schema(
                f"{args['kafka_topic_name']}-value",
                format="AVRO",
                subject_name=f"{args['kafka_topic_name']}-value",
                rest_endpoint=sandbox_schema_registry.rest_endpoint,
                credentials=confluent.SchemaCredentialsArgs(
                    key=config.require("confluentRegistryKey"),
                    secret=config.require_secret("confluentRegistrySecret"),
                ),
                schema="""{
        "type": "record",
        "name": "User",
        "fields": [
            {"name": "name", "type": "string"},
            {"name": "age", "type": "int"}
        ]
    }""",
                schema_registry_cluster=confluent.SchemaSchemaRegistryClusterArgs(
                    id=sandbox_schema_registry.id,
                ),
            )

[VenelinMartinov]

This looks related to confluentinc/terraform-provider-confluent#296

The upstream issue seems to be that the TF provider does not handle manual changes in the console.

[VenelinMartinov]

I tried to repro with the following modified program but I keep getting 401s. (EDIT: this is #478)

"""A Python Pulumi program"""

import os
import pulumi

import pulumi_confluentcloud as confluent


SANDBOX_ENV_NAME = "sandbox"
SANDBOX_ENV_ID = "env-*"
SANDBOX_SCHEMA_REGISTRY_ID = "lsrc-*"
CONFLUENT_NETWORK_CIDR_BLOCK = "#.#.#.#/#"

sandbox_env = confluent.Environment(SANDBOX_ENV_NAME)

example = confluent.get_schema_registry_region(
    cloud="AWS", region="us-east-2", package="ESSENTIALS"
)
essentials = confluent.SchemaRegistryCluster(
    "essentials",
    package=example.package,
    environment=confluent.SchemaRegistryClusterEnvironmentArgs(
        id=sandbox_env.id,
    ),
    region=confluent.SchemaRegistryClusterRegionArgs(
        id=example.id,
    ),
)

confluent_network = confluent.Network(
    "network-aws-transit-gateway",
    display_name="Work Delivery Dev AWS Peering Network",
    cloud="AWS",
    region="us-east-1",
    connection_types=["PEERING"],
    cidr=CONFLUENT_NETWORK_CIDR_BLOCK,
    environment=confluent.NetworkEnvironmentArgs(id=sandbox_env.id),
    opts=pulumi.ResourceOptions(protect=True),
)

dev_cluster = confluent.KafkaCluster(
    "dev-cluster",
    display_name="Shared Dev Cluster",
    availability="SINGLE_ZONE",
    cloud="AWS",
    region="us-east-1",
    dedicated=confluent.KafkaClusterDedicatedArgs(cku=1),
    network=confluent.KafkaClusterNetworkArgs(id=confluent_network.id),
    environment=confluent.KafkaClusterEnvironmentArgs(id=sandbox_env.id),
    opts=pulumi.ResourceOptions(protect=True),
)

cluster_managed_resource = confluent.ApiKeyManagedResourceArgs(
    id=dev_cluster.id,
    api_version=dev_cluster.api_version,
    kind=dev_cluster.kind,
    environment=confluent.ApiKeyManagedResourceEnvironmentArgs(id=sandbox_env.id),
)

cluster_manager = confluent.ServiceAccount(
    "cluster-manager", description="Service account to manage Dev Kafka cluster"
)

cluster_admin_role_binding = confluent.RoleBinding(
    "cluster-manager-kafka-cluster-admin",
    principal=pulumi.Output.concat("User:", cluster_manager.id),
    role_name="CloudClusterAdmin",
    crn_pattern=dev_cluster.rbac_crn,
    opts=pulumi.ResourceOptions(protect=True),
)

cluster_manager_api_key_owner = confluent.ApiKeyOwnerArgs(
    id=cluster_manager.id,
    api_version=cluster_manager.api_version,
    kind=cluster_manager.kind,
)

cluster_manager_api_key = confluent.ApiKey(
    "cluster-manager-kafka-api-key",
    description="Kafka API Key that is owned by 'cluster-manager' " "service account",
    owner=cluster_manager_api_key_owner,
    managed_resource=cluster_managed_resource,
    opts=pulumi.ResourceOptions(depends_on=[cluster_admin_role_binding]),
)

kafka_topic = confluent.KafkaTopic(
    "test",
    kafka_cluster=confluent.KafkaTopicKafkaClusterArgs(id=dev_cluster.id),
    topic_name="test",
    rest_endpoint=dev_cluster.rest_endpoint,
    credentials=confluent.KafkaTopicCredentialsArgs(
        key=cluster_manager_api_key.id,
        secret=cluster_manager_api_key.secret,
    ),
    config={"retention.ms": "604800000"},
)

pulumi.Output.all(kafka_topic_name=kafka_topic.topic_name).apply(
    lambda args: confluent.Schema(
        f"{args['kafka_topic_name']}-value",
        format="AVRO",
        subject_name=f"{args['kafka_topic_name']}-value",
        rest_endpoint=essentials.rest_endpoint,
        credentials=confluent.SchemaCredentialsArgs(
            key=os.environ["CONFLUENT_CLOUD_API_KEY"],
            secret=os.environ["CONFLUENT_CLOUD_API_SECRET"],
        ),
        schema="""{
        "type": "record",
        "name": "User",
        "fields": [
            {"name": "name", "type": "string"},
            {"name": "age", "type": "int"}
        ]
    }""",
        schema_registry_cluster=confluent.SchemaSchemaRegistryClusterArgs(
            id=essentials.id,
        ),
    )
)

@tusharshahrs are you able to get a self-contained program which reproduces the issue?

Also the upstream issue linked above suggests that the error comes up when the schema was deleted in the confluentcloud console - can you verify if this is the case here?

EDIT: Solved the 401s with the credentials but I have been unable to deploy the program since the API times out every time.

I'd appreciate it if you checked if the upstream issue is related to your problem.

[VenelinMartinov]

Okay, I've reproduced the issue with some help from @mikhailshilkov and with --refresh - assuming the original reporter did that too. LMK if that's not the case as I have not been able to repro without --refresh. This matches confluentinc/terraform-provider-confluent#296

Program:

"""A Python Pulumi program"""

import pulumi

import pulumi_confluentcloud as confluent


SANDBOX_ENV_NAME = "sandbox"

sandbox_env = confluent.Environment(SANDBOX_ENV_NAME)

example = confluent.get_schema_registry_region(
    cloud="AWS", region="us-east-2", package="ESSENTIALS"
)
essentials = confluent.SchemaRegistryCluster(
    "essentials",
    package=example.package,
    environment=confluent.SchemaRegistryClusterEnvironmentArgs(
        id=sandbox_env.id,
    ),
    region=confluent.SchemaRegistryClusterRegionArgs(
        id=example.id,
    ),
)

dev_cluster = confluent.KafkaCluster(
    "dev-cluster",
    display_name="Shared Dev Cluster",
    availability="SINGLE_ZONE",
    cloud="AWS",
    region="us-east-1",
    basic=confluent.KafkaClusterBasicArgs(),
    environment=confluent.KafkaClusterEnvironmentArgs(id=sandbox_env.id),
)

cluster_managed_resource = confluent.ApiKeyManagedResourceArgs(
    id=dev_cluster.id,
    api_version=dev_cluster.api_version,
    kind=dev_cluster.kind,
    environment=confluent.ApiKeyManagedResourceEnvironmentArgs(id=sandbox_env.id),
)

cluster_manager = confluent.ServiceAccount(
    "cluster-manager", description="Service account to manage Dev Kafka cluster"
)

cluster_admin_role_binding = confluent.RoleBinding(
    "cluster-manager-kafka-cluster-admin",
    principal=pulumi.Output.concat("User:", cluster_manager.id),
    role_name="CloudClusterAdmin",
    crn_pattern=dev_cluster.rbac_crn,
)

cluster_manager_api_key_owner = confluent.ApiKeyOwnerArgs(
    id=cluster_manager.id,
    api_version=cluster_manager.api_version,
    kind=cluster_manager.kind,
)

cluster_manager_api_key = confluent.ApiKey(
    "cluster-manager-kafka-api-key",
    description="Kafka API Key that is owned by 'cluster-manager' " "service account",
    owner=cluster_manager_api_key_owner,
    managed_resource=cluster_managed_resource,
    opts=pulumi.ResourceOptions(depends_on=[cluster_admin_role_binding]),
)

kafka_topic = confluent.KafkaTopic(
    "test",
    kafka_cluster=confluent.KafkaTopicKafkaClusterArgs(id=dev_cluster.id),
    topic_name="test",
    rest_endpoint=dev_cluster.rest_endpoint,
    credentials=confluent.KafkaTopicCredentialsArgs(
        key=cluster_manager_api_key.id,
        secret=cluster_manager_api_key.secret,
    ),
    config={"retention.ms": "604800000"},
)

confluent.Schema(
    "test-value",
    format="AVRO",
    subject_name=kafka_topic.topic_name.apply(lambda name: f"{name}-value"),
    rest_endpoint=essentials.rest_endpoint,
    schema="""{
        "type": "record",
        "name": "User",
        "fields": [
            {"name": "name", "type": "string"},
            {"name": "age", "type": "int"}
        ]
    }""",
    schema_registry_cluster=confluent.SchemaSchemaRegistryClusterArgs(
        id=essentials.id,
    ),
)

Steps:

1. pulumi up
2. Manually delete the schema in the console
3. pulumi up --refresh
4. Observe the error.

The workaround in this case is to delete the resource from the state too and run pulumi up after.

[VenelinMartinov]

@tusharshahrs can we confirm if the user ran with --refresh and if indeed the schema was manually deleted in the console?

[mikhailshilkov]

Response from the user:

We did not delete the schema in confluent cloud prior to hitting this error. We temporarily removed this code; I will retest later this week and report on current state.