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

    • 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

    • 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