Spanner connector#

An at-least-once sink applies one Spanner Mutation per record. The module also provides a bounded snapshot source and an unbounded Change Streams source. Both dialects, GoogleSQL and PostgreSQL, are supported from the same code; the dialect is a property of the database rather than a builder option.

Every option is listed on the Spanner options page.

Credentials#

The bounded source, Change Streams source, and sink use Application Default Credentials when neither serviceAccountKeyFile(...) nor emulatorEndpoint(...) is set. Set it only when the job must select a service-account JSON key that the runtime environment cannot supply through ADC.

SpannerSink.<OrderEvent>builder()
        .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
        .serializer(orderSerializer)
        .serviceAccountKeyFile("/var/run/secrets/spanner/key.json")
        .build();

The connector serializes only the path into the job graph and reads the file in each process that creates a client. The sink reads it on each TaskManager when a writer starts. The bounded source reads it on the JobManager when a fresh or restored enumerator starts and on each TaskManager when a reader starts. The Change Streams source reads it on the JobManager when its coordinator initializes and on each TaskManager when a reader opens. Table lookup functions read it on their TaskManager when the synchronous or asynchronous lookup function opens, including lookup providers with a partial cache. A deployment must therefore mount the same configured path in every applicable JobManager and TaskManager container. Restart and restore load the mounted file again instead of retaining credential material in checkpointed state.

serviceAccountKeyFile(...) and emulatorEndpoint(...) are mutually exclusive because the emulator channel deliberately uses no credentials. Prefer an attached service account or Workload Identity over a long-lived key where the deployment supports one. The option accepts service-account JSON only, and a loading failure is sanitized so neither the path nor credential contents enter the exception.

Source#

The source reads a Spanner database at one snapshot and finishes.

Source<Singer, ?, ?> source =
        SpannerSource.<Singer>builder()
                .database(DatabaseDestination.of("my-project", "my-instance", "my-db"))
                .readOperation(
                        SpannerReadOperation.query(
                                Statement.of("SELECT id, name FROM singers")))
                .deserializer(new SingerDeserializer())
                .build();

env.fromSource(source, WatermarkStrategy.noWatermarks(), "singers");

Bounded is not the same as batch-only: a bounded source runs inside a streaming pipeline and simply ends, which is what makes reading a Spanner table and joining it against an unbounded stream work.

The options are in the reference.

Fetch hand-off bounds#

Each fetch hands at most maxRowsPerFetch input rows and targets at most maxBytesPerFetch of decoded logical field content before returning control to Flink. The default bounds are 1,000 rows and 12 MiB. Whichever bound is reached first ends the fetch. The row bound preserves the source’s established narrow-row hand-off, while the byte target ends wide-row batches much earlier.

The byte estimate counts UTF-8 bytes for text and JSON, byte payloads for BYTES and PROTO, natural widths for fixed-width values, and recursively counts arrays and structs. Field content of a type newer than the estimator is counted as 64 bytes rather than failing a read. It is deliberately not an estimate of retained JVM heap because object layouts and client internals are not a stable connector contract. Measuring variable fields can force a value that the Spanner client held lazily to decode.

The reader looks ahead by one row to decide whether adding it would take a non-empty batch over the byte target. That row stays with the same open partition for the next fetch. A single row larger than the target is handed over alone so the source always makes progress. If a wake-up cancels the read while a look-ahead row is pending, the row is discarded with the stream and the whole partition is reopened from its start, preserving the recovery behavior below.

These are TaskManager hand-off bounds. They do not configure Spanner transport paging, maxPartitions, or partitionSizeBytes. SDK transport buffers, the one look-ahead row, the element queue’s own overhead and capacity for multiple batches, records produced by the deserializer, and downstream operators remain outside the byte target.

Each active source subtask queues batches independently, so source parallelism can multiply retained input. Tune the per-fetch bounds together with the planned split count and the TaskManager memory budget.

Splits, partitions and recovery#

Spanner decides how the read is divided, not the job. The enumerator opens one batch read-only transaction, asks the service to partition it, and turns each partition into one split. Every subtask then rejoins that same transaction and streams the partitions it is given, so every row the job sees comes from one consistent snapshot — across every subtask, with no column to split on and no bounds to supply.

maxPartitions and partitionSizeBytes are hints, which is Google’s own word for them. The service may return more partitions or fewer, and a job that plans for a particular parallelism has to cope with getting a different one. The enumerator logs the count it received, and warns when it is smaller than the parallelism — the subtasks left without a partition finish immediately.

A partition is the unit of progress, and the unit of re-reading. There is no position inside a partition to resume at: execute replays the whole thing, and a partitioned query’s row order is not contractual — a query is partitionable only when the first operator of its execution plan is a distributed union, which rules out the top-level ORDER BY that would fix an order — so a count of rows already read would not mean the same thing on a second execution. A checkpoint therefore records which partitions a reader still holds, and recovery re-reads each of them from the start.

That makes the delivery guarantee at-least-once, with a duplicate window of one partition. It applies in two places, and the second one is the surprising one:

• a job that fails and restarts re-reads the partitions that were in flight;
• a partition cancelled mid-read while the job is running — which is what Flink does to a reader it needs to interrupt — is opened again at its start, so the rows it had already emitted are emitted again. It is logged at WARN and counted by partitionsReread, which is the metric that explains duplicate output from a run that never failed.

If your pipeline cannot tolerate that, deduplicate on the primary key downstream. The alternative — reading on rather than interrupting — would hold a subtask until the client’s own read deadline whenever the service went quiet, including while a job was being cancelled.

The snapshot, and how long it lasts#

The default is a strong read: the latest committed data at the moment the read is planned. ofReadTimestamp and ofExactStaleness are the other two bounds a batch read can take, and a stale read is cheaper for the service to serve because any replica can answer it.

ofMaxStaleness and ofMinReadTimestamp are rejected by the builder. Spanner allows those two only on a single-use transaction, and a batch read is by construction a multi-use one — every partition rejoins the transaction the plan was made in.

What bounds how long a read may take is the database’s version_retention_period (one hour by default, up to a week). A batch read holds a snapshot at a fixed timestamp, and once that timestamp falls outside the retention window the data behind it is gone: a long backfill either finishes inside the window or runs against a database whose retention was raised first. The same applies to a savepoint — resuming a read whose snapshot has expired is not possible, whatever the state file holds.

Read shapes and push-down#

A read is either a query or a table read, never both:

• SpannerReadOperation.query(Statement) — any root-partitionable query. Predicates and projections are pushed to the service, and a parameterised Statement carries its bindings.
• SpannerReadOperation.read(table, keySet, columns) — a key set and a column list, which is the cheapest shape for reading a table or a key range whole.
• SpannerReadOperation.readUsingIndex(table, index, keySet, columns) — the same through a secondary index, with the key set interpreted in the index’s key space. A column the index does not store is read back from the base table, which costs a lookup per row.

Not every query can be read this way. Spanner partitions a query only when its execution plan begins with a distributed union — in practice a scan of one table, with predicates and projections but no aggregate, no ORDER BY and no LIMIT. A query that is not root-partitionable is refused when the source plans, and the service’s own message says which part it could not distribute; the connector surfaces it rather than wrapping it, because that message is the only thing that tells you what to change. Reading a non-partitionable query on a single subtask is deliberately not offered (#36 holds the deferral).

Deserialization#

SpannerStructDeserializationSchema turns one Struct into zero or more records through a Flink Collector, and declares its own TypeInformation so a job needs no returns(...) after the source. Emitting nothing skips the row: it is emitted nowhere, is not a failure, and recordsSkipped is the only thing that reports it. Every output must be non-null and emitted synchronously during the call; retaining the collector or using it after the method returns is invalid. A row you could not read is a failure and should be thrown, not skipped.

Serverless reads with Data Boost#

dataBoostEnabled(true) runs the read on compute that is not the instance’s, so a large scan does not contend with the workload the instance is serving. Three things come with it:

• the caller needs spanner.databases.useDataBoost on the database, which roles/spanner.databaseReader does not carry — roles/spanner.databaseReaderWithDataBoost is the role that does;
• the read is billed separately;
• its concurrency has a quota of its own, so RESOURCE_EXHAUSTED is a shape a boosted read can meet that an ordinary one does not.

rpcPriority is the cheaper lever, and a different one: LOW keeps the read on the instance’s own compute but tells Spanner to shed it first when that instance is at capacity, so a backfill yields to serving traffic instead of competing with it. It costs nothing extra, where Data Boost is billed separately; it also does not remove the read from the instance, where Data Boost does. The priority applies to the reads that move the rows — the partition the service plans carries it and the read replays it — and not to the one call that plans the partitions.

Data Boost has been exercised end to end against the service, by the gated real-GCP suite (#224, measured 2026-08-10): a boosted read of a 5,000-row table returned every row, both through partitionQuery directly and through a job built with dataBoostEnabled(true). Two things that measurement settles. It needs no edition upgrade — Google lists Data Boost from STANDARD up, and the suite’s instance is a 100-processing-unit STANDARD one — and a read succeeding at all is the spanner.databases.useDataBoost permission being honoured. What it does not reach is Data Boost’s concurrency quota or its billing; a suite this size touches neither, so the RESOURCE_EXHAUSTED a boosted read can meet is still described here rather than demonstrated.

Reading against the emulator#

The emulator is a convenience, never evidence about the service, and on the read path it deviates in one direction that is easy to get backwards. Measured against emulator:1.5.56 on 2026-08-10:

Change Streams source#

The Change Streams source continuously reads data-change records from the generated Spanner read function. The change-stream API is @PublicEvolving rather than frozen: it may change at a minor release, announced in the release notes. It uses the checkpointed partition lineage described in ADR-0099 and the bounded asynchronous reader in ADR-0101.

SpannerChangeStreamSource<OrderChange> source =
        SpannerChangeStreamSource.<OrderChange>builder()
                .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
                .changeStreamName("order_changes")
                .deserializer(new OrderChangeDeserializer())
                .serviceAccountKeyFile("/var/run/secrets/spanner/key.json")
                .startPosition(StartPosition.latest())
                .maxConcurrentQueriesPerSubtask(8)
                .build();

env.fromSource(source, WatermarkStrategy.noWatermarks(), "spanner-order-changes")
        .setParallelism(4);

The complete builder surface is in the reference. In this example the configured query capacity is 32: four subtasks multiplied by eight concurrent queries per subtask. That product is a connector bound, not a published Spanner quota. TaskManagers and slots must provide resources for that parallelism, but they do not distribute partition queries by themselves. Changing source parallelism redistributes checkpointed split ownership only when the job restarts from a checkpoint or savepoint with the new parallelism.

Records and schema changes#

After output filtering, the deserializer receives each data-change record that was not table-filtered or skipped by skipMessagesWithoutChange as one DataChangeRecord and emits zero or more output records through a Flink Collector. It includes the commit timestamp, transaction and partition counts, record sequence, transaction tag, table, modification type, value-capture type, watched column descriptors, and row modifications. Every output collected from the same data-change record carries that record’s commit timestamp as its Flink event timestamp. The collector is valid only during that deserializer call; emit synchronously and do not retain it.

Each Mod exposes keys, new values, and old values as normalized JSON with object members sorted recursively. An empty optional means that Spanner omitted that member, while a present "null" means that the member was present as JSON null. This distinction matters because value-capture modes deliberately omit different halves of a change.

Each column keeps the complete recursive type descriptor as normalized JSON with object members sorted recursively rather than reducing it to the client library’s current type enum. Nested array descriptors, annotations, TOKENLIST, and service codes added before a client-library release therefore survive deserialization and Java serialization. The record’s own valueCaptureType and columnTypes describe the configuration and schema in effect when that change was captured, so a running job can cross both kinds of change without rebuilding the source.

TypeInformation.of(DataChangeRecord.class) selects the connector’s field serializer rather than reflective Kryo. The serializer preserves the immutable record, its collection order, and absent values across network and state boundaries without requiring access to JDK implementation fields. If a transformation erases a stream that still emits DataChangeRecord, declare the same type information with .returns(TypeInformation.of(DataChangeRecord.class)).

Returning successfully without emitting output skips that data-change record and increments recordsSkipped once. If deserialization fails, the split progress does not advance, so recovery can replay the data-change record under the source’s at-least-once contract. Heartbeats and child-partitions records do not call the user deserializer.

Output filters#

The source can filter tables and project columns after decoding a data-change record and before calling the user deserializer. Each entry in the four filter lists is a Java regular expression matched against the complete identifier, not a substring. Table expressions see the record’s Spanner-reported tableName. Column expressions see tableName + "." + columnName, so orders\.status does not select a same-named column from audit_orders. The connector does not fold case or interpret quoting, so patterns must use the exact table and column names carried by the decoded record.

When all four filter lists are empty, the reader passes the original decoded DataChangeRecord directly to the deserializer without evaluating the filter or creating a filter result. The reader determines this once when it is constructed, so skipMessagesWithoutChange(true) alone does not add per-record filter work. That option affects records only when a configured column filter can remove reported values. An active column filter that retains every column in the record’s metadata returns the original record before parsing or rebuilding its mod value JSON.

An include list retains an identifier when any expression matches, while an exclude list removes an identifier when any expression matches. The builder rejects setting both lists for the same table or column scope. Primary-key columns and their type metadata are always retained, even when a column expression matches them. For every other column, projection removes the same member from columnTypes and from every mod’s old and new value objects. Absent values and explicit JSON null remain distinct.

SpannerChangeStreamSource<OrderChange> source =
        SpannerChangeStreamSource.<OrderChange>builder()
                .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
                .changeStreamName("all_changes")
                .deserializer(new OrderChangeDeserializer())
                .tableIncludeList(List.of("orders", "order_items"))
                .columnExcludeList(List.of("orders\\.internal_note", ".*\\.debug_payload"))
                .build();

If projection removes every reported non-key value, the default still calls the deserializer with empty projected value objects. This preserves transaction activity that a downstream consumer might need. Setting skipMessagesWithoutChange(true) instead skips that record before deserialization.

Filters are evaluated from each record’s own table and column metadata, so a later schema addition does not require an INFORMATION_SCHEMA lookup or a source restart. Changing the filter configuration on a restored job leaves the checkpointed split positions unchanged and applies the new filters to records processed from those positions. The filter change does not itself move or replay a position, but the source’s normal inclusive restore can repeat records at the checkpoint boundary.

These options are connector-side output filters. Spanner still returns the complete record to the source process, so they do not reduce partition-query concurrency, Spanner processing, or traffic between Spanner and Flink, and they do not prevent an excluded value from entering the TaskManager process. Use the Change Stream DDL watch definition when the service must exclude tables or columns before producing records.

Partition queries and capacity#

The coordinator begins with the null partition token and schedules children only after every parent they name has finished. Each reader subtask opens several scheduled partition queries concurrently, up to maxConcurrentQueriesPerSubtask, and keeps excess restored splits in a checkpointed FIFO until capacity returns.

Before releasing readers, the coordinator reads the database dialect and the stream’s CHANGE_STREAMS, CHANGE_STREAM_TABLES, and CHANGE_STREAM_OPTIONS metadata. It logs the watch scope, effective retention, partition mode, value-capture type, and every exclusion option. It warns when the stream watches an explicit column list because columns added later are not watched automatically; alter the Change Stream when its intended schema changes. This startup check also rejects an unsupported partition mode before any query is assigned.

Each query uses executeQueryAsync on a strong single-use read-only transaction. The callback hands the reader one undrained record and pauses; the Flink mailbox emits or processes it before resuming that query. This keeps callback threads out of Flink output and bounds buffered query results to one per active query. A query failure fails the task and retains its split, while only successful end-of-query reports the partition finished.

Change Streams queries do not use Data Boost. The builder’s rpcPriority applies to every query in both dialects.

Checkpoints and delivery#

A reader checkpoint contains every active and queued split with its greatest consumed record timestamp and watermark. Recovery starts the partition query at that timestamp inclusively because several records can share one commit timestamp. The boundary can therefore be delivered again; advancing past it could skip another record from the same timestamp.

The coordinator checkpoint contains only unfinished partition entries. When an already-discovered child is still waiting for another parent, a completed parent remains only as a compact split-id proof until that child becomes schedulable. The proof then leaves the checkpoint too, so partition-lineage state follows the unfinished topology rather than all historical transitions. Enumerator-state versions 1 and 2 remain restorable and are compacted when read; new checkpoints use version 3.

The source is at least once. Downstream state that requires uniqueness should deduplicate with a domain key that includes enough of the Spanner record identity, such as the server transaction id and record sequence, rather than assuming the commit timestamp is unique.

The start position is resolved once on the coordinator. A valid restored partition ledger takes precedence over the configured fresh start. If any unfinished restored position has expired, the default is to fail; configuring resumeFallback permits discarding the whole stale ledger and starting one new null-token query, which loses the unavailable interval and can repeat records at or after the fallback. An absent explicit retention row uses absentRetentionFallback, seven days by default. Readers do not open restored queries until this retention check finishes. That ordering prevents an expired token from reaching Spanner before the coordinator can fail the restore or replace the whole ledger with the explicit fallback.

The source requires database read access for both the generated Change Stream read function and its startup INFORMATION_SCHEMA queries. In particular, the workload principal needs spanner.databases.select plus the session permissions needed by the Spanner client. The connector does not create a metadata table and needs no write permission for its partition ledger; Flink checkpoints own that state.

Event time#

Data records carry their Spanner commit timestamp as the Flink event timestamp. Heartbeat records advance their partition’s safe progress in the coordinator ledger. The source watermark is one coordinator-owned minimum across every unfinished partition, including discovered, scheduled, reader-queued, and running partitions. The coordinator broadcasts that frontier to every source reader, including readers with no assigned split, so failover and rescaling do not omit temporarily unowned work. The conversion subtracts one millisecond from the heartbeat instant after truncation because a later nanosecond timestamp can occupy the same Flink millisecond; this prevents an on-time record from being classified as late. Quiet partitions are not marked idle because doing so would remove them from the safety minimum. The default two-second heartbeat can be configured from one second through five minutes.

WatermarkStrategy.noWatermarks() in the example prevents downstream timestamp assignment from replacing these source watermarks. A job that supplies another strategy is choosing that strategy’s timestamp and watermark behavior instead.

Table API CDC#

The Table API can map one physical DDL to this source and emit either a full retract changelog or a keyed upsert changelog. Its value-capture constraints, row reconstruction, and DDL options are documented under Change Streams scan behavior.

Sink#

The destination is a database, not a table#

The sink is configured with a DatabaseDestination. Which table a record goes to is not configured at all: the serializer returns a Mutation, and a mutation names its own table. One sink therefore writes to as many tables of that database as its serializer produces.

That is a shape of its own among the connectors here. The BigQuery, Pub/Sub, Cloud Tasks and Bigtable sinks resolve a destination per record, through a resolver the builder is given; this one fixes the database and leaves the table to the mutation, so there is no resolver to configure.

Sink<OrderEvent> sink =
        SpannerSink.<OrderEvent>builder()
                .database(DatabaseDestination.of("my-project", "my-instance", "orders-db"))
                .serializer(
                        (event, context) ->
                                Mutation.newInsertOrUpdateBuilder("Orders")
                                        .set("OrderId")
                                        .to(event.getId())
                                        .set("Total")
                                        .to(event.getTotal())
                                        .build())
                .build();

How the sink writes#

The sink buffers mutations and applies them with DatabaseClient.batchWriteAtLeastOnce, one mutation per mutation group. The group is the unit Spanner reports a status for, so a refusal names exactly the record it is about — which is what makes per-record failure routing possible at all. A plain commit would fail the whole batch over one bad row.

The writer is synchronous: there is no mailbox, no callback thread and no in-flight bookkeeping, because this RPC has no asynchronous or self-batching form to wrap. It is one streaming call the writer makes and consumes to completion on the task thread. The decision, and the facts behind it, are in ADR-0075.

Skipping records#

Returning null from the serializer skips the record: it is written nowhere, is not a failure, never reaches the failure handler, and is counted by recordsSkipped. Throwing marks the record as failed and routes it instead. This is the same contract every connector here follows.

Delivery guarantee, and why the mutation operation is your decision#

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

The sink is at-least-once and stateless: it keeps nothing across checkpoints, because it empties its batch before the barrier passes. A completed checkpoint means every record up to it was applied, skipped by the serializer, or handed to the failure handler.

Spanner’s batch write has no replay protection — the service’s own documentation says a mutation may be applied more than once. A record can therefore reach the database twice: after a job restart, and also within one attempt, when a request whose outcome never arrived is re-sent.

Which mutation operation the serializer builds is what decides whether that matters:

This per-mutation idempotence is not a latest-value ordering guarantee. The writer sends records as separate BatchWrite mutation groups, and Spanner may apply independent groups in an unspecified order. Successive records for the same key can therefore finish out of input order even without a replay.

Error handling#

Every refusal Spanner reports comes back per mutation group, so the sink can act per record. It sorts them three ways.

Routed to the failedMutationHandler — the service refused this one mutation and would refuse it again:

• ALREADY_EXISTS — a replayed insert, or a collision on a UNIQUE index.
• INVALID_ARGUMENT — a malformed argument.

Retried on the sink’s own budget — the service, not the mutation: ABORTED, UNAVAILABLE, DEADLINE_EXCEEDED, RESOURCE_EXHAUSTED.

Fails the job — everything else, and every failure of the request whatever its status. A request-level failure names no mutation, and the mutations it carried have no reported outcome, so dropping them would discard records the service may not have looked at yet.

What does not reach the failure handler, and why#

This is the part worth reading before configuring a dead-letter queue. Measured against the emulator, one run, 2026-08-09:

So the ordinary schema violations are not routed by default — and that default is a choice you can change.

constraintViolationPolicy(ROUTE_TO_FAILURE_HANDLER) moves both of those statuses into the failure handler, which then decides between failing, dropping and dead-lettering like it does for every other per-record refusal. The default is FAIL_JOB for two reasons:

• A constraint violation usually says the mapping from records to columns is wrong, not that one record is anomalous. Every record of that shape will be refused, so shedding them one at a time hides a systematic problem behind a green job.
• FAILED_PRECONDITION has one documented cause that is neither data nor permanent: while a database’s CMEK key is disabled, destroyed or unreachable, every write is refused with it, and service is restored automatically when the key comes back. A job routing that status into a dropping handler would shed its whole stream through a key incident rather than waiting it out.

Failing the job loses nothing: the records are replayed from the source on restart. What it costs is progress, which is the trade the opt-in exists to let you make differently.

Two things that are not covered by that policy, because they are not constraint violations: a duplicate key or unique-index collision is ALREADY_EXISTS and is always routed, and a schema change in progress is ABORTED, which the sink retries.

NOT_FOUND is not routed either: a missing table or column fails every record alike, so it is a configuration error rather than a bad row. This sink creates nothing — no database, no table, no column.

The last row of that table is the uncomfortable one. An update whose row has since been deleted is data, and it still fails the job, because NOT_FOUND does not say whether the table or the row is the thing that is missing — and routing it would mean dropping every record of a job that merely has a table name wrong. If your stream can legitimately update rows that may be gone, insertOrUpdate is the operation that says so.

The reasoning, the full measurement and the reopen condition are in ADR-0076.

Retries belong to the sink#

Unlike every other Google client this project builds on, the Spanner client library does not retry the RPC this sink writes with — its generated settings give batch write an empty retryable-code set. So the retry loop is the sink’s: Spanner’s recovery knobs budget the write retry loop itself, whereas Bigtable’s recovery knobs budget only table auto-creation repair.

A retry re-sends exactly what is still undecided: the mutations whose group came back transient, plus the mutations whose group the service never reported on, which is what a stream failing part-way through leaves behind. Mutations already applied are never re-sent, so a retry does not multiply the duplicates an at-least-once sink can produce. Exhausting recoveryMaxAttempts fails the job.

batchWriteTimeout bounds one complete attempt, including a response stream that reports some groups and then stalls. The 30-second default replaces the client library’s one-hour timeout only for the data client’s BatchWrite RPC; reads and administration keep their client-library settings. A deadline is a transient failure, so the connector retries only the mutations whose outcome is still undecided and increments errorClass.DEADLINE_EXCEEDED.errors and mutationsRetried.

The maximum time spent in the write loop is bounded by the attempts, the per-attempt timeout, and the jittered backoffs:

A × T + (1 + j) × Σ(i=1..A-1) min(Bmax, B0 × 2^(i-1))

With the defaults, the upper bound is 369.375 seconds: 300 seconds in ten attempts and up to 69.375 seconds in nine backoffs. This is the bound for one invocation of the write loop, not for a checkpoint. A record-triggered synchronous flush may already be running when a checkpoint barrier reaches the sink, and the checkpoint flush may then invoke the loop again. Choose the attempt timeout and recovery schedule so that this work, processing, alignment, other operators, and checkpoint transport fit within the checkpoint timeout. If larger batches or service load make valid attempts approach the configured timeout, raise batchWriteTimeout and recompute the combined budget.

Dead-letter payloads#

FailedMutation.getPayloadBytes() is the Java-serialized Mutation, not a protobuf. That is not a preference: the client library exposes no public route from a Mutation to its wire form, and its debug rendering truncates every string value at 36 characters, so it cannot stand in. A handler that wants the mutation itself should take FailureHandler<FailedMutation> and read getMutation(); the bytes exist for the cross-connector dead-letter queue, which sees only the shared FailedElement view.

Batch limits and metrics#

This section covers write-batch correctness bounds first, then the metrics that show which limit binds and how the source and sink behave.

Batching#

A request Spanner refuses is refused as a whole, taking every mutation in it with it, so a sink that accumulates mutations has to bound the request it builds. That is what the three batch limits are for, and it makes them correctness rather than tuning.

Which limit each one defends is narrower than three knobs make it look. Spanner’s quotas page documents “mutations per commit (including indexes)” of 80,000 and a “commit size” of 100 MiB — both about Commit, which this sink does not use. The one row that names batch write is “mutations per mutation group in a batch write request”, also 80,000. The batch write page adds a single sentence, about size only: “the maximum size for a batch write request is the same as the limit for a commit request”. So:

• No per-request mutation count is documented for batch write at all. maxBatchCells and maxBatchMutations bound the request as a proxy for its size, and keep a batch far below the per-mutation-group 80,000 — which this sink, putting one mutation in each group, reaches only through a single mutation that breaches it alone: a range delete over a table with secondary indexes, which costs one mutation for the table plus one per index for every row the range matches. On a table with no secondary index a range delete costs one however many rows it hits.
• maxBatchBytes is the one defending a documented request-level limit, and how large that limit is could be read two ways: 100 MiB by the sentence above, or 10 MiB by the quotas page’s “request size other than for commits”. The gated real-GCP suite measured it (#441): the service accepts a request of roughly 12 MiB and refuses one of roughly 110 MiB, naming 104857600 bytes — 100 MiB exactly. Note what the refusal is: a transport-level RESOURCE_EXHAUSTED, which this sink treats as transient and retries, not a fail-fast rejection.

All three are bounded at the setter, so a value a request could not carry fails the job at submission rather than on a task manager:

• maxBatchBytes at 100 MiB — the figure the service was measured to enforce. This is the ceiling that defends a refusal Spanner documents.
• maxBatchCells at 80,000. Precautionary rather than a refusal anyone has seen: Spanner documents no request-level mutation count either way, so the cap holds a batch to the only mutation figure it does publish.
• maxBatchMutations at 80,000 too — and derived from the cell ceiling rather than repeated, because every mutation costs at least one cell, so a batch never holds more mutations than cells.

The three limits are ANDed — a batch flushes on whichever binds first — so raising one alone usually changes nothing: maxBatchCells of 500,000 against the default 1 MiB and 500 mutations produces exactly the batches the defaults did. maxBatchCells and maxBatchBytes are the pair to reach for; lower maxBatchMutations to cap a batch by count regardless of how wide the rows are.

That ANDing has one case worth naming, because nothing else would tell you: maxBatchMutations set above the configured maxBatchCells can never take effect, since the cell cap is reached first however cheap the mutations are. The configuration works, so it is not refused — building the options logs a warning instead, naming both values. It is written wherever the job’s main runs: the client log under flink run, the JobManager log in application mode, the console in an IDE. The warning suggests no remedy, because neither obvious one is safe — lowering maxBatchMutations below maxBatchCells does not make the count cap bind (what each mutation costs in cells decides that), and raising maxBatchCells spends the headroom above.

maxBatchCells is counted the way Spanner counts a mutation. A written column costs one cell for the table plus one for every secondary index that contains it — as a key column or as a STORING column, since both rewrite an index entry — and a delete costs one plus the table’s index entries. The index part is a property of the schema, so the sink reads it from the database’s INFORMATION_SCHEMA once, when the writer opens. The read covers every visible user schema and keeps the schema in each table’s identity, so equal short table names in different schemas receive independent weights. GoogleSQL names match case-insensitively, while PostgreSQL catalog names preserve the distinction created by quoted identifiers. Mutation.getTable() carries a native data-API name rather than SQL syntax, so a named-schema serializer supplies the decoded schema.table catalog name without backticks or double quotes. That read needs spanner.databases.select as well as write access. On a wide row that count is a better proxy for the request’s size than a mutation count is, which is why the sink keeps it rather than counting mutations alone.

Two consequences worth knowing:

• A table the sink did not see is counted without its index entries — one created after the job started, or one hidden from the writer’s database role. That undercounts, and the default maxBatchCells of 5,000 is deliberately 16 times under the 80,000 ceiling so the undercount has room. Raising the limit toward 80,000 removes that room.
• The byte limit is an estimate. The client library exposes no public way to size a Mutation as it goes on the wire, so the sink adds up the values it can see and ignores framing — about sixty bytes a mutation, measured — and so it reads low. The default of 1 MiB sits 100 times under the 100 MiB limit, which is the room the estimate is allowed to be wrong in. One value type is not counted as itself: a BYTES column is counted at its base64 length, because a Spanner value travels inside a google.protobuf.Value, which has no bytes kind. That was measured, not assumed — 83,886,080 raw bytes arrived as 111,852,884, four thirds of themselves. Counting the raw length made the estimate read a quarter low for a BYTES-heavy batch, which is more than the room above once maxBatchBytes is raised toward its ceiling.

The defaults are Apache Beam’s, and Beam batches for Commit rather than for batch write — which is where the commit-shaped 80,000 entered this connector. They sit far under every reading of every limit all the same.

There is no primary-key sorted batching. Apache Beam defaults its grouping factor to 1 — which skips its sort — for unbounded input, and a streaming sink’s per-checkpoint batches do not amortize one.

ADR-0077 carries the measurements and the declined alternatives.

Sink metrics#

Registered on the sink writer’s metric group.

Two of these mean something here that they cannot mean on the sibling sinks. mutationsRetried and the transient half of errorClass.CODE.errors are visible because this connector performs its own retries — on the connectors whose SDK retries internally, the same work happens out of sight and only the give-ups are counted. Read bufferedCells beside bufferedBytes to see which of the three batch limits is actually firing, and therefore which knob to move.

There are no per-destination counters. The sink writes one database but any number of its tables, so the meaningful cut would be per table — and its cardinality is the serializer’s to decide, not a bill the connector should sign on your behalf.

currentSendTime is deliberately unset: a batch write’s latency covers a whole request of unrelated mutations, so attributing it to records would say nothing an operator can act on.

Batch source metrics#

Registered on the split enumerator’s group, on the coordinator, and on the source reader’s group, on each subtask.

There is no bytes-read counter and no rows-remaining gauge. The client hands over a decoded Struct and says nothing about what it cost on the wire, so any byte figure would be this connector’s arithmetic wearing the look of the quantity Spanner bills for; and a partition is an opaque token, so nothing knows how many rows are left inside one.

Change Streams source metrics#

Registered on the split enumerator’s coordinator group and on each source reader’s group.

Heartbeats move the coordinator-owned source watermark without incrementing numRecordsIn. Child-partitions records and query-completion signals are coordinator events, so neither counter treats them as user records. Table-filtered records and records skipped without a projected change advance partition progress but do not increment numRecordsIn or recordsSkipped. changeStreamColumnOccurrencesFiltered counts one occurrence for removed columnTypes metadata and one for each removed member in each old or new value object, and counts only records passed to the deserializer. The connector does not register its own copies of the standard metrics: the Flink source runtime derives the metrics available in that Flink version from the commit timestamps and source watermarks the reader emits. No metric uses a partition token as a label, because split and merge would make those labels unbounded.

Read the query and lag metrics together:

• If aggregate activeChangeStreamQueries reaches source parallelism * maxConcurrentQueriesPerSubtask while queued or unassigned lag grows, increase source parallelism or the per-subtask bound after checking subtask resources.
• If active query counts remain uneven after restoring at a new parallelism, check split redistribution and stalled readers before adding capacity.
• Queued partitions beside a free query slot indicate an unhealthy reader scheduling or backpressure path.
• Unassigned partitions beside free reader query slots indicate an unhealthy split-request or coordinator assignment path.
• The partition-ledger gauge should follow current unfinished topology rather than job age; growth without corresponding active, queued, or unassigned work points to partitions that are not completing.
• Finished-parent proofs are transient fan-in dependencies; a sustained count means created children are still waiting for other parents to finish.
• Alert before either partition-lag gauge approaches the configured Change Streams retention period.

Changing maxConcurrentQueriesPerSubtask changes the capacity of each running reader. Changing source parallelism requires a restart from a checkpoint or savepoint so Flink can redistribute split ownership. TaskManager slots supply resources for that parallelism but do not redistribute queries themselves. The connector exposes these capacity signals but does not change operator parallelism automatically.

Testing#

Functional coverage runs against the Cloud Spanner emulator in testcontainers over both dialects. The sink tests drive the production writer-creation path, so the client, schema read, and batch write are the real ones. The Change Streams tests run the production source through a MiniCluster across schema and value-capture changes, and separate failover jobs require complete at-least-once output after recovery in both dialects. The gated real-GCP class adds the service-only acceptance: both physical result shapes, metadata and retention discovery, explicit-column warnings, exclusion-option logging, a start before stream creation, checkpoint restart, savepoint restore, heartbeat watermarks, service-created child partitions, expired-state failure, and explicit fallback.

The emulator image is pinned separately from the other connectors’ google-cloud-cli bundle: the Spanner emulator implements the BatchWrite RPC only from v1.5.31, and the bundled one predates it, so this sink’s entire write path would answer UNIMPLEMENTED against it.

Emulator deviations#

An emulator is a convenience for fast feedback, never evidence about the service. Where the two disagree, the service decides. Known deviations, and what covers them instead:

The gated real-GCP suite#

Everything above runs in an ordinary build. What only the service can answer runs in a separate, opt-in suite (#224): the rejection statuses and their per-group reporting, the mutation-cell weights read from the service’s own INFORMATION_SCHEMA in both dialects, how many partitions Spanner plans and for which query shapes, and Data Boost end to end. The source class also verifies named-schema Table API writes, bounded index scans, and synchronous and asynchronous lookups in both dialects without creating another billed instance. It now verifies Change Streams in both dialects on the same ephemeral instance. The measured recovery run delivered all 5,000 unique mutation ids after an intentional post-checkpoint failure, with 500 repeated deliveries at the inclusive checkpoint boundary, then restored a savepoint and consumed a mutation written while the job was stopped. It also verifies that restored readers wait for retention validation, so expired state fails in the coordinator unless an explicit fallback replaces it. It is also the only place the connector’s clients reach the real service. The suite authenticates those clients through ADC; configured key-file coverage instead verifies runtime loading and the exact non-emulator client settings without opening a service connection.

The suite is opt-in per command rather than per shell: each class carries @Tag("gated"), which every ordinary build excludes, and just e2e is the one thing that clears the exclusion. Each class creates a 100-processing-unit regional Spanner instance in the STANDARD edition, uses it, and deletes it — nothing persistent is provisioned, because an instance bills for as long as it exists. STANDARD is enough, and the suite shows it rather than citing it: the Data Boost tests run on that instance and read every row, so exercising Data Boost needs no edition upgrade. Instance names carry their creation time, so a run that dies before its teardown is reclaimed both by the next run and by a scheduled sweep.

Scope and provenance#

The module README records implementation status and provenance. The connector-specific design decisions on this page link to their durable ADRs.

Not here yet#

• MUTABLE_KEY_RANGE partition-event records; the Change Streams source supports IMMUTABLE_KEY_RANGE and rejects another mode at startup.