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

Class BigtableSinkBuilder<T>

java.lang.Object
io.github.flink.gcp.connector.bigtable.sink.BigtableSinkBuilder<T>
Type Parameters:
T - type of the records written by the sink
@Public public class BigtableSinkBuilder<T> extends Object
Builder for Bigtable sinks, obtained from BigtableSink.builder().

Required settings: a destination — table(TableDestination) for one fixed table, or destinationResolver(DestinationResolver) to route per record — and a serialization schema. Staged exactly-once delivery additionally requires an explicit transactional profile and reserved marker family through stagedOptions(BigtableStagedOptions).

By default the sink creates no table: every table it writes to, and the column families the mutations name, must exist. createDisposition(CreateDisposition) with CreateDisposition.CREATE_IF_NEEDED and tableCreateOptions(TableCreateOptions) opts into creating them for at-least-once delivery, from one schema that serves every table the sink creates.

  • Method Details

    • deliveryGuarantee

      public BigtableSinkBuilder<T> deliveryGuarantee(BigtableDeliveryGuarantee value)
      Selects eager writes or checkpoint-owned writes; final service acceptance remains required.

    • stagedOptions

      public BigtableSinkBuilder<T> stagedOptions(BigtableStagedOptions value)
      Configures checkpoint-owned writes, including the explicitly provisioned marker family.

    • table

      public BigtableSinkBuilder<T> table(TableDestination table)
      Writes every mutation to the given table. Sugar for a DestinationResolver returning that table for every record; this and destinationResolver(DestinationResolver) set the same field, so the last call wins.
      Parameters:
      table - the destination table
      Returns:
      this builder

    • destinationResolver

      public BigtableSinkBuilder<T> destinationResolver(DestinationResolver<? super T> destinationResolver)
      Resolves the destination table per record, so one sink writes to many tables. The resolver runs before the serializer, and its result is what a failed mutation is reported against.

      This and table(TableDestination) set the same field, so the last call wins.

      Each distinct table costs a bulk mutation batcher of its own, and beside CreateDisposition.CREATE_IF_NEEDED each unseen table is created from the one tableCreateOptions(TableCreateOptions) schema — so a resolver's cardinality decides what the sink holds, and what it may create.

      Parameters:
      destinationResolver - the resolver
      Returns:
      this builder

    • serializer

      public BigtableSinkBuilder<T> serializer(BigtableSerializationSchema<? super T> serializer)
      Sets the record serialization schema. Required; a schema returning null skips the record rather than failing it.
      Parameters:
      serializer - the serialization schema
      Returns:
      this builder

    • appProfileId

      public BigtableSinkBuilder<T> appProfileId(String appProfileId)
      Routes the client's requests through the given application profile, which is what selects an instance's routing policy and its priority. Optional; when unset the instance's default profile applies.

      It is a sink option rather than part of TableDestination because it chooses a path to the data, not the data's address.

      Parameters:
      appProfileId - the application profile id
      Returns:
      this builder

    • writerOptions

      public BigtableSinkBuilder<T> writerOptions(BigtableWriterOptions writerOptions)
      Sets the writer tuning options (the batch thresholds and the in-flight bounds). Optional; defaults to a fresh builder's values. This setting is rejected for staged delivery, whose RPC settings belong to BigtableStagedOptions.
      Parameters:
      writerOptions - the options
      Returns:
      this builder

    • failedMutationHandler

      public BigtableSinkBuilder<T> failedMutationHandler(FailureHandler<? super FailedMutation> failedMutationHandler)
      Sets the policy for mutations that terminally fail — a record the serializer rejects, and a mutation Bigtable rejects as malformed or oversized. Defaults to FailureHandler.failJob(); transient failures never reach it, since the client retries them and an exhausted retry fails the job in the default at-least-once mode. Staged delivery requires failJob() and makes one SDK attempt per conditional request.
      Parameters:
      failedMutationHandler - the handler
      Returns:
      this builder

    • serviceAccountKeyFile

      public BigtableSinkBuilder<T> serviceAccountKeyFile(String serviceAccountKeyFile)
      Authenticates the sink with the service-account JSON key at the given path instead of application-default credentials. The file is read on each TaskManager when its writer or staged committer creates service clients, so every eligible TaskManager must see the same path. Optional; when unset the sink 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 BigtableSinkBuilder<T> emulatorEndpoint(String emulatorEndpoint)
      Points the sink at a Bigtable emulator instead of the production service. The connection to the given host:port uses a plaintext channel with no credentials, so this must only ever be used against an emulator. Optional; when unset the sink connects to Bigtable with application-default credentials.

      The value is parsed here, so a malformed host:port is rejected on the client instead of surfacing as a connection failure once the job has been deployed.

      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 BigtableSinkBuilder<T> exportClientMetricsToFlink(boolean exportClientMetricsToFlink)
      Exports the Bigtable client's own metrics through Flink's metric system, so that every configured Flink metric reporter sees them. Optional; off by default.

      They are registered as rpcMethod.METHOD.leaf: by the writer on its metric group, for MutateRows, or under EXACTLY_ONCE by the committer on its own, for CheckAndMutateRow, since the staging writer has no client. Per RPC method they are the client's operation and attempt counts and latencies, its retries, its connectivity errors, and its server, first-response and application time. The client's throttling time is not reported: the pinned client records it in the wrong unit. 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. Each series moves when the client records into it: the attempt series as each RPC attempt ends, the operation series as each operation ends.

      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 BigtableSinkBuilder<T> exportClientMetricsToCloudMonitoring(boolean exportClientMetricsToCloudMonitoring)
      Sets whether the Bigtable client keeps exporting its built-in client-side metrics to Cloud Monitoring. Optional; on by default.

      Off, the client creates no Cloud Monitoring exporter at all. Besides the export, that removes the exporter's platform detection when a client closes: off Google Cloud, where nothing answers on the metadata server's address, each close otherwise waits for that connection attempt to time out. It is independent of exportClientMetricsToFlink(boolean), which neither needs nor stops this export.

      Parameters:
      exportClientMetricsToCloudMonitoring - whether the client exports to Cloud Monitoring
      Returns:
      this builder

    • createDisposition

      public BigtableSinkBuilder<T> createDisposition(CreateDisposition createDisposition)
      Sets whether the sink may create the destination table when a mutation finds it — or one of its column families — missing. Defaults to CreateDisposition.CREATE_NEVER, under which the table and its families must exist.

      CreateDisposition.CREATE_IF_NEEDED requires tableCreateOptions(TableCreateOptions): a Bigtable table's schema is its column families and their garbage-collection policies, which the sink cannot guess.

      Parameters:
      createDisposition - the disposition
      Returns:
      this builder

    • tableCreateOptions

      public BigtableSinkBuilder<T> tableCreateOptions(TableCreateOptions tableCreateOptions)
      Sets column family names, value types and optional garbage-collection rules for CreateDisposition.CREATE_IF_NEEDED. Missing declared families are added and existing GC rules remain unchanged. Options containing aggregate types require every existing declared type to match and inspect them before a destination first receives data, requiring bigtable.tables.get. Raw-only options retain name-only reconciliation of existing families; see TableCreateOptions.
      Parameters:
      tableCreateOptions - the creation settings
      Returns:
      this builder

    • build

      public org.apache.flink.api.connector.sink2.Sink<T> build()
      Builds the sink.
      Returns:
      the sink
      Throws:
      IllegalStateException - if the destination, serializer or required staged options were not set, if the create disposition and the table-creation options disagree, or if a service-account key file was combined with an emulator endpoint
      IllegalArgumentException - if the staged configuration is invalid, including a missing application profile or incompatible creation/failure-handler settings