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

Class PubSubSubscriberOptions.Builder

java.lang.Object
io.github.flink.gcp.connector.pubsub.source.PubSubSubscriberOptions.Builder
Enclosing class:
PubSubSubscriberOptions
@Public public static final class PubSubSubscriberOptions.Builder extends Object
Builder for PubSubSubscriberOptions.

  • Method Details

    • flowControlMaxOutstandingElementCount

      public PubSubSubscriberOptions.Builder flowControlMaxOutstandingElementCount(long flowControlMaxOutstandingElementCount)
      Caps the messages one subscriber holds outstanding; the client library stops pulling once the cap is reached. Optional; defaults to the SDK's limit of 1000 messages. Because the source acknowledges only on checkpoint completion, everything received since the last completed checkpoint counts against this cap.
      Parameters:
      flowControlMaxOutstandingElementCount - the outstanding-message limit, positive
      Returns:
      this builder

    • flowControlMaxOutstandingRequestBytes

      public PubSubSubscriberOptions.Builder flowControlMaxOutstandingRequestBytes(long flowControlMaxOutstandingRequestBytes)
      Caps the bytes one subscriber holds outstanding; the client library stops pulling once the cap is reached. Optional; defaults to the SDK's limit of 100 MB. This is the byte-level bound the message-count cap cannot provide.
      Parameters:
      flowControlMaxOutstandingRequestBytes - the outstanding-byte limit, positive
      Returns:
      this builder

    • subscriberBufferMaxMessages

      public PubSubSubscriberOptions.Builder subscriberBufferMaxMessages(long subscriberBufferMaxMessages)
      Sets the hard aggregate cap on messages retained in the subscriber buffers of one source reader. The first delivery that would cross the cap is rejected before entering the acknowledgement tracker or buffer. If every assigned split is paused by watermark alignment, their subscribers are parked; otherwise the source coordinator fails the job so the response remains visible while downstream polling is stopped.

      This cap does not include records already handed to Flink's fetcher queues. Use maxRecordsPerFetch(int) and Flink's source.reader.element.queue.capacity for that separate footprint.

      Parameters:
      subscriberBufferMaxMessages - the reader-wide retained-message cap, positive
      Returns:
      this builder

    • subscriberBufferMaxBytes

      public PubSubSubscriberOptions.Builder subscriberBufferMaxBytes(long subscriberBufferMaxBytes)
      Sets the hard aggregate serialized-byte cap corresponding to subscriberBufferMaxMessages(long). Both caps apply, and the first delivery that would cross either one is rejected.

      Sizes use PubsubMessage.getSerializedSize(), the same unit as the SDK's flow control.

      Parameters:
      subscriberBufferMaxBytes - the reader-wide serialized-byte cap, positive
      Returns:
      this builder

    • pausedSplitBufferMaxMessages

      public PubSubSubscriberOptions.Builder pausedSplitBufferMaxMessages(long pausedSplitBufferMaxMessages)
      Caps how many messages a split paused by watermark alignment may buffer before the reader stops its subscriber, returning the messages to Pub/Sub and opening a fresh subscriber when the split resumes. Optional; defaults to twice the effective flowControlMaxOutstandingElementCount(long) — so 2000 messages when nothing is set.

      The factor is what the lapse itself is worth, which is why the default needs no number of its own. Once maxAckExtensionPeriod(Duration) passes for a message nobody is draining, the client library stops extending its lease and releases its flow-control permit while the reader still holds it — a whole window of permits per expiry wave, so the first wave carries a paused split's buffer past twice the limit (#357). A bound at the limit itself would be crossed by less than that: a message larger than the byte limit is admitted anyway, a dead-letter subscription's delivery-attempt attribute is added after the client reserves, and a redelivery is held beside the copy it supersedes — so it would park healthy splits.

      Set this lower to bound a paused split's memory more tightly, at the cost of parking sooner and so redelivering more; set it higher to tolerate a longer pause before the subscriber is stopped, and lowering it makes the following more likely rather than introducing it: stopping the subscriber returns everything the split has not had acknowledged, including records it already emitted under a checkpoint still in flight, so those are emitted again on resume. That is within the at-least-once contract, but it is duplication on a running job rather than at a restart.

      It is a bound on what is held between checks, not a cap the buffer cannot pass. The reader evaluates it once per fetch, so a burst delivered between two fetches can overshoot it — bounded in turn by what the client library will deliver at once, which is its flow-control window (measured at 104 and 121 buffered against a bound of 60, over two runs of a 50-message window).

      Parameters:
      pausedSplitBufferMaxMessages - the buffered-message cap, positive
      Returns:
      this builder

    • pausedSplitBufferMaxBytes

      public PubSubSubscriberOptions.Builder pausedSplitBufferMaxBytes(long pausedSplitBufferMaxBytes)
      Caps the bytes a split paused by watermark alignment may buffer, as pausedSplitBufferMaxMessages(long) caps the count. Optional; defaults to twice the effective flowControlMaxOutstandingRequestBytes(long) — so 200 MB when nothing is set.

      Both caps apply, and the reader stops the subscriber when either is exceeded, because which one binds depends on message size: with small messages the count limit is reached first and a byte cap alone would let a pause run for days, while with large ones the byte limit binds and a count cap alone would let the buffer reach gigabytes.

      Sizes are the serialized message size, the same unit the client library's flow control counts in.

      Parameters:
      pausedSplitBufferMaxBytes - the buffered-byte cap, positive
      Returns:
      this builder

    • parallelPullCount

      public PubSubSubscriberOptions.Builder parallelPullCount(int parallelPullCount)
      Sets how many streaming-pull connections one subscriber opens. Optional; defaults to the SDK's single connection. More connections raise a single split's throughput at the cost of more gRPC streams.

      Cannot be combined with OrderingMode.PER_KEY; PubSubSourceBuilder.build() rejects the combination and explains why.

      Parameters:
      parallelPullCount - the streaming-pull connection count, positive
      Returns:
      this builder

    • maxAckExtensionPeriod

      public PubSubSubscriberOptions.Builder maxAckExtensionPeriod(Duration maxAckExtensionPeriod)
      Sets how long the client library keeps extending a message's acknowledgement deadline before giving up on it. Optional; defaults to the SDK's 1 hour.

      This is the ceiling on how long a message may wait for the checkpoint that acknowledges it: once the budget is spent the lease expires and Pub/Sub redelivers, so it must stay comfortably above the checkpoint interval.

      Duration.ZERO is settable and means what the client library means by it: "a zero duration effectively disables auto deadline extensions", so every message's lease expires at the subscription's own acknowledgement deadline. A setting this connector forwards to the SDK stays settable as the SDK defines it (ADR-0068).

      Only a negative budget is refused. Unlike the retry* knobs this one takes no millisecond floor, and that is measured rather than assumed: MessageDispatcher spends it as now().plus(maxAckExtensionPeriod), an instant at nanosecond resolution with no toMillis() anywhere, so a sub-millisecond value is a very short budget rather than a zero in disguise — and a floor whose message promised millisecond granularity would be promising something untrue here.

      Parameters:
      maxAckExtensionPeriod - the total extension budget, non-negative
      Returns:
      this builder

    • minDurationPerAckExtension

      public PubSubSubscriberOptions.Builder minDurationPerAckExtension(Duration minDurationPerAckExtension)
      Sets the smallest deadline extension the client library requests at a time. Optional; defaults to the SDK's adaptive choice, which derives the extension from observed acknowledgement latencies.
      Parameters:
      minDurationPerAckExtension - the smallest single extension, positive and below maxDurationPerAckExtension(Duration)
      Returns:
      this builder

    • maxDurationPerAckExtension

      public PubSubSubscriberOptions.Builder maxDurationPerAckExtension(Duration maxDurationPerAckExtension)
      Sets the largest deadline extension the client library requests at a time. Optional; defaults to the SDK's adaptive choice.
      Parameters:
      maxDurationPerAckExtension - the largest single extension, positive and above minDurationPerAckExtension(Duration)
      Returns:
      this builder

    • awaitAckConfirmation

      public PubSubSubscriberOptions.Builder awaitAckConfirmation(Duration awaitAckConfirmation)
      Makes each completed checkpoint wait for its acknowledgements to be confirmed by the server, failing the job if they are not confirmed within the given time. Optional; defaults to fire-and-forget acknowledgement, which adds no latency.

      This exists because a failed acknowledgement is otherwise invisible. On an ordinary subscription the client library does not retry one — it logs a warning and stops. No data is lost, since an unacknowledged message has its lease expire and is redelivered, which is the at-least-once contract; but a persistent failure such as a revoked permission becomes a silent reprocessing loop.

      The timeout is the only detector. On a subscription without exactly-once delivery the acknowledgement future completes with SUCCESSFUL on success and never completes at all on failure, so there is no error to observe — only the absence of a confirmation. Choose a value comfortably above a normal acknowledgement round trip.

      The wait happens on the task thread when the checkpoint completes, so it delays processing by up to this long. That is the price of the confirmation.

      Parameters:
      awaitAckConfirmation - how long to wait for confirmation, at least 1 ms
      Returns:
      this builder

    • shutdownTimeout

      public PubSubSubscriberOptions.Builder shutdownTimeout(Duration shutdownTimeout)
      Sets how long closing a subscriber waits for it to release its messages. Defaults to 5 s.

      It bounds a reader's whole close, not each split's: the reader nacks every split and asks every client to stop before it waits on any, so the waits overlap however many splits it owns. Keep it under Flink's source.reader.close.timeout (30 s by default), past which a reader is abandoned mid-close and the messages it had not yet released only return after their acknowledgement deadline.

      Parameters:
      shutdownTimeout - the shutdown budget, at least 1 ms and at most Duration.ofNanos(Long.MAX_VALUE)
      Returns:
      this builder

    • maxRecordsPerFetch

      public PubSubSubscriberOptions.Builder maxRecordsPerFetch(int maxRecordsPerFetch)
      Caps how many messages one fetch drains from one split. Defaults to 1000. Bounds how much a single fetch buffers while still amortizing the element-queue handoff; flow control, not this, is the real limit on in-flight messages.
      Parameters:
      maxRecordsPerFetch - the drain size, positive
      Returns:
      this builder

    • firstCheckpointTimeout

      public PubSubSubscriberOptions.Builder firstCheckpointTimeout(Duration firstCheckpointTimeout)
      Sets how long a reader holding unacknowledged messages waits for its first checkpoint before failing the job, measured from the reader's first split assignment. Defaults to 10 min; Duration.ZERO disables the detector.

      The source acknowledges only on checkpoint completion, so a job running without checkpointing never acknowledges anything and stalls silently once flow control fills. Raise this above the checkpoint interval for jobs that checkpoint less often than every 10 min. See MissingCheckpointDetector for why the guard measures elapsed time rather than reading the checkpoint configuration, and why it measures it from the first assignment rather than from the reader's creation.

      Parameters:
      firstCheckpointTimeout - the detector budget, non-negative and at most Duration.ofNanos(Long.MAX_VALUE); zero disables it
      Returns:
      this builder

    • build

      public PubSubSubscriberOptions build()
      Builds the options.
      Returns:
      the options