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

Class SpannerSink

java.lang.Object
io.github.flink.gcp.connector.spanner.sink.SpannerSink
@Public public final class SpannerSink extends Object
Entry point for building a Spanner sink.

The sink applies one mutation per record through batchWriteAtLeastOnce, at-least-once, and flushes everything it holds at each checkpoint barrier. A mutation names its own table, so one sink writes to as many tables of the configured database as its serializer produces; no table is ever created, and a missing one fails the job.

Replay is the serializer's problem to make harmless. Spanner's batch write offers no replay protection, so a mutation may be applied more than once — after a job restart, and also within one attempt when a request whose outcome never arrived is retried. What that costs depends on the operation the serializer built, and all five behave differently:

  • insertOrUpdate and replace — idempotent for that mutation.
  • delete — idempotent, and a delete of a row that is not there is simply applied.
  • insert — the replay is rejected with ALREADY_EXISTS, which is routed to the failure handler as a per-mutation failure.
  • update — idempotent, but if the row was deleted between the two attempts Spanner answers NOT_FOUND, which is not a per-mutation refusal and fails the job.

This per-mutation property is not a same-key ordering guarantee: separate BatchWrite mutation groups may be applied in an unspecified order.

That at-least-once statement assumes the default FailureHandler.failJob() policy. Under a dropping policy configured through SpannerSinkBuilder.failedMutationHandler(FailureHandler), a completed checkpoint means every record up to the barrier was either applied, skipped by the serializer, or handed to that handler.

Example: 


 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();

  • Method Details

    • builder

      public static <T> SpannerSinkBuilder<T> builder()
      Creates a new SpannerSinkBuilder.

      The sink returned by the builder implements LineageVertexProvider and returns an empty physical dataset list. A database does not establish the tables a user serializer will write, and lineage extraction never calls that serializer or opens a client. The Table API sink separately carries its fixed table in the gcp physical-resource facet. Flink 2.x extracts lineage metadata automatically; Flink 1.20 supports direct inspection but not native listener delivery.

      Type Parameters:
      T - type of the records written by the sink
      Returns:
      a new builder