Package io.github.flink.gcp.connector.cloudtasks.sink

Class CloudTasksSink

java.lang.Object
io.github.flink.gcp.connector.cloudtasks.sink.CloudTasksSink
@Public public final class CloudTasksSink extends Object
Entry point for building a Cloud Tasks sink.

By default the sink creates one task per record, at-least-once, and flushes every outstanding creation at each checkpoint barrier. Dispatch pacing is not configured here: Cloud Tasks paces execution on the queue, so the queue's rate limits and retry policy — applied by whoever created it — decide how fast the tasks run. The sink only decides how fast tasks are handed over.

Opt-in CloudTasksDeliveryGuarantee.EXACTLY_ONCE stages immutable named envelopes and creates tasks after checkpoint completion, within a bounded retention and recovery scope. It requires a fixed queue, checkpointed streaming and the built-in fail-job handler; it does not make handler execution exactly once. See the DataStream guide's recovery runbook before use.

That at-least-once statement assumes the default FailureHandler.failJob() policy. Under a dropping policy configured through CloudTasksSinkBuilder.failedTaskHandler(FailureHandler), a completed checkpoint means every record up to the barrier was either durably accepted, skipped by the serializer, or handed to that handler.

The returned sink implements LineageVertexProvider. A fixed queue contributes one dataset with namespace cloudtasks://PROJECT/LOCATION, name QUEUE, and a gcp physical-resource facet containing the project, location and queue. This naming is a project convention. A user-defined destination resolver contributes no dataset and is never evaluated during extraction. Task URLs, payloads, authentication subjects and task IDs are not datasets; queue lineage does not prove dispatch or handler execution. Extraction performs no authentication, client creation, RPC or serializer call. Flink 2.x extracts this metadata automatically; Flink 1.20 supports direct metadata inspection but not native listener delivery.

Example: 


 Sink<OrderEvent> sink =
         CloudTasksSink.<OrderEvent>builder()
                 .queue(QueueDestination.of("my-project", "asia-northeast1", "webhooks"))
                 .serializer(
                         CloudTasksSerializationSchema.httpTarget(
                                         "https://api.example.com/v1/orders")
                                 .withBody(new MyEventJsonSerializationSchema())
                                 .withHeaders(
                                         e -> Map.of("Content-Type", "application/json"))
                                 .withOidcToken(
                                         "dispatcher@my-project.iam.gserviceaccount.com"))
                 .build();