Ability to submit Flink statements in snapshot mode.

.claude/rules/testing/unit-tests.md

@@ -51,3 +51,32 @@ assignment instead: `obj["methodName"] = sandbox.stub()`.
 
 - Unit test fixtures in `tests/fixtures/`
 - Shared stubs in `tests/stubs/`
+
+## macOS: Local Electron Test Host Won't Launch (e.g. VS Code 1.131.0)
+
+`npx gulp test` downloads a real VS Code build into `.vscode-test/vscode-darwin-arm64-<version>/`
+(gitignored cache) and spawns it via `@vscode/test-electron@2.3.9` as the Extension Development
+Host. Two independent problems show up together on newer VS Code releases (first seen with 1.131.0):
+
+**1. `spawn .../Electron ENOENT`** — `@vscode/test-electron@2.3.9` hardcodes the launcher binary
+name as `Electron`, but this VS Code build ships it renamed to `Code`. Fix locally (never commit —
+this only touches the gitignored cache):
+
+```bash
+cd ".vscode-test/vscode-darwin-arm64-<version>/Visual Studio Code.app/Contents/MacOS/"
+ln -s Code Electron
+```
+
+**2. `"Visual Studio Code" is damaged and can't be opened`** — adding that symlink modifies the
+signed `.app` bundle's contents, which invalidates its code signature. macOS then refuses to launch
+it and reports it as "damaged" (this is a broken-signature error, not a Gatekeeper quarantine issue
+— no `com.apple.quarantine` xattr needs to be present for it to happen). Fix by re-signing ad hoc
+after adding the symlink:
+
+```bash
+codesign --force --deep --sign - "Visual Studio Code.app"
+```
+
+Re-run `npx gulp test` after both steps. If it still fails with `SIGKILL` or an Electron "bad
+option" error, that's a genuine headless/no-GUI-session environment (e.g. a sandboxed agent session
+without display access) — not a code or signing issue, and not fixable with the above.

.vscode/launch.json

@@ -10,7 +10,9 @@
       "type": "extensionHost",
       "request": "launch",
       "args": [
-        "--extensionDevelopmentPath=${workspaceFolder}/out"
+        "--extensionDevelopmentPath=${workspaceFolder}/out",
+        "--disable-extension",
+        "confluentinc.vscode-confluent"
       ],
       "outFiles": [
         "${workspaceFolder}/out/**/*.js"

CHANGELOG.md

@@ -4,6 +4,12 @@ All notable changes to this extension will be documented in this file.
 
 ## Unreleased
 
+### Added
+
+- Support for submitting Flink statements in
+  "[snapshot](https://docs.confluent.io/cloud/current/flink/concepts/snapshot-queries.html)" mode.
+  New codelens control for toggling streaming vs snapshot submission.
+
 ## 2.3.1
 
 ### Fixed

package.json

@@ -420,6 +420,12 @@
         "title": "Set CCloud Flink Database for Flink Statement",
         "category": "Confluent: Flink SQL"
       },
+      {
+        "command": "confluent.document.flinksql.toggleSnapshotMode",
+        "icon": "$(confluent-logo)",
+        "title": "Toggle Snapshot Mode for Flink Statement",
+        "category": "Confluent: Flink SQL"
+      },
       {
         "command": "confluent.flink.configureFlinkDefaults",
         "icon": "$(settings-gear)",
@@ -1447,6 +1453,10 @@
           "command": "confluent.document.flinksql.setCCloudDatabase",
           "when": "false"
         },
+        {
+          "command": "confluent.document.flinksql.toggleSnapshotMode",
+          "when": "false"
+        },
         {
           "command": "confluent.flink.configureFlinkDefaults",
           "when": "true"

src/codelens/flinkSqlProvider.test.ts

@@ -14,6 +14,7 @@ import { FLINK_CONFIG_COMPUTE_POOL, FLINK_CONFIG_DATABASE } from "../extensionSe
 import type { CCloudResourceLoader } from "../loaders";
 import { CCloudEnvironment } from "../models/environment";
 import { CCloudFlinkComputePool } from "../models/flinkComputePool";
+import { FlinkSnapshotMode } from "../models/flinkStatement";
 import { CCloudKafkaCluster } from "../models/kafkaCluster";
 import type { EnvironmentId } from "../models/resource";
 import * as ccloud from "../sidecar/connections/ccloud";
@@ -27,6 +28,7 @@ import {
   getCatalogDatabaseFromMetadata,
   getComputePoolFromMetadata,
   getDefaultCatalogDatabase,
+  getSnapshotModeFromMetadata,
 } from "./flinkSqlProvider";
 
 const testUri = Uri.parse("file:///test/file.sql");
@@ -202,11 +204,22 @@ describe("codelens/flinkSqlProvider.ts", () => {
 
         const codeLenses: CodeLens[] = await provider.provideCodeLenses(fakeDocument);
 
-        assert.strictEqual(codeLenses.length, 3);
+        assert.strictEqual(codeLenses.length, 4);
 
-        const poolLens = codeLenses[0];
-        const dbLens = codeLenses[1];
-        const resetLens = codeLenses[2];
+        const snapshotModeLens = codeLenses[0];
+        const poolLens = codeLenses[1];
+        const dbLens = codeLenses[2];
+        const resetLens = codeLenses[3];
+
+        assert.strictEqual(
+          snapshotModeLens.command?.command,
+          "confluent.document.flinksql.toggleSnapshotMode",
+        );
+        assert.strictEqual(snapshotModeLens.command?.title, "Mode: Streaming");
+        assert.deepStrictEqual(snapshotModeLens.command?.arguments, [
+          fakeDocument.uri,
+          FlinkSnapshotMode.STREAMING,
+        ]);
 
         assert.strictEqual(
           dbLens.command?.command,
@@ -244,11 +257,18 @@ describe("codelens/flinkSqlProvider.ts", () => {
 
         const codeLenses: CodeLens[] = await provider.provideCodeLenses(fakeDocument);
 
-        assert.strictEqual(codeLenses.length, 3);
+        assert.strictEqual(codeLenses.length, 4);
 
-        const poolLens = codeLenses[0];
-        const dbLens = codeLenses[1];
-        const resetLens = codeLenses[2];
+        const snapshotModeLens = codeLenses[0];
+        const poolLens = codeLenses[1];
+        const dbLens = codeLenses[2];
+        const resetLens = codeLenses[3];
+
+        assert.strictEqual(
+          snapshotModeLens.command?.command,
+          "confluent.document.flinksql.toggleSnapshotMode",
+        );
+        assert.strictEqual(snapshotModeLens.command?.title, "Mode: Streaming");
 
         assert.strictEqual(
           dbLens.command?.command,
@@ -291,17 +311,28 @@ describe("codelens/flinkSqlProvider.ts", () => {
 
       const codeLenses: CodeLens[] = await provider.provideCodeLenses(fakeDocument);
 
-      assert.strictEqual(codeLenses.length, 4);
+      assert.strictEqual(codeLenses.length, 5);
 
       const submitLens = codeLenses[0];
-      const poolLens = codeLenses[1];
-      const dbLens = codeLenses[2];
-      const resetLens = codeLenses[3];
+      const snapshotModeLens = codeLenses[1];
+      const poolLens = codeLenses[2];
+      const dbLens = codeLenses[3];
+      const resetLens = codeLenses[4];
 
       assert.strictEqual(submitLens.command?.command, "confluent.statements.create");
       assert.strictEqual(submitLens.command?.title, "▶️ Submit Statement");
       assert.deepStrictEqual(submitLens.command?.arguments, [fakeDocument.uri, pool, database]);
 
+      assert.strictEqual(
+        snapshotModeLens.command?.command,
+        "confluent.document.flinksql.toggleSnapshotMode",
+      );
+      assert.strictEqual(snapshotModeLens.command?.title, "Mode: Streaming");
+      assert.deepStrictEqual(snapshotModeLens.command?.arguments, [
+        fakeDocument.uri,
+        FlinkSnapshotMode.STREAMING,
+      ]);
+
       assert.strictEqual(dbLens.command?.command, "confluent.document.flinksql.setCCloudDatabase");
       assert.strictEqual(
         dbLens.command?.title,
@@ -323,6 +354,56 @@ describe("codelens/flinkSqlProvider.ts", () => {
       assert.strictEqual(resetLens.command?.title, "Clear Settings");
       assert.deepStrictEqual(resetLens.command?.arguments, [fakeDocument.uri]);
     });
+
+    it("should provide 'Mode: Snapshot' codelens when snapshot mode metadata is set to BATCH", async () => {
+      const pool: CCloudFlinkComputePool = TEST_CCLOUD_FLINK_COMPUTE_POOL;
+      const database: CCloudKafkaCluster = TEST_CCLOUD_KAFKA_CLUSTER;
+      resourceManagerStub.getUriMetadata.resolves({
+        [UriMetadataKeys.FLINK_COMPUTE_POOL_ID]: pool.id,
+        [UriMetadataKeys.FLINK_CATALOG_ID]: TEST_CCLOUD_ENVIRONMENT.id,
+        [UriMetadataKeys.FLINK_CATALOG_NAME]: TEST_CCLOUD_ENVIRONMENT.name,
+        [UriMetadataKeys.FLINK_DATABASE_ID]: database.id,
+        [UriMetadataKeys.FLINK_DATABASE_NAME]: database.name,
+        [UriMetadataKeys.FLINK_SNAPSHOT_MODE]: FlinkSnapshotMode.BATCH,
+      });
+      ccloudLoaderStub.getEnvironments.resolves([testEnvWithPoolAndCluster]);
+
+      const codeLenses: CodeLens[] = await provider.provideCodeLenses(fakeDocument);
+
+      const snapshotModeLens = codeLenses[1];
+      assert.strictEqual(
+        snapshotModeLens.command?.command,
+        "confluent.document.flinksql.toggleSnapshotMode",
+      );
+      assert.strictEqual(snapshotModeLens.command?.title, "Mode: Snapshot");
+      assert.deepStrictEqual(snapshotModeLens.command?.arguments, [
+        fakeDocument.uri,
+        FlinkSnapshotMode.BATCH,
+      ]);
+    });
+  });
+
+  describe("getSnapshotModeFromMetadata()", () => {
+    it("should return STREAMING when metadata is undefined", () => {
+      assert.strictEqual(getSnapshotModeFromMetadata(undefined), FlinkSnapshotMode.STREAMING);
+    });
+
+    it("should return STREAMING when snapshot mode metadata is unset", () => {
+      const metadata: UriMetadata = { [UriMetadataKeys.FLINK_COMPUTE_POOL_ID]: "pool-123" };
+      assert.strictEqual(getSnapshotModeFromMetadata(metadata), FlinkSnapshotMode.STREAMING);
+    });
+
+    it("should return STREAMING when snapshot mode metadata is explicitly cleared (null)", () => {
+      const metadata: UriMetadata = { [UriMetadataKeys.FLINK_SNAPSHOT_MODE]: null };
+      assert.strictEqual(getSnapshotModeFromMetadata(metadata), FlinkSnapshotMode.STREAMING);
+    });
+
+    it("should return BATCH when snapshot mode metadata is set to BATCH", () => {
+      const metadata: UriMetadata = {
+        [UriMetadataKeys.FLINK_SNAPSHOT_MODE]: FlinkSnapshotMode.BATCH,
+      };
+      assert.strictEqual(getSnapshotModeFromMetadata(metadata), FlinkSnapshotMode.BATCH);
+    });
   });
 
   describe("getComputePoolFromMetadata()", () => {

src/codelens/flinkSqlProvider.ts

@@ -6,6 +6,7 @@ import { CCloudResourceLoader } from "../loaders";
 import { Logger } from "../logging";
 import type { CCloudEnvironment } from "../models/environment";
 import type { CCloudFlinkComputePool } from "../models/flinkComputePool";
+import { FlinkSnapshotMode } from "../models/flinkStatement";
 import type { CCloudKafkaCluster } from "../models/kafkaCluster";
 import { hasCCloudAuthSession } from "../sidecar/connections/ccloud";
 import { UriMetadataKeys } from "../storage/constants";
@@ -87,6 +88,20 @@ export class FlinkSqlCodelensProvider extends DisposableCollection implements Co
     const computePool: CCloudFlinkComputePool | undefined =
       await getComputePoolFromMetadata(uriMetadata);
     const { catalog, database } = await getCatalogDatabaseFromMetadata(uriMetadata, computePool);
+    const snapshotMode: FlinkSnapshotMode = getSnapshotModeFromMetadata(uriMetadata);
+
+    // codelens for toggling between streaming (default, continuous) and snapshot (one-shot,
+    // bounded) statement execution mode
+    const isSnapshotMode = snapshotMode === FlinkSnapshotMode.BATCH;
+    const toggleSnapshotModeCommand: Command = {
+      title: isSnapshotMode ? "Mode: Snapshot" : "Mode: Streaming",
+      command: "confluent.document.flinksql.toggleSnapshotMode",
+      tooltip: isSnapshotMode
+        ? "Statement will run once against a snapshot of current data, then complete. Click to switch to streaming mode."
+        : "Statement will run continuously against new data as it arrives. Click to switch to one-time snapshot mode.",
+      arguments: [document.uri, snapshotMode],
+    };
+    const snapshotModeLens = new CodeLens(range, toggleSnapshotModeCommand);
 
     // codelens for selecting a compute pool, which we'll use to derive the rest of the properties
     // needed for various Flink operations (env ID, provider/region, etc)
@@ -129,11 +144,11 @@ export class FlinkSqlCodelensProvider extends DisposableCollection implements Co
         arguments: [document.uri, computePool, database],
       };
       const submitLens = new CodeLens(range, submitCommand);
-      // show the "Submit Statement" | <current pool> | <current catalog+db> codelenses
-      codeLenses.push(submitLens, computePoolLens, databaseLens, resetLens);
+      // show the "Submit Statement" | mode | <current pool> | <current catalog+db> codelenses
+      codeLenses.push(submitLens, snapshotModeLens, computePoolLens, databaseLens, resetLens);
     } else {
       // don't show the submit codelens if we don't have a compute pool and database
-      codeLenses.push(computePoolLens, databaseLens, resetLens);
+      codeLenses.push(snapshotModeLens, computePoolLens, databaseLens, resetLens);
     }
 
     return codeLenses;
@@ -162,6 +177,17 @@ export async function getComputePoolFromMetadata(
   return await loader.getFlinkComputePool(computePoolId);
 }
 
+/**
+ * Get the snapshot ("batch") vs streaming execution mode from the metadata stored in the document.
+ * Defaults to {@link FlinkSnapshotMode.STREAMING} when unset or explicitly cleared.
+ * @param metadata The metadata stored in the document.
+ */
+export function getSnapshotModeFromMetadata(metadata: UriMetadata | undefined): FlinkSnapshotMode {
+  return metadata?.[UriMetadataKeys.FLINK_SNAPSHOT_MODE] === FlinkSnapshotMode.BATCH
+    ? FlinkSnapshotMode.BATCH
+    : FlinkSnapshotMode.STREAMING;
+}
+
 export interface CatalogDatabase {
   catalog?: CCloudEnvironment;
   database?: CCloudKafkaCluster;

src/commands/documents.test.ts

@@ -13,6 +13,7 @@ import {
 } from "../extensionSettings/constants";
 import * as statementUtils from "../flinkSql/statementUtils";
 import type { CCloudResourceLoader } from "../loaders";
+import { FlinkSnapshotMode } from "../models/flinkStatement";
 import * as flinkComputePoolsQuickPick from "../quickpicks/flinkComputePools";
 import * as flinkDatabaseQuickpick from "../quickpicks/kafkaClusters";
 import * as ccloudConnections from "../sidecar/connections/ccloud";
@@ -22,6 +23,7 @@ import {
   resetCCloudMetadataForUriCommand,
   setCCloudComputePoolForUriCommand,
   setCCloudDatabaseForUriCommand,
+  toggleSnapshotModeForUriCommand,
 } from "./documents";
 
 const testUri = Uri.parse("file:///path/to/test.sql");
@@ -428,8 +430,69 @@ describe("commands/documents.ts resetCCloudMetadataForUriCommand()", () => {
       [UriMetadataKeys.FLINK_CATALOG_NAME]: null,
       [UriMetadataKeys.FLINK_DATABASE_ID]: null,
       [UriMetadataKeys.FLINK_DATABASE_NAME]: null,
+      [UriMetadataKeys.FLINK_SNAPSHOT_MODE]: null,
     });
     sinon.assert.calledOnce(uriMetadataSetFireStub);
     sinon.assert.calledOnceWithExactly(uriMetadataSetFireStub, testUri);
   });
 });
+
+describe("commands/documents.ts toggleSnapshotModeForUriCommand()", () => {
+  let sandbox: sinon.SinonSandbox;
+
+  let setFlinkDocumentMetadataStub: sinon.SinonStub;
+  let hasCCloudAuthSessionStub: sinon.SinonStub;
+
+  beforeEach(() => {
+    sandbox = sinon.createSandbox();
+
+    setFlinkDocumentMetadataStub = sandbox
+      .stub(statementUtils, "setFlinkDocumentMetadata")
+      .resolves();
+    hasCCloudAuthSessionStub = sandbox
+      .stub(ccloudConnections, "hasCCloudAuthSession")
+      .returns(true);
+  });
+
+  afterEach(() => {
+    sandbox.restore();
+  });
+
+  it("should do nothing when no Uri is provided", async () => {
+    await toggleSnapshotModeForUriCommand();
+
+    sinon.assert.notCalled(setFlinkDocumentMetadataStub);
+  });
+
+  it("should do nothing when no CCloud auth session is available", async () => {
+    hasCCloudAuthSessionStub.returns(false);
+
+    await toggleSnapshotModeForUriCommand(testUri);
+
+    sinon.assert.notCalled(setFlinkDocumentMetadataStub);
+  });
+
+  it("should set mode to BATCH when current mode is undefined", async () => {
+    await toggleSnapshotModeForUriCommand(testUri, undefined);
+
+    sinon.assert.calledOnceWithExactly(setFlinkDocumentMetadataStub, testUri, {
+      snapshotMode: FlinkSnapshotMode.BATCH,
+    });
+  });
+
+  it("should set mode to BATCH when current mode is STREAMING", async () => {
+    await toggleSnapshotModeForUriCommand(testUri, FlinkSnapshotMode.STREAMING);
+
+    sinon.assert.calledOnceWithExactly(setFlinkDocumentMetadataStub, testUri, {
+      snapshotMode: FlinkSnapshotMode.BATCH,
+    });
+  });
+
+  it("should set mode to STREAMING when current mode is BATCH", async () => {
+    await toggleSnapshotModeForUriCommand(testUri, FlinkSnapshotMode.BATCH);
+
+    sinon.assert.calledOnceWithExactly(setFlinkDocumentMetadataStub, testUri, {
+      snapshotMode: FlinkSnapshotMode.STREAMING,
+    });
+  });
+});