Cloud Tasks SQL Connector#

Start with the Quickstart for the basic DataStream job, or use the Cloud Tasks examples for complete Table sink requests and cross-connector pipelines.

Overview and setup#

The cloud-tasks table connector is a sink provided by the flink-connector-gcp-cloudtasks module. It maps onto the DataStream sink, which documents checkpoint behavior, retries, task naming and queue pacing. This page defines how SQL rows become external HTTP or App Engine requests.

Cloud Tasks is a request dispatch queue rather than an API-specific client. The target API therefore decides whether a request uses JSON, another body format, a query string, or no body at all. SQL represents that split with a Flink format for the body and writable metadata for the rest of the request.

CREATE TABLE order_tasks (
  order_id   STRING,
  amount     DECIMAL(12, 2),
  trace      MAP<STRING, STRING> METADATA FROM 'headers',
  schedule_at TIMESTAMP_LTZ(6)   METADATA FROM 'schedule-time',
  dedupe_key STRING              METADATA FROM 'task-id'
) WITH (
  'connector' = 'cloud-tasks',
  'project'   = 'my-project',
  'location'  = 'asia-northeast1',
  'queue'     = 'orders',
  'http.url'  = 'https://orders-abc-an.a.run.app/tasks',
  'http.method' = 'POST',
  'http.headers.Content-Type' = 'application/json',
  'http.oidc.service-account-email' =
    'dispatcher@my-project.iam.gserviceaccount.com',
  'http.oidc.audience' = 'https://orders-abc-an.a.run.app',
  'format' = 'json'
);

INSERT INTO order_tasks
SELECT order_id,
       amount,
       MAP['X-Trace-Id', trace_id],
       dispatch_at,
       order_id
FROM staged_orders;

The JSON format sees only order_id and amount. The three metadata columns configure the request outside that body. Because target.type defaults to http, this DDL remains compatible with tables created before App Engine target support.

Getting the connector onto the classpath#

Use flink-sql-connector-gcp-cloudtasks, the relocated SQL uber-jar, for SQL Client deployments. Place flink-sql-connector-gcp-cloudtasks-<version>.jar in Flink’s lib/ before starting the cluster, or load it for one SQL Client session:

ADD JAR '/path/to/flink-sql-connector-gcp-cloudtasks-1.1.0.jar';

The jar bundles flink-connector-gcp-cloudtasks and the runtime dependency tree it needs. Third-party dependency packages and internal helpers move under io.github.flink.gcp.connector.cloudtasks.shaded, including the matching native-resource rename required by the already-shaded gRPC Netty transport. Conscrypt remains unrelocated because it owns native libraries and is optional; the other unrelocated third-party packages are annotations only. The generated META-INF/NOTICE enumerates every bundled artifact, with pinned permissive licence texts under META-INF/licenses/.

The shared lineage values PhysicalResourceFacet and ResourceIdentifier also retain their original package names so one listener can consume them across SQL connector jars. See Lineage for the class loader configuration and listener contract.

Keep sibling SQL connector jars as separate files in lib/ or add each with its own ADD JAR. Merging them into another fat jar without merging service descriptors can silently discard one of the factory registrations. A Maven or Gradle DataStream job should instead depend on the plain flink-connector-gcp-cloudtasks module and resolve its transitive dependencies normally.

Table sink#

Body format and request metadata#

This overview covers generic body serialization and the request metadata projected before encoding. Later sibling sections keep the form-specific examples, bodyless methods and writable metadata independently visible in the page outline.

format is any SerializationFormatFactory available on the job classpath, such as json, csv, Avro or raw where its schema requirements are met. The connector does not interpret bytes from those generic formats, so set the matching Content-Type header for the target API. The worked request-body examples show their SQL input and exact JSON, CSV, raw and Avro bytes.

The module also provides the form-urlencoded format for application/x-www-form-urlencoded bodies. External HTTP requests carry those bodies under POST, PUT and PATCH, while App Engine requests carry them under POST and PUT. It accepts only these physical SQL column types:

All other physical types are rejected when Flink creates the sink. This includes CHAR, numeric, Boolean, binary and temporal types, nested ROW and MAP types, and every array type other than one-dimensional ARRAY<STRING>. Cast scalar values to STRING so the SQL states their wire representation explicitly. For structured values, flatten the required members into STRING columns or serialize the value to a chosen string representation before inserting it into the sink. The format does not choose a bracket, dotted-name or JSON convention for nested values because application/x-www-form-urlencoded does not define one.

Fields follow physical schema order, repeated values follow array order, and both names and values use UTF-8 form encoding. A null field or array is omitted, an empty string is preserved as name=, an empty array adds no field, and a null array element fails the row because a form cannot represent it. Writable metadata columns are projected out before encoding and never become form fields.

The form format adds Content-Type: application/x-www-form-urlencoded automatically. You do not need to set that header in the selected target’s fixed headers or in metadata. An equivalent value is canonicalized, while a different value or a value with media-type parameters is rejected as a conflict.

Form request examples#

The examples below show the HTTP request definition created from one INSERT row. Cloud Tasks may dispatch that request more than once under the queue retry policy.

Repeated and joined array values#

An ARRAY<STRING> column repeats its column name, while ARRAY_JOIN converts an array to one scalar form value when the receiving API expects a delimiter.

CREATE TABLE form_tasks (
  order_id   STRING,
  note       STRING,
  tags       ARRAY<STRING>,
  categories STRING
) WITH (
  'connector' = 'cloud-tasks',
  'project' = 'my-project',
  'location' = 'asia-northeast1',
  'queue' = 'forms',
  'http.url' = 'https://api.example.com/orders',
  'http.method' = 'POST',
  'format' = 'form-urlencoded'
);

INSERT INTO form_tasks
VALUES (
  '42',
  '東京 + pickup',
  ARRAY['urgent', 'gift'],
  ARRAY_JOIN(ARRAY['books', 'sale'], ',')
);

The inserted row produces this request body.

POST /orders HTTP/1.1
Content-Type: application/x-www-form-urlencoded

order_id=42&note=%E6%9D%B1%E4%BA%AC+%2B+pickup&tags=urgent&tags=gift&categories=books%2Csale

The comma is part of the categories value and is therefore percent-encoded as %2C. The receiving form parser recovers the value books,sale.

Null and empty values#

The same table can distinguish an empty string from an omitted value.

INSERT INTO form_tasks
VALUES (
  '43',
  '',
  CAST(NULL AS ARRAY<STRING>),
  CAST(NULL AS STRING)
);

The empty note remains present, while the null tags array and null categories value add no fields. An empty array supplied by an upstream table also adds no fields.

POST /orders HTTP/1.1
Content-Type: application/x-www-form-urlencoded

order_id=43&note=

Bracket and dotted-name conventions#

Some servers interpret brackets or dots in field names as a nested structure. The connector does not assign those meanings, but quoted SQL column names can produce the names a specific server expects.

CREATE TABLE nested_form_tasks (
  `items[]`              ARRAY<STRING>,
  `customer.name`        STRING,
  `customer[postalCode]` STRING,
  `attributes[priority]` STRING
) WITH (
  'connector' = 'cloud-tasks',
  'project' = 'my-project',
  'location' = 'asia-northeast1',
  'queue' = 'forms',
  'http.url' = 'https://api.example.com/orders',
  'http.method' = 'POST',
  'format' = 'form-urlencoded'
);

INSERT INTO nested_form_tasks
SELECT items,
       customer.name,
       customer.postal_code,
       attributes['priority']
FROM incoming_orders;

For an input containing items ['book', 'pen'], customer ('Alice', '100-0001') and priority high, the request body is:

items%5B%5D=book&items%5B%5D=pen&customer.name=Alice&customer%5BpostalCode%5D=100-0001&attributes%5Bpriority%5D=high

This projection works when the nested members and map keys are known in the sink schema. It cannot turn arbitrary map keys or an arbitrary number of array indexes into field names because physical column names are fixed when Flink plans the job.

JSON in one form field#

JSON_OBJECT can convert selected structured values to one STRING column before the form format encodes it.

CREATE TABLE json_parameter_tasks (
  payload STRING
) WITH (
  'connector' = 'cloud-tasks',
  'project' = 'my-project',
  'location' = 'asia-northeast1',
  'queue' = 'forms',
  'http.url' = 'https://api.example.com/orders',
  'http.method' = 'POST',
  'format' = 'form-urlencoded'
);

INSERT INTO json_parameter_tasks
SELECT JSON_OBJECT(
         KEY 'name' VALUE customer.name,
         KEY 'postalCode' VALUE customer.postal_code,
         KEY 'items' VALUE items
       )
FROM incoming_orders;

For the preceding customer and items, the JSON value can produce the following URL-encoded payload field.

payload=%7B%22items%22%3A%5B%22book%22%2C%22pen%22%5D%2C%22name%22%3A%22Alice%22%2C%22postalCode%22%3A%22100-0001%22%7D

JSON object member order is not part of the form format contract and may differ across Flink versions. The receiving API must interpret the decoded value as JSON rather than depend on its member order.

Fully custom form bodies#

Dynamic field names such as items[0], items[1] and every key from an arbitrary map require a serializer that owns the complete form body. One SQL approach is an application-provided scalar function combined with Flink’s raw format.

CREATE TABLE custom_form_tasks (
  body STRING
) WITH (
  'connector' = 'cloud-tasks',
  'project' = 'my-project',
  'location' = 'asia-northeast1',
  'queue' = 'forms',
  'http.url' = 'https://api.example.com/orders',
  'http.method' = 'POST',
  'http.headers.Content-Type' = 'application/x-www-form-urlencoded',
  'format' = 'raw'
);

-- TO_API_FORM is a scalar function supplied and registered by the job.
INSERT INTO custom_form_tasks
SELECT TO_API_FORM(items, attributes)
FROM incoming_orders;

The function must return a complete UTF-8 form body with every name and value form-encoded. For example, it could return:

items%5B0%5D=book&items%5B1%5D=pen&attributes=priority%3Ahigh%2Ccolor%3Ablue

This example uses indexed array names and joins the map as priority:high,color:blue inside one form value. The function owns the map entry order, delimiters, escaping and null behavior because those rules come from the receiving API rather than from the media type.

Do not send a pre-encoded body through form-urlencoded. That format treats &, = and percent signs as data inside one field and encodes them again. The DataStream API is another option when the custom transformation is easier to express as a CloudTasksSerializationSchema than as a SQL function.

Methods without a body#

External HTTP requests carry the encoded body only under POST, PUT and PATCH. App Engine requests carry it only under POST and PUT. Every other method allowed by the selected protobuf request type leaves the body empty and does not invoke the format for that row.

For a GET API, construct the query string as part of http.url or the url metadata column. For example:

CREATE TABLE search_tasks (
  unused_body STRING,
  target_url STRING NOT NULL METADATA FROM 'url'
) WITH (
  'connector' = 'cloud-tasks',
  'project' = 'my-project',
  'location' = 'asia-northeast1',
  'queue' = 'search',
  'http.method' = 'GET',
  'format' = 'raw'
);

INSERT INTO search_tasks
SELECT '',
       'https://api.example.com/search?q=' || URL_ENCODE(query_text)
FROM pending_searches;

URL_ENCODE is a built-in Flink SQL function on the supported Flink 2.x line. Flink 1.20 jobs must register an equivalent UTF-8 URL-encoding scalar UDF or construct the complete URL before the row reaches SQL. Concatenating an unescaped value changes the query syntax and is not made safe by Cloud Tasks.

Multipart bodies are not part of this connector contract. A job that needs multipart can implement CloudTasksSerializationSchema through the DataStream API, where the boundary and part encodings can be controlled explicitly.

Writable metadata#

Metadata is appended after the physical columns before the runtime serializer receives a row. The connector projects the physical prefix before invoking the format, so request metadata never appears in JSON, CSV or another body by accident. The sink advertises only metadata belonging to the selected target family, so url cannot be used with App Engine and the App Engine address and routing keys cannot be used with HTTP.

App Engine targets#

Set target.type to app-engine for an App Engine handler in the queue’s project and region. The connector creates the task’s AppEngineHttpRequest arm rather than translating routing into an external URL. Task-level service, version and instance selectors are independent, and a queue-level appEngineRoutingOverride remains authoritative when the queue defines one.

The worked App Engine example shows the target and routing metadata together in one planned statement.

An empty or absent routing value lets App Engine choose its default service, version and an available instance. Instance routing is valid only for a manually scaled service. Reserved headers including Host, Content-Length, X-Google-* and X-AppEngine-* are set by Cloud Tasks or App Engine and cannot be overridden.

Authentication has two independent identities#

service-account-key-file authenticates the eager writer or staged committer to the Cloud Tasks API. When it is absent that component uses application-default credentials. The file path, not the credential contents, travels in the job graph, and each TaskManager reads the file when its eager writer or staged committer starts.

The http.oidc.* and http.oauth.* options configure a token that Cloud Tasks attaches later when it dispatches the HTTP request. They do not authenticate the Flink process and cannot replace permission to call CreateTask.

Use OIDC for Cloud Run, Cloud Run functions, and another endpoint that validates a Google-issued OIDC token. The OIDC service account must belong to the same project as the queue. The principal that creates tasks needs permission to enqueue tasks and iam.serviceAccounts.actAs on that service account; the service account in turn needs Cloud Run Invoker on the target service or function. The connector does not create or modify those IAM bindings.

When the task URL contains a path, set http.oidc.audience to the stable root URL of the Cloud Run service or function, normally its default run.app URL. If the option is absent, Cloud Tasks uses the complete target URL, including its path, as the audience. The eager writer or staged committer uses the identity selected independently through service-account-key-file or application-default credentials to call CreateTask.

A public Compute Engine, GKE or on-premises handler can also use OIDC, but the application must validate the signature, issuer, audience and intended service-account identity itself. Cloud Tasks headers such as X-CloudTasks-TaskName are request metadata, not proof of identity. The endpoint must be reachable by Cloud Tasks; an internal-only address is not made reachable by setting a token.

Use OAuth for Google API endpoints on *.googleapis.com that require an access token and scope. OIDC and OAuth are mutually exclusive because the Cloud Tasks request stores them in one protobuf oneof.

App Engine targets do not accept the http.oidc.* or http.oauth.* options. Cloud Tasks dispatches them through the same-project, same-region App Engine integration instead of attaching an external HTTP authorization token.

Lineage#

A Cloud Tasks table reports its configured queue in the gcp physical-resource facet. Flink owns the logical SQL dataset name, such as default_catalog.default_database.tasks. The dataset namespace is cloudtasks://{project}/{location}; the facet separately retains the physical queue name, kind cloudtasks-queue, and configured project, location and queue. The queue naming is a project convention. The internal Table adapter preserves this distinction without changing the insert-only contract or writable metadata.

HTTP and App Engine tables report the same queue identity, whether task IDs or target addresses come from options or row metadata. Extraction performs no authentication, client creation, RPC, row serialization or task-ID extraction. Individual tasks, target URLs or routes, authentication subjects and payloads do not add datasets, and queue lineage does not prove dispatch, handler execution or downstream business effects. Flink 2.x extracts this metadata during planning; Flink 1.20 supports direct connector metadata inspection without native listener delivery. See Lineage for the custom facet and listener requirements.

Options#

The queue is fixed for a table. SQL exposes no dynamic queue metadata and never creates or configures a queue. Queue rate limits, dispatch concurrency, delivery retries, and paused or disabled queue behavior are service-side settings and behaviors that apply equally to Table jobs; they are documented under Queues, rate limits and sink concurrency rather than repeated as Table options. Writer concurrency and CreateTask RPC recovery use the options below and follow the same runtime behavior as the DataStream sink.

An option left out leaves the corresponding DataStream setting at its existing default, except target.type, http.method and app-engine.method, whose SQL defaults are http, POST and POST respectively.

Queue, body and writer identity#

service-account-key-file and emulator-endpoint are mutually exclusive. Service-account keys are long-lived secrets, so prefer an attached service account or Workload Identity where the deployment supports one.

HTTP request defaults#

Every explicitly configured http.* option is rejected when target.type is app-engine.

App Engine request defaults#

Every explicitly configured app-engine.* option is rejected when target.type is http.

Writer tuning#

These map one-for-one onto CloudTasksWriterOptions.Builder. The DataStream tuning section explains why NOT_FOUND has a separate short budget and why no setting controls dispatch rate.

Checkpointed creation settings#

These keys map onto the same delivery mode and CloudTasksStagedOptions used by DataStream. Any explicitly supplied sink.staged.* key requires sink.delivery-guarantee = 'exactly-once', even when its value equals the default. The factory reports invalid values using their SQL keys; retention minus clock skew and request timeout must leave at least one whole millisecond.

The staging reference explains the accounting and recovery assumptions behind these values. The clock-skew allowance is a deployment convention, not a service guarantee. The existing in-flight, retry, transport and per-destination settings configure the committer in this mode.

Delivery guarantees and task identity#

See Write and key-collision semantics for the Table and DataStream API comparison.

The connector accepts an insert-only Flink changelog. That planner contract prevents update and delete rows from becoming new HTTP requests, but it does not deduplicate task creation by itself. In the default at-least-once mode, replay creates another unnamed task when writable task-id metadata is absent.

Selecting task-id installs the DataStream sink’s existing task-id extractor for every row. A remembered duplicate returns ALREADY_EXISTS, which the sink treats as successful creation without comparing the existing task’s payload or schedule. Cloud Tasks cannot update a task after creation, so reusing an ID does not replace the originally created task definition, even when only the executed or deleted task’s retained name remains. The metadata value must identify an immutable logical task, or include a content or schedule version when a changed row must create another task.

In the eager mode, stable keys provide bounded effectively-once task creation. Cloud Tasks may dispatch the handler more than once, so the handler still needs an idempotent operation or its own durable event ledger.

Checkpointed task creation#

Select sink.delivery-guarantee = 'exactly-once' to stage named tasks and create them after their owning checkpoint completes. This uses the DataStream writer and committer, with the same guarantee of exactly-once task creation per staged envelope within the documented scope and recovery window. The task-id metadata remains optional: selecting it hashes the application key, while omitting it generates a persisted random identity for each accepted record. Two independently accepted copies of a logical event remain distinct without a stable key. Restore reuses the original queue, name and serialized task, including its schedule-time; it never invokes the format or metadata extractor again for a restored envelope. If that schedule is already in the past after checkpoint or commit latency, the task is immediately eligible for dispatch under the queue’s pacing and state.

The following example requires a pre-provisioned queue and a durable filesystem mounted at the same path on every JobManager and TaskManager. Replace the example destination, handler URL and checkpoint directory for the deployment; a private local filesystem on one process is insufficient for distributed recovery. The queue must retain names for the configured assumption, and the creating identity needs both task-creation permission and cloudtasks.queues.get for the retention preflight. The example’s datagen source is synthetic input, not an event deduplication protocol.

SET 'execution.runtime-mode' = 'STREAMING';
SET 'execution.checkpointing.interval' = '1 s';
SET 'execution.checkpointing.mode' = 'EXACTLY_ONCE';
SET 'execution.checkpointing.checkpoints-after-tasks-finish' = 'true';
SET 'execution.checkpointing.storage' = 'filesystem';
SET 'execution.checkpointing.dir' = 'file:///shared/flink/checkpoints/cloudtasks';
SET 'execution.checkpointing.externalized-checkpoint-retention' = 'RETAIN_ON_CANCELLATION';
SET 'execution.checkpointing.timeout' = '5 min';
SET 'restart-strategy.type' = 'fixed-delay';
SET 'restart-strategy.fixed-delay.attempts' = '3';
SET 'restart-strategy.fixed-delay.delay' = '1 s';

CREATE TABLE generated_tasks (
  payload STRING
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '10'
);

CREATE TABLE checkpointed_tasks (
  payload STRING
) WITH (
  'connector' = 'cloud-tasks',
  'project' = 'my-project',
  'location' = 'asia-northeast1',
  'queue' = 'webhooks',
  'http.url' = 'https://api.example.com/tasks',
  'format' = 'json',
  'sink.delivery-guarantee' = 'exactly-once',
  'sink.staged.max-tasks' = '1000'
);

INSERT INTO checkpointed_tasks SELECT payload FROM generated_tasks;

The Table runtime refuses Context.isBounded() == true, which the supported Flink batch planners report. Streaming planners report false even for finite input, so finite STREAMING jobs are supported when exactly-once checkpoints and checkpoints after tasks finish are enabled; their tail commits after a completed checkpoint. The shared graph check rejects BATCH, AUTOMATIC, disabled checkpointing, at-least-once checkpoint alignment and disabled checkpoints after tasks finish. In a Table job, execution.runtime-mode=AUTOMATIC does not get that far on the supported Flink versions: their Table API accepts only an explicit BATCH or STREAMING mode and refuses AUTOMATIC before the sink is planned. These are planning checks, not remote queue checks. The committer performs the retention readback at startup; an emulator endpoint skips it, and disabling verification makes retention the operator’s responsibility. No path creates a queue or updates queue policy.

Recovery requires the latest retained checkpoint, a finite restart policy, and the same mapped operator state and queue. Preserve the SQL job’s topology and operator identities when restoring; keeping the same table name alone does not establish a state mapping. Flink’s UID generation documentation explains persisted compiled plans and why setting table.exec.uid.generation to ALWAYS alone does not establish stable identities across translations. Keep the original deployment artifact and checkpoint path with the recovery record.

Follow the shared recovery runbook for poison state, expiry and stop-with-savepoint. The SQL sink.staged.expired-envelope-policy values select the corresponding DataStream policies: assume-committed is justified after a stop-with-savepoint demonstrably reached FINISHED; create-anyway accepts duplicates; drop accepts loss or relies on handler records proving completion. These overrides apply only to expired envelopes; corrupt state, unknown deadlines, queue mismatches and retention failures still fail. Return to fail after recovery. If a savepoint was created but the stop fails to reach FINISHED, preserve and recover that savepoint: an external restart from an older checkpoint can create fresh random names for tasks already created by the savepoint commit.

The heap and checkpoint sizing rules and operational metrics apply to both APIs. Writer caps do not bound Flink’s total pending collector; use peak pending counts, task size and representation overhead, and size the checkpoint timeout for all pending commit waves and retries. Task visibility is incremental after checkpoint completion, and handler execution remains at-least-once. Administrative removal of name protection and unbounded late service effects remain outside the guarantee. The mode is experimental. It has #1245’s adopted real-service recovery evidence, and #1246 measured the DataStream sink it runs on keeping up with the at-least-once path at every rate it tried, without a separate Table measurement, with most of a checkpoint interval added to task visibility latency; the DataStream page has the figures.

Testing#

Planner tests translate every source-backed statement through connector discovery and sink validation without submitting a job or calling GCP. Table MiniCluster tests additionally execute the production factory, transport and committer against a loopback fake service, covering checkpoint-triggered creation, success with a lost response, retained-checkpoint recovery, expiry and explicit overrides, finite streaming input and stable-key collisions. The fake records accepted task names separately from received requests; these tests are connector recovery evidence, not real-service acceptance. Serializer and factory tests cover physical-column projection, target-family metadata, header precedence, body methods, OIDC and OAuth selection, and task-ID extraction. The emulator integration tests add HTTP dispatch, metadata overrides, named-task deduplication, form bytes, and an inspectable App Engine task; the App Engine real-service suite establishes routing and dispatch behavior against the service.