Package io.github.flink.gcp.connector.spanner.source

Class SpannerSourceBuilder<T>

java.lang.Object
io.github.flink.gcp.connector.spanner.source.SpannerSourceBuilder<T>
Type Parameters:
T - the record type produced
@Public public class SpannerSourceBuilder<T> extends Object
Builds a SpannerSource.

Everything is validated at build() or at the setter that took it, so a configuration mistake fails where the job is assembled rather than on a JobManager once the read is planned.

  • Field Details

    • DEFAULT_MAX_ROWS_PER_FETCH

      public static final int DEFAULT_MAX_ROWS_PER_FETCH
      The default maximum rows one fetch hands to the task thread.
      See Also:

    • DEFAULT_MAX_BYTES_PER_FETCH

      public static final long DEFAULT_MAX_BYTES_PER_FETCH
      The default target maximum decoded input bytes one fetch hands to the task thread.
      See Also:

  • Method Details

    • database

      public SpannerSourceBuilder<T> database(DatabaseDestination database)
      Sets the database to read. Required.
      Parameters:
      database - the database
      Returns:
      this builder

    • readOperation

      public SpannerSourceBuilder<T> readOperation(SpannerReadOperation readOperation)
      Sets what to read: a query, or a table with its columns and key set. Required.
      Parameters:
      readOperation - the read operation
      Returns:
      this builder

    • deserializer

      public SpannerSourceBuilder<T> deserializer(SpannerStructDeserializationSchema<T> deserializer)
      Sets the deserializer turning rows into records. Required.
      Parameters:
      deserializer - the deserializer
      Returns:
      this builder

    • timestampBound

      public SpannerSourceBuilder<T> timestampBound(com.google.cloud.spanner.TimestampBound timestampBound)
      Sets the snapshot to read at. Optional; the default is TimestampBound.strong(), the latest committed data.

      Only strong(), ofReadTimestamp and ofExactStaleness can bound a batch read. The other two modes are refused here rather than on a JobManager: Spanner restricts them to single-use transactions, and a batch read is by construction a multi-use one — every partition rejoins the same transaction.

      A stale read is cheaper for the service to serve and can be answered by any replica; a strong one may have to reach the leader. Nothing about the source changes either way — every partition is read at the one timestamp the plan fixed.

      Parameters:
      timestampBound - the bound
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the bound is one a batch read cannot take

    • maxPartitions

      public SpannerSourceBuilder<T> maxPartitions(long maxPartitions)
      Hints how many partitions the read should be divided into. Optional; unset leaves the choice entirely to the service.

      A hint, and Spanner documents it as one: it may return more or fewer. Setting it to the job's parallelism is the reasonable thing to ask for, not something to rely on having got.

      Parameters:
      maxPartitions - the desired maximum number of partitions
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the value is not positive

    • partitionSizeBytes

      public SpannerSourceBuilder<T> partitionSizeBytes(long partitionSizeBytes)
      Hints how much data each partition should cover. Optional; unset leaves the choice entirely to the service.

      A hint, like maxPartitions(long), and the two are asked for together at the discretion rather than combined by this connector.

      Parameters:
      partitionSizeBytes - the desired size of one partition, in bytes
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the value is not positive

    • dataBoostEnabled

      public SpannerSourceBuilder<T> dataBoostEnabled(boolean dataBoostEnabled)
      Runs the read on Data Boost's independent compute. Optional; off by default.

      Data Boost serves a partitioned read from 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, the read is billed separately, and 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.

      Parameters:
      dataBoostEnabled - whether to enable Data Boost
      Returns:
      this builder

    • rpcPriority

      public SpannerSourceBuilder<T> rpcPriority(SpannerRpcPriority rpcPriority)
      Sets the priority Spanner schedules the reads at. Optional; unset leaves the service's own handling in place, which is the same as HIGH.

      LOW is what a backfill of a large table wants: Spanner sheds low-priority work first when an instance is at capacity, so the read yields to the traffic the instance is serving rather than competing with it. MEDIUM is a step down from the default rather than a restatement of it, since Spanner treats an unspecified priority as HIGH.

      The priority applies to the reads that move the rows, which is where a job's load on the instance actually is. It does not apply to the one call that plans the partitions.

      Not a substitute for dataBoostEnabled(boolean): a low-priority read still runs on the instance's own compute and still competes for it, while Data Boost does not use it at all. Lowering the priority costs nothing extra; Data Boost is billed separately.

      Parameters:
      rpcPriority - the priority
      Returns:
      this builder

    • serviceAccountKeyFile

      public SpannerSourceBuilder<T> serviceAccountKeyFile(String serviceAccountKeyFile)
      Authenticates the source with the service-account JSON key at the given path instead of application-default credentials. The JobManager reads the file when it creates or restores the enumerator, and each TaskManager reads it when it creates a reader. Every eligible process must therefore see the same path. Optional; when unset the real-service path uses application-default credentials.

      Service-account keys are long-lived secrets. Prefer an attached service account or Workload Identity where the deployment supports one. This setting cannot be combined with emulatorEndpoint(String), whose plaintext channel carries no credentials.

      Parameters:
      serviceAccountKeyFile - the service-account JSON key-file path
      Returns:
      this builder

    • emulatorEndpoint

      public SpannerSourceBuilder<T> emulatorEndpoint(String emulatorEndpoint)
      Points the source at an emulator, over a plaintext channel with no credentials. Never production.
      Parameters:
      emulatorEndpoint - the emulator endpoint as host:port
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the endpoint is not host:port with a port in 1..65535

    • maxRowsPerFetch

      public SpannerSourceBuilder<T> maxRowsPerFetch(int maxRowsPerFetch)
      Sets the maximum input rows one fetch hands to Flink's element queue. Optional; defaults to 1000.

      The fetch returns when either this row limit or maxBytesPerFetch(long) is reached. Lower values can reduce the source reader's queued memory and the delay before a checkpoint or cancellation can be observed, at the cost of more fetch hand-offs. This is a TaskManager hand-off bound, not a hint for how Spanner plans read partitions.

      Parameters:
      maxRowsPerFetch - the positive maximum number of rows per fetch
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the value is not positive

    • maxBytesPerFetch

      public SpannerSourceBuilder<T> maxBytesPerFetch(long maxBytesPerFetch)
      Sets the target maximum decoded input bytes one fetch hands to Flink's element queue. Optional; defaults to 12582912L bytes.

      The estimate counts logical field content rather than JVM object overhead. A fetch stops before a row would take a non-empty batch over the target, while one oversized row is handed over alone so the source always makes progress. Lower values can reduce the source reader's queued memory for multi-row batches, while higher values can reduce fetch hand-offs for wide rows. This is a TaskManager hand-off bound, not a Spanner transport or partition-planning setting.

      Parameters:
      maxBytesPerFetch - the positive target maximum estimated bytes per fetch
      Returns:
      this builder
      Throws:
      IllegalArgumentException - if the value is not positive

    • exportClientMetricsToFlink

      public SpannerSourceBuilder<T> exportClientMetricsToFlink(boolean exportClientMetricsToFlink)
      Exports the Spanner client's own metrics through Flink's metric system, so that every configured Flink metric reporter sees them. Optional; off by default.

      The split enumerator and each reader register them under their own metric group as rpcMethod.METHOD.leaf, for the RPC methods their clients call: per method, the client's operation and attempt counts and latencies, and per attempt the Google front end's and the Spanner API front end's timing and the attempts that arrived without it. Latencies are reported as a count and a millisecond sum, so a backend divides their rates for the mean over any window; no percentile is reported. Against an emulator the client records nothing, so the series stay at zero.

      This does not stop the client's built-in export of the same metrics to Cloud Monitoring; exportClientMetricsToCloudMonitoring(boolean) does.

      Parameters:
      exportClientMetricsToFlink - whether to export the client's metrics to Flink
      Returns:
      this builder

    • exportClientMetricsToCloudMonitoring

      public SpannerSourceBuilder<T> exportClientMetricsToCloudMonitoring(boolean exportClientMetricsToCloudMonitoring)
      Sets whether the Spanner clients of the split enumerator and the readers take part in the client's built-in export of client-side metrics to Cloud Monitoring. Optional; on by default, unless the SPANNER_DISABLE_BUILTIN_METRICS environment variable turns the export off.

      The client exports through one exporter per class loader, which the first client that takes part starts and which stays until the JVM exits. Off, this source's clients neither start it nor record into it; an exporter another client started keeps running. It is independent of exportClientMetricsToFlink(boolean), which neither needs nor stops this export.

      Parameters:
      exportClientMetricsToCloudMonitoring - whether the clients export to Cloud Monitoring
      Returns:
      this builder

    • build

      public org.apache.flink.api.connector.source.Source<T,BatchReadSplit,SpannerBatchReadEnumeratorState> build()
      Builds the source.
      Returns:
      the source
      Throws:
      IllegalStateException - if a required option was not set