Cloud Tasks examples#

Starting from the Cloud Tasks quickstart job.

The DataStream connector explains sink runtime behavior, while the Table connector owns DDL, writable metadata, and planner restrictions.

For checkpointed task creation, the DataStream configuration and SQL example include retained checkpoints and bounded restart settings. They share the same creation and recovery protocol; the examples below use eager creation unless explicitly configured otherwise.

DataStream sink#

Basic dispatch job#

The Quickstart is the canonical basic job: it creates one external HTTP task for each input record and leaves dispatch pacing to the queue. The examples below change its destination or use the Table API instead of copying that job.

Sharding across queues#

The dynamic destinations guide places this sharding pattern in the shared resolver contract.

CloudTasksSink.<OrderEvent>builder()
        .destinationResolver(
                (element, context) ->
                        QueueDestination.of(
                                "my-project",
                                "asia-northeast1",
                                "webhooks-"
                                        + Math.floorMod(
                                                element.customerId().hashCode(), 4)))
        .serializer(
                CloudTasksSerializationSchema.httpTarget(
                                "https://api.example.com/v1/orders")
                        .withBody(new OrderEventSchema()))
        .build();

A single Cloud Tasks client serves every queue, and the sink creates no per-queue client, stream, publisher, or batcher to cache or evict. When CloudTasksWriterOptions.builder().perDestinationMetrics(true).build() is supplied through writerOptions(...), each queue with a recorded send or failure registers counters that remain for the task lifetime because Flink cannot unregister metrics. That optional metric registry state is separate from service-client state.

Sharding this way is how a pipeline exceeds the per-queue throughput ceiling. The aggregate limits, and why they rarely matter for the workload this connector exists for, are on the Cloud Tasks connector page. All the queues must exist; the sink creates none of them.

Table sink#

The Table sink encodes physical columns as a request body and projects writable metadata into the task and its HTTP or App Engine request. The following cases organize that split around the system Cloud Tasks invokes.

An App Engine handler#

An App Engine target uses the queue project’s native AppEngineHttpRequest arm rather than an external URL. The writable relative-uri, service, version, and instance columns select the handler and its task-level routing for each row.

CREATE TABLE app_engine_tasks (
  payload        STRING,
  target_path    STRING NOT NULL METADATA FROM 'relative-uri',
  service_name   STRING          METADATA FROM 'app-engine-service',
  version_name   STRING          METADATA FROM 'app-engine-version',
  instance_name  STRING          METADATA FROM 'app-engine-instance',
  dedupe_key     STRING          METADATA FROM 'task-id'
) WITH (
  'connector' = 'cloud-tasks',
  'project' = 'my-project',
  'location' = 'asia-northeast1',
  'queue' = 'app-engine-orders',
  'target.type' = 'app-engine',
  'app-engine.method' = 'POST',
  'app-engine.headers.Content-Type' = 'application/json',
  'format' = 'json'
);

INSERT INTO app_engine_tasks
VALUES ('ready', '/tasks/42?source=sql', 'worker', 'v2', CAST(NULL AS STRING), 'order-42');

An absent service or version uses the App Engine default when the task is attempted, and an absent instance selects an available instance. A specific instance requires manual scaling, and a queue-level appEngineRoutingOverride still wins over the values in the row. The Table connector page defines the reserved headers and routing constraints.

An authenticated Cloud Run function#

An HTTP Cloud Run function uses the external HTTP target with an OIDC token. The target URL includes the function path, while the audience remains the function’s stable root run.app URL.

CREATE TABLE function_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' = 'functions',
  'http.url' = 'https://process-order-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://process-order-abc-an.a.run.app',
  'format' = 'json'
);

INSERT INTO function_tasks
VALUES (
  'o-42',
  CAST(19.95 AS DECIMAL(12, 2)),
  MAP['X-Trace-Id', 'trace-42'],
  CAST(CURRENT_TIMESTAMP + INTERVAL '5' MINUTE AS TIMESTAMP_LTZ(6)),
  'order-o-42'
);

The OIDC settings are fixed options rather than writable metadata because the connector exposes no per-row dispatch identity. The identity that calls CreateTask still comes from the Flink writer’s application-default or key-file credentials. That creator identity needs iam.serviceAccounts.actAs on the configured dispatch service account. Cloud Tasks later generates the OIDC token for dispatcher@my-project.iam.gserviceaccount.com when it dispatches the task, and that service account needs permission to invoke the function. Creating the task proves neither that the function ran nor that it succeeded. The authentication reference lists both identities and their permissions.

An external API request#

An external API often varies the resource URL and request details per row. This table keeps the JSON body in physical columns and supplies method, URL, headers, schedule time, and task identity through writable metadata.

CREATE TABLE external_api_tasks (
  order_id       STRING,
  status         STRING,
  amount         DECIMAL(12, 2),
  target_url     STRING NOT NULL    METADATA FROM 'url',
  request_method STRING             METADATA FROM 'http-method',
  request_headers 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' = 'partner-api',
  'format' = 'json'
);

INSERT INTO external_api_tasks
VALUES (
  'o-42',
  'ready',
  CAST(19.95 AS DECIMAL(12, 2)),
  'https://partner.example.com/orders/o-42',
  'PATCH',
  MAP['Content-Type', 'application/json', 'X-Tenant', 'north'],
  CAST(CURRENT_TIMESTAMP + INTERVAL '5' MINUTE AS TIMESTAMP_LTZ(6)),
  'partner-order-o-42-v1'
);

The PATCH body contains only order_id, status, and amount; the five metadata columns are projected out before JSON serialization. The task-id value deduplicates task creation within Cloud Tasks’ retained-name window, but it does not make the external API operation exactly once.

Once the connector successfully creates the task, Cloud Tasks owns dispatch and its retries. A 2xx handler response completes the task; a non-2xx response or a missed deadline is retried under the queue’s retry policy. The connector observes task creation only, so the API operation must be idempotent or deduplicate using a durable business key.

Pub/Sub events enriched from Bigtable#

This pipeline consumes order events from Pub/Sub, looks up each customer’s endpoint and tenant in Bigtable, and creates an external API task from the enriched row.

SET 'execution.checkpointing.interval' = '10 s';

CREATE TABLE incoming_orders (
  event_id    STRING,
  customer_id STRING,
  order_id    STRING,
  amount      DECIMAL(12, 2),
  dispatch_at TIMESTAMP_LTZ(6),
  proc_time AS PROCTIME()
) WITH (
  'connector' = 'pubsub',
  'project' = 'my-project',
  'subscription' = 'orders-sub',
  'format' = 'json'
);

CREATE TABLE customer_routes (
  rowkey  STRING,
  routing ROW<endpoint STRING, tenant STRING>,
  PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'customer-routes',
  'lookup.async' = 'true'
);

CREATE TABLE enriched_order_tasks (
  event_id       STRING,
  order_id       STRING,
  amount         DECIMAL(12, 2),
  customer_id    STRING,
  target_url     STRING NOT NULL    METADATA FROM 'url',
  request_method STRING             METADATA FROM 'http-method',
  request_headers 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' = 'partner-api',
  'format' = 'json'
);

INSERT INTO enriched_order_tasks
SELECT e.event_id,
       e.order_id,
       e.amount,
       e.customer_id,
       r.routing.endpoint || '/orders/' || e.order_id,
       'POST',
       MAP['Content-Type', 'application/json', 'X-Tenant', r.routing.tenant],
       e.dispatch_at,
       e.event_id
FROM incoming_orders AS e
JOIN customer_routes FOR SYSTEM_TIME AS OF e.proc_time AS r
  ON e.customer_id = r.rowkey
WHERE e.event_id IS NOT NULL
  AND e.order_id IS NOT NULL
  AND r.routing.endpoint IS NOT NULL
  AND r.routing.tenant IS NOT NULL;

Checkpointing is required here: Pub/Sub acknowledges consumed messages at completed checkpoints, and the Cloud Tasks sink flushes outstanding task creations at the same boundary. The equality condition covers Bigtable’s complete row key, so the temporal join becomes one point read per input event. An inner join emits no task when the route is absent. The null predicates also drop incomplete route rows and events without the order or task identity needed to construct a valid request. Monitor those rejected inputs in a production pipeline rather than treating their absence as a successful dispatch. The event ID becomes the task identity, while the lookup result selects the URL and tenant header. Treat the routing table as trusted configuration and constrain endpoint values to an allowlist before this insert; the connector forwards each URL but does not enforce which hosts the job may target. Validate order_id as one path segment before concatenating it, rejecting path separators, dot segments, query delimiters and fragment delimiters that could select a different resource on an allowed host.

The Pub/Sub examples cover subscription startup, checkpoint acknowledgement, and per-key ordering. The Bigtable lookup example covers the lookup-key rule and synchronous default. Its Pub/Sub enrichment example compares synchronous and asynchronous lookup with the cache choices. Composing those connectors does not strengthen their delivery or ordering guarantees, and the Cloud Tasks handler remains responsible for idempotent execution.

Table API request bodies#

The cloud-tasks Table sink passes physical columns to the selected Flink serialization format and projects writable request metadata out before encoding. The metadata columns in the four format-specific examples after this overview therefore set X-Trace-Id without appearing in the body. This projection is a connector guarantee.

The connector treats bytes from a generic format as opaque and does not select their media type. Each table therefore sets the Content-Type expected by its HTTP handler. External HTTP POST, PUT and PATCH requests carry the encoded body, while App Engine POST and PUT requests do; other methods do not serialize the row or carry a body. Those method rules are connector guarantees, while the byte representation inside the body belongs to the selected Flink format.

Use the matching format reference for the Flink version deployed with the job:

The following sibling sections keep each format visible in the page outline and name only the options that determine its body.

Nested JSON#

JSON derives its object shape from the physical table schema. This table includes a nested row, an array of rows, a map, a null value and text that requires JSON escaping.

CREATE TABLE json_tasks (
  order_id STRING,
  customer ROW<name STRING, city STRING>,
  items ARRAY<ROW<sku STRING, quantity INT>>,
  attributes MAP<STRING, STRING>,
  note STRING,
  request_headers MAP<STRING, STRING> METADATA FROM 'headers'
) WITH (
  'connector' = 'cloud-tasks',
  'project' = 'my-project',
  'location' = 'asia-northeast1',
  'queue' = 'orders',
  'http.url' = 'https://api.example.com/orders',
  'http.method' = 'POST',
  'http.headers.Content-Type' = 'application/json',
  'format' = 'json',
  'json.encode.ignore-null-fields' = 'false'
);

INSERT INTO json_tasks
VALUES (
  'o-42',
  CAST(ROW('Alice "A"', '東京') AS ROW<name STRING, city STRING>),
  ARRAY[
    CAST(ROW('book', 2) AS ROW<sku STRING, quantity INT>),
    CAST(ROW('pen', 1) AS ROW<sku STRING, quantity INT>)
  ],
  MAP['priority', 'high'],
  CAST(NULL AS STRING),
  MAP['X-Trace-Id', 'trace-42']
);

The HTTP handler receives these UTF-8 bytes:

{"order_id":"o-42","customer":{"name":"Alice \"A\"","city":"東京"},"items":[{"sku":"book","quantity":2},{"sku":"pen","quantity":1}],"attributes":{"priority":"high"},"note":null}

The current supported Flink lines write row members in physical schema order. JSON object order has no semantic meaning, and Flink does not promise that map iteration or object member order remains byte-for-byte stable across future versions. HTTP handlers should parse the object instead of comparing its member order.

CSV quoting and nulls#

CSV derives one output record from the physical row. This table selects a pipe delimiter, the ordinary double-quote character and an explicit null literal.

CREATE TABLE csv_tasks (
  order_id STRING,
  note STRING,
  missing_value STRING,
  request_headers MAP<STRING, STRING> METADATA FROM 'headers'
) WITH (
  'connector' = 'cloud-tasks',
  'project' = 'my-project',
  'location' = 'asia-northeast1',
  'queue' = 'orders',
  'http.url' = 'https://api.example.com/import',
  'http.method' = 'POST',
  'http.headers.Content-Type' = 'text/csv; charset=UTF-8',
  'format' = 'csv',
  'csv.field-delimiter' = '|',
  'csv.quote-character' = '"',
  'csv.null-literal' = 'NULL'
);

INSERT INTO csv_tasks
SELECT '42',
       U&'line 1 | "quoted"\000Aline 2',
       CAST(NULL AS STRING),
       MAP['X-Trace-Id', 'trace-42'];

The body contains one line break inside the quoted second field and no line separator after NULL:

"42"|"line 1 | ""quoted""
line 2"|NULL

Flink owns the delimiter, quoting, escaping and null-literal behavior. Its CSV schema supports scalar fields and one level of ARRAY or ROW whose members are simple types; it rejects MAP and deeper nesting. Flatten structured input or use JSON or Avro when a CSV consumer needs another convention.

A pre-serialized raw body#

The raw format accepts exactly one physical column, so that column can hold a complete body prepared by SQL or an upstream function. Writable metadata does not count toward the one-column boundary because the connector projects it out first.

CREATE TABLE raw_tasks (
  body STRING,
  request_headers MAP<STRING, STRING> METADATA FROM 'headers'
) WITH (
  'connector' = 'cloud-tasks',
  'project' = 'my-project',
  'location' = 'asia-northeast1',
  'queue' = 'orders',
  'http.url' = 'https://api.example.com/text',
  'http.method' = 'POST',
  'http.headers.Content-Type' = 'text/plain; charset=UTF-16BE',
  'format' = 'raw',
  'raw.charset' = 'UTF-16BE'
);

INSERT INTO raw_tasks
VALUES ('東京', MAP['X-Trace-Id', 'trace-42']);

UTF-16BE encodes the body as four bytes:

67 71 4E AC

For a VARBINARY physical column, raw forwards the supplied byte sequence unchanged and raw.charset has no effect. The string charset and numeric endianness options are upstream Flink behavior rather than connector policy.

Binary Avro#

Avro derives its writer schema from the physical table schema and physical field order. Declare the fields NOT NULL when the receiving schema must not contain nullable unions.

CREATE TABLE avro_tasks (
  order_id STRING NOT NULL,
  quantity INT NOT NULL,
  gift BOOLEAN NOT NULL,
  request_headers MAP<STRING, STRING> METADATA FROM 'headers'
) WITH (
  'connector' = 'cloud-tasks',
  'project' = 'my-project',
  'location' = 'asia-northeast1',
  'queue' = 'orders',
  'http.url' = 'https://api.example.com/avro-orders',
  'http.method' = 'POST',
  'http.headers.Content-Type' = 'application/octet-stream',
  'format' = 'avro',
  'avro.encoding' = 'binary'
);

INSERT INTO avro_tasks
VALUES ('o-7', 3, TRUE, MAP['X-Trace-Id', 'trace-7']);

Flink derives this writer schema from the three physical columns:

{
  "type": "record",
  "name": "record",
  "namespace": "org.apache.flink.avro.generated",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "quantity", "type": "int"},
    {"name": "gift", "type": "boolean"}
  ]
}

The binary datum is 06 6F 2D 37 06 01 in hexadecimal, or Bm8tNwYB in base64. Decoded with the derived schema, it contains order_id = "o-7", quantity = 3 and gift = true.

The body is an Avro binary datum without an object-container header, embedded schema or magic bytes. The handler must use the same writer schema, and changing field order, nullability or types changes the wire representation. The Cloud Tasks SQL uber-jar does not bundle generic Flink formats; JSON, CSV and raw are available in the Flink SQL distribution, while Avro requires the version-matched flink-avro format artifact on the SQL Client and cluster classpaths.

Local development#

Running against the emulator#

Google publishes no Cloud Tasks emulator; the one the integration tests use is aertje/cloud-tasks-emulator (MIT). Queues are declared at startup, since neither the emulator nor the sink creates one on demand:

docker run --rm -p 8123:8123 --add-host=host.docker.internal:host-gateway \
    ghcr.io/aertje/cloud-tasks-emulator:2.0.1 \
    -host 0.0.0.0 -port 8123 \
    -initial-queue projects/my-project/locations/asia-northeast1/queues/webhooks

CloudTasksSink.<String>builder()
        .queue(QueueDestination.of("my-project", "asia-northeast1", "webhooks"))
        .serializer(
                // Not localhost: the emulator dispatches from inside the container, where
                // that would be the container itself. --add-host above is what makes this
                // name resolve to the host on Linux; Docker Desktop provides it already.
                CloudTasksSerializationSchema.httpTarget(
                                "http://host.docker.internal:9000/orders")
                        .withBody(new SimpleStringSchema()))
        .emulatorEndpoint("localhost:8123")
        .build();

Unlike the Pub/Sub emulator this one dispatches over real HTTP, so a server on your machine sees exactly what the tasks carry — which is the whole reason it is worth running, and also why the target URL has to be reachable from the container’s network rather than yours. (The module’s own tests solve the same problem with testcontainers’ exposeHostPorts(...).)

Use real GCP to establish the service’s task-name deduplication window, task-size boundaries, scheduling and App Engine routing behavior, per the rule about emulators. The emulator’s HTTP dispatch supports OIDC only, and its unimplemented UpdateQueue leaves queue-level uriOverride routing untested here. Transient failure injection remains in the connector’s tests with fake clients.