diff --git a/src/commands/kafkaClusters.test.ts b/src/commands/kafkaClusters.test.ts index 1fab69f57..e2fea8785 100644 --- a/src/commands/kafkaClusters.test.ts +++ b/src/commands/kafkaClusters.test.ts @@ -452,7 +452,7 @@ describe("commands/kafkaClusters.ts", () => { it("should fire the topicChanged event with change='deleted' after successful deletion", async () => { showInputBoxStub.resolves(TEST_CCLOUD_KAFKA_TOPIC.name); - stubbedLoader.getKafkaClustersForEnvironmentId.resolves([TEST_CCLOUD_KAFKA_CLUSTER]); + stubbedLoader.getClusterForTopic.resolves(TEST_CCLOUD_KAFKA_CLUSTER); await deleteTopicCommand(TEST_CCLOUD_KAFKA_TOPIC); diff --git a/src/commands/kafkaClusters.ts b/src/commands/kafkaClusters.ts index 8ce6517d6..3a5d5ffdc 100644 --- a/src/commands/kafkaClusters.ts +++ b/src/commands/kafkaClusters.ts @@ -152,8 +152,7 @@ export async function deleteTopicCommand(topic: KafkaTopic) { // look up the parent Kafka cluster in order to fire the topicChanged event const loader = ResourceLoader.getInstance(topic.connectionId); - const clusters = await loader.getKafkaClustersForEnvironmentId(topic.environmentId); - const cluster = clusters.find((c) => c.id === topic.clusterId); + const cluster = await loader.getClusterForTopic(topic); if (cluster) { topicChanged.fire({ change: "deleted", cluster }); } diff --git a/src/commands/scaffold.test.ts b/src/commands/scaffold.test.ts index 535241bc7..fb63aefb0 100644 --- a/src/commands/scaffold.test.ts +++ b/src/commands/scaffold.test.ts @@ -7,6 +7,7 @@ import { TEST_CCLOUD_SCHEMA_REGISTRY, } from "../../tests/unit/testResources"; import { TEST_CCLOUD_FLINK_COMPUTE_POOL } from "../../tests/unit/testResources/flinkComputePool"; +import * as errors from "../errors"; import type { CCloudResourceLoader } from "../loaders"; import { CCloudEnvironment } from "../models/environment"; import { KafkaTopic } from "../models/topic"; @@ -102,19 +103,23 @@ describe("commands/scaffold.ts", () => { ); }); - it("should show an error notification if environment for topic is not found", async () => { + it("should show an error notification and log to Sentry if no cluster is found for the topic", async () => { const showErrorNotificationWithButtonsStub = sandbox.stub( notifications, "showErrorNotificationWithButtons", ); + const logErrorStub = sandbox.stub(errors, "logError"); - // make a new topic with an environment ID that won't be found - const topicWithMissingEnv = new KafkaTopic({ + // topic whose cluster can't be resolved within its environment + const topicWithMissingCluster = new KafkaTopic({ ...TEST_CCLOUD_KAFKA_TOPIC, clusterId: "missing-cluster-id", }); + stubbedResourceLoader.getClusterForTopic + .withArgs(topicWithMissingCluster) + .resolves(undefined); - await resourceScaffoldProjectCommand(topicWithMissingEnv); + await resourceScaffoldProjectCommand(topicWithMissingCluster); sinon.assert.notCalled(scaffoldProjectRequestStub); sinon.assert.calledOnce(showErrorNotificationWithButtonsStub); @@ -122,12 +127,13 @@ describe("commands/scaffold.ts", () => { showErrorNotificationWithButtonsStub, `Unable to find Kafka cluster for topic "${TEST_CCLOUD_KAFKA_TOPIC.name}".`, ); + sinon.assert.calledOnce(logErrorStub); }); it("should call scaffoldProjectRequest with correct parameters for KafkaTopic", async () => { - stubbedResourceLoader.getKafkaClustersForEnvironmentId - .withArgs(TEST_CCLOUD_KAFKA_TOPIC.environmentId) - .resolves([TEST_CCLOUD_KAFKA_CLUSTER]); + stubbedResourceLoader.getClusterForTopic + .withArgs(TEST_CCLOUD_KAFKA_TOPIC) + .resolves(TEST_CCLOUD_KAFKA_CLUSTER); await resourceScaffoldProjectCommand(TEST_CCLOUD_KAFKA_TOPIC); diff --git a/src/commands/scaffold.ts b/src/commands/scaffold.ts index 2b202307b..61bd66d06 100644 --- a/src/commands/scaffold.ts +++ b/src/commands/scaffold.ts @@ -2,6 +2,7 @@ import type * as vscode from "vscode"; import { registerCommandWithLogging } from "."; +import { logError } from "../errors"; import { CCloudResourceLoader, ResourceLoader } from "../loaders"; import type { CCloudFlinkComputePool } from "../models/flinkComputePool"; import { KafkaCluster } from "../models/kafkaCluster"; @@ -52,9 +53,15 @@ export async function resourceScaffoldProjectCommand( telemetrySource = "cluster"; } else { // KafkaTopic - const clusters = environment.kafkaClusters; - const cluster = clusters.find((c) => c.id === item.clusterId); + const cluster = await loader.getClusterForTopic(item); if (!cluster) { + // a topic should always resolve to a cluster in its environment; log to Sentry so we + // learn if this unexpected condition happens in the wild. + logError( + new Error("Unable to find Kafka cluster for topic during project scaffolding"), + "Kafka cluster not found for topic during project scaffolding", + { extra: { clusterId: item.clusterId, environmentId: item.environmentId } }, + ); void showErrorNotificationWithButtons( `Unable to find Kafka cluster for topic "${item.name}".`, ); diff --git a/src/loaders/resourceLoader.test.ts b/src/loaders/resourceLoader.test.ts index 67a83c250..fe248196b 100644 --- a/src/loaders/resourceLoader.test.ts +++ b/src/loaders/resourceLoader.test.ts @@ -471,6 +471,53 @@ describe("ResourceLoader::getTopicsForCluster()", () => { }); }); +describe("ResourceLoader::getClusterForTopic()", () => { + let loaderInstance: ResourceLoader; + let sandbox: sinon.SinonSandbox; + let getKafkaClustersForEnvironmentIdStub: sinon.SinonStub; + + beforeEach(() => { + sandbox = sinon.createSandbox(); + loaderInstance = LocalResourceLoader.getInstance(); + getKafkaClustersForEnvironmentIdStub = sandbox.stub( + loaderInstance, + "getKafkaClustersForEnvironmentId", + ); + }); + + afterEach(() => { + sandbox.restore(); + }); + + it("Raises error for a topic from a mismatched connection", async () => { + await assert.rejects(loaderInstance.getClusterForTopic(TEST_CCLOUD_KAFKA_TOPIC), (err) => { + return (err as Error).message.startsWith(`Mismatched connectionId ${LOCAL_CONNECTION_ID}`); + }); + + sinon.assert.notCalled(getKafkaClustersForEnvironmentIdStub); + }); + + it("Returns the cluster matching the topic's clusterId", async () => { + getKafkaClustersForEnvironmentIdStub + .withArgs(TEST_LOCAL_KAFKA_TOPIC.environmentId) + .resolves([TEST_LOCAL_KAFKA_CLUSTER]); + + const cluster = await loaderInstance.getClusterForTopic(TEST_LOCAL_KAFKA_TOPIC); + + assert.strictEqual(cluster, TEST_LOCAL_KAFKA_CLUSTER); + }); + + it("Returns undefined when no cluster in the environment matches the topic's clusterId", async () => { + getKafkaClustersForEnvironmentIdStub + .withArgs(TEST_LOCAL_KAFKA_TOPIC.environmentId) + .resolves([]); + + const cluster = await loaderInstance.getClusterForTopic(TEST_LOCAL_KAFKA_TOPIC); + + assert.strictEqual(cluster, undefined); + }); +}); + describe("ResourceLoader::getConsumerGroupsForCluster()", () => { let loaderInstance: LocalResourceLoader; let sandbox: sinon.SinonSandbox; diff --git a/src/loaders/resourceLoader.ts b/src/loaders/resourceLoader.ts index 58123993d..20bc1d72b 100644 --- a/src/loaders/resourceLoader.ts +++ b/src/loaders/resourceLoader.ts @@ -145,6 +145,28 @@ export abstract class ResourceLoader extends DisposableCollection implements IRe forceRefresh?: boolean, ): Promise; + /** + * Get the {@link KafkaCluster} that a given {@link KafkaTopic} belongs to. + * + * Connection-aware (implemented atop {@link getKafkaClustersForEnvironmentId}), so unlike a + * raw cluster-id lookup it resolves correctly for direct-connection topics. + * + * @param topic The topic to resolve. Must belong to the same connection as this loader. + * @returns The topic's {@link KafkaCluster}, or `undefined` if no cluster in the topic's + * environment matches its {@linkcode KafkaTopic.clusterId}. + * @throws Error if the topic is not from the same connection as this loader. + */ + public async getClusterForTopic(topic: KafkaTopic): Promise { + if (topic.connectionId !== this.connectionId) { + throw new Error( + `Mismatched connectionId ${this.connectionId} for topic ${topic.name} (${topic.connectionId})`, + ); + } + + const clusters = await this.getKafkaClustersForEnvironmentId(topic.environmentId); + return clusters.find((cluster) => cluster.id === topic.clusterId); + } + // Schema registry methods. /**