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

Class SpannerChangeStreamSourceBuilder<T>

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

  • Field Details

    • DEFAULT_HEARTBEAT_INTERVAL

      public static final Duration DEFAULT_HEARTBEAT_INTERVAL
      Default heartbeat interval. Heartbeats are what advance a quiet partition's watermark; see heartbeatInterval(Duration).

    • DEFAULT_ABSENT_RETENTION_FALLBACK

      public static final Duration DEFAULT_ABSENT_RETENTION_FALLBACK
      Default retention assumed when the change stream declares no explicit retention; see absentRetentionFallback(Duration).

    • DEFAULT_MAX_CONCURRENT_QUERIES_PER_SUBTASK

      public static final int DEFAULT_MAX_CONCURRENT_QUERIES_PER_SUBTASK
      Default bound on the open change-stream queries in one source subtask; see maxConcurrentQueriesPerSubtask(int).
      See Also:

  • Method Details

    • database

      public SpannerChangeStreamSourceBuilder<T> database(DatabaseDestination database)
      Sets the database holding the change stream. Required.
      Parameters:
      database - the database
      Returns:
      this builder

    • changeStreamName

      public SpannerChangeStreamSourceBuilder<T> changeStreamName(String changeStreamName)
      Sets the change stream to read. Required.

      The name is the one the CREATE CHANGE STREAM statement gave it; the source calls the read function Spanner generates for it.

      Parameters:
      changeStreamName - the change stream name, not blank and without surrounding whitespace
      Returns:
      this builder

    • deserializer

      public SpannerChangeStreamSourceBuilder<T> deserializer(SpannerChangeStreamDeserializationSchema<T> deserializer)
      Sets the schema turning each data-change record into output records. Required.

      Zero or more outputs per record, collected synchronously; see the schema's own documentation for what emitting nothing means.

      Parameters:
      deserializer - the deserialization schema
      Returns:
      this builder

    • startPosition

      public SpannerChangeStreamSourceBuilder<T> startPosition(StartPosition startPosition)
      Sets where a fresh stream begins. Optional; defaults to StartPosition.latest().

      Resolved once, on the coordinator. It is ignored on a restore, where the checkpointed partition ledger decides — see resumeFallback(StartPosition) for what happens when that ledger has expired.

      Parameters:
      startPosition - the start position
      Returns:
      this builder

    • resumeFallback

      public SpannerChangeStreamSourceBuilder<T> resumeFallback(StartPosition resumeFallback)
      Sets where to restart when a restored partition position has fallen outside the stream's retention. Optional; unset means such a restore fails.

      Setting it is a decision to lose data rather than stop: the coordinator discards the whole checkpointed ledger and starts one new query from here, so the unavailable interval is never read and records at or after this position can be delivered again.

      Parameters:
      resumeFallback - the fallback start position
      Returns:
      this builder

    • boundedTimestamp

      public SpannerChangeStreamSourceBuilder<T> boundedTimestamp(Instant boundedTimestamp)
      Makes the source bounded at a commit timestamp. Optional; unset means the source is unbounded.

      The bound is inclusive: every partition query, including those of the child partitions the service reports before the bound, reads the records committed at or before it and then finishes, and the source finishes once every partition has. A fresh start position, or a resumeFallback(StartPosition), that resolves to an instant after the bound fails the source when its coordinator starts.

      The bound belongs to the checkpointed partition ledger: a restore must configure the same bound, or none for a ledger checkpointed without one, and otherwise fails rather than read to a bound other than the one configured. A restore whose expired ledger the resume fallback replaces starts under the configured bound.

      Parameters:
      boundedTimestamp - the last commit timestamp the source reads, within Spanner's TIMESTAMP range of 0001-01-01T00:00:00Z through 9999-12-31T23:59:59.999999999Z
      Returns:
      this builder

    • absentRetentionFallback

      public SpannerChangeStreamSourceBuilder<T> absentRetentionFallback(Duration fallback)
      Sets the retention used when INFORMATION_SCHEMA.CHANGE_STREAM_OPTIONS has no explicit retention row. The default is seven days. It must be longer than the one-minute retention safety margin used when resolving the earliest readable timestamp.
      Parameters:
      fallback - the retention to assume
      Returns:
      this builder

    • heartbeatInterval

      public SpannerChangeStreamSourceBuilder<T> heartbeatInterval(Duration interval)
      Sets the service heartbeat interval. Optional; defaults to two seconds.

      Heartbeats are what advance a quiet partition's watermark, so this is the upper bound on how long the source's event time can stand still while the stream is idle.

      Parameters:
      interval - the heartbeat interval, from one second through five minutes and expressible in whole milliseconds
      Returns:
      this builder

    • rpcPriority

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

      Applies to every partition query in both dialects. Change Streams queries do not use Data Boost, so this is the only lever over what they cost the instance.

      Parameters:
      rpcPriority - the priority
      Returns:
      this builder

    • maxConcurrentQueriesPerSubtask

      public SpannerChangeStreamSourceBuilder<T> maxConcurrentQueriesPerSubtask(int maximum)
      Bounds the TVF partition queries opened concurrently by one source subtask.

      The default is eight. Job-wide configured capacity is source parallelism multiplied by this value; it is a connector bound, not a published Spanner quota.

      Parameters:
      maximum - the per-subtask query bound, positive
      Returns:
      this builder

    • tableIncludeList

      public SpannerChangeStreamSourceBuilder<T> tableIncludeList(Collection<String> patterns)
      Includes tables whose Spanner-reported names fully match at least one Java regular expression.

      An empty collection disables this filter. It is mutually exclusive with tableExcludeList(Collection).

      Parameters:
      patterns - the Java regular expressions, each matching a complete table name
      Returns:
      this builder

    • tableExcludeList

      public SpannerChangeStreamSourceBuilder<T> tableExcludeList(Collection<String> patterns)
      Excludes tables whose Spanner-reported names fully match at least one Java regular expression.

      An empty collection disables this filter. It is mutually exclusive with tableIncludeList(Collection).

      Parameters:
      patterns - the Java regular expressions, each matching a complete table name
      Returns:
      this builder

    • columnIncludeList

      public SpannerChangeStreamSourceBuilder<T> columnIncludeList(Collection<String> patterns)
      Includes non-key columns whose table.column identifiers fully match at least one Java regular expression.

      Primary-key columns are always retained. An empty collection disables this filter. It is mutually exclusive with columnExcludeList(Collection).

      Parameters:
      patterns - the Java regular expressions, each matching a complete table.column identifier
      Returns:
      this builder

    • columnExcludeList

      public SpannerChangeStreamSourceBuilder<T> columnExcludeList(Collection<String> patterns)
      Excludes non-key columns whose table.column identifiers fully match at least one Java regular expression.

      Primary-key columns are always retained. An empty collection disables this filter. It is mutually exclusive with columnIncludeList(Collection).

      Parameters:
      patterns - the Java regular expressions, each matching a complete table.column identifier
      Returns:
      this builder

    • skipMessagesWithoutChange

      public SpannerChangeStreamSourceBuilder<T> skipMessagesWithoutChange(boolean skip)
      Skips a data-change record when column filtering removes every non-key value it reported.

      The default is false, which delivers the record with empty projected value objects so that transaction activity remains visible.

      Parameters:
      skip - whether to skip such a record
      Returns:
      this builder

    • serviceAccountKeyFile

      public SpannerChangeStreamSourceBuilder<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 the coordinator initializes, and each TaskManager reads it when its reader opens. Only the path is serialized. 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 SpannerChangeStreamSourceBuilder<T> emulatorEndpoint(String emulatorEndpoint)
      Points the source at a Spanner emulator, over a plaintext channel with no credentials. Never production.

      The endpoint is parsed here rather than when the source connects, so a malformed value fails where the job is assembled instead of on a TaskManager.

      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

    • exportClientMetricsToFlink

      public SpannerChangeStreamSourceBuilder<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 coordinator (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. A Change Streams query is one operation per stream it opens, counted when the stream ends; a stream the client resumes is a new one. 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 SpannerChangeStreamSourceBuilder<T> exportClientMetricsToCloudMonitoring(boolean exportClientMetricsToCloudMonitoring)
      Sets whether the Spanner clients of the coordinator 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 SpannerChangeStreamSource<T> build()
      Builds the source.
      Returns:
      the source
      Throws:
      IllegalStateException - if a required option was not set, if both sides of a mutually exclusive filter pair were set, or if a service-account key file was combined with an emulator endpoint