Dispatch a stream as Cloud Tasks#

Assumes the artifacts and credentials from the Quickstart index, and the imports an IDE resolves from the Java API reference.

Create the queue first. The sink will not create one, and that is deliberate: the queue’s rate limits are the entire reason to use the service, and a queue created with defaults would carry Cloud Tasks’ own (500 dispatches/second, 1000 concurrent) rather than the pacing the target endpoint can absorb.

gcloud tasks queues create webhooks --location=asia-northeast1 \
    --max-dispatches-per-second=10 --max-concurrent-dispatches=5

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
// Not optional: the sink is at-least-once only with checkpointing, which is what makes
// Flink wait for every outstanding task creation before the barrier passes.
env.enableCheckpointing(60_000);

env.fromData("{\"order_id\":\"a-1\"}")
        .sinkTo(
                CloudTasksSink.<String>builder()
                        .queue(
                                QueueDestination.of(
                                        "my-project", "asia-northeast1", "webhooks"))
                        .serializer(
                                CloudTasksSerializationSchema.httpTarget(
                                                "https://api.example.com/v1/orders")
                                        .withBody(new SimpleStringSchema())
                                        .withHeaders(
                                                element ->
                                                        Map.of(
                                                                "Content-Type",
                                                                "application/json")))
                        .build());

env.execute("cloudtasks-quickstart");

The endpoint must be reachable from Cloud Tasks, which for an HTTP target generally means a public IP — the exception, and how to authorize against a Cloud Run service with withOidcToken(...), is on the Cloud Tasks connector page.

Tasks are unnamed by default, so a record Flink replays after a failure creates a second task and calls the endpoint twice. taskIdExtractor(...) opts into deduplication, at a latency cost Google documents as significant.

Next#

The Cloud Tasks examples continue with dynamic queue routing on the DataStream sink; App Engine, Cloud Run function, external API, and cross-connector requests on the Table sink; request-body formats; and an emulator run whose dispatches land on a server that can be inspected locally.

Use the DataStream connector for runtime, delivery, failure, metric, and tuning behavior, or the Table connector for DDL, formats, writable metadata, and planner restrictions.