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

Class PubSubPublisherOptions.Builder

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

  • Method Details

    • batchElementCountThreshold

      public PubSubPublisherOptions.Builder batchElementCountThreshold(long batchElementCountThreshold)
      Sets how many messages a publisher batches into one publish request. Optional; defaults to the SDK's threshold.
      Parameters:
      batchElementCountThreshold - the element-count threshold, positive
      Returns:
      this builder

    • batchRequestByteThreshold

      public PubSubPublisherOptions.Builder batchRequestByteThreshold(long batchRequestByteThreshold)
      Sets how many bytes a publisher batches into one publish request. Optional; defaults to the SDK's threshold.
      Parameters:
      batchRequestByteThreshold - the request-byte threshold, positive
      Returns:
      this builder

    • batchDelayThreshold

      public PubSubPublisherOptions.Builder batchDelayThreshold(Duration batchDelayThreshold)
      Sets how long a publisher waits for a batch to fill before sending it. Optional; defaults to the SDK's threshold.
      Parameters:
      batchDelayThreshold - the delay threshold, positive
      Returns:
      this builder

    • retryTotalTimeout

      public PubSubPublisherOptions.Builder retryTotalTimeout(Duration retryTotalTimeout)
      Sets the total time budget of a publish including its retries. Optional; defaults to the SDK's timeout.

      Cannot be combined with enableMessageOrdering(true), which build() rejects: an ordering-enabled SDK publisher replaces this and retryMaxAttempts(int) with an effectively infinite budget, so setting either would promise a bound the publisher does not have.

      Duration.ZERO is settable and means what gax means by it: retries are bounded by the attempt count instead of by time. A setting this connector forwards to the SDK stays settable as the SDK defines it; a positive sub-millisecond value is refused instead, because gax reads this with toMillis() and it would silently become that zero (ADR-0068).

      Parameters:
      retryTotalTimeout - the total timeout, at least 1 ms or Duration.ZERO
      Returns:
      this builder

    • retryInitialDelay

      public PubSubPublisherOptions.Builder retryInitialDelay(Duration retryInitialDelay)
      Sets the delay before the first publish retry. Optional; defaults to the SDK's delay.

      Duration.ZERO is settable and means what gax means by it: no delay before the first retry, which is gax's own default. A setting this connector forwards to the SDK stays settable as the SDK defines it; a positive sub-millisecond value is refused instead, because gax reads this with toMillis() and it would silently become that zero (ADR-0068).

      Parameters:
      retryInitialDelay - the initial retry delay, at least 1 ms or Duration.ZERO
      Returns:
      this builder

    • retryDelayMultiplier

      public PubSubPublisherOptions.Builder retryDelayMultiplier(double retryDelayMultiplier)
      Sets the factor the retry delay grows by per attempt. Optional; defaults to the SDK's multiplier.
      Parameters:
      retryDelayMultiplier - the delay multiplier, at least 1.0
      Returns:
      this builder

    • retryMaxDelay

      public PubSubPublisherOptions.Builder retryMaxDelay(Duration retryMaxDelay)
      Caps the delay between publish retries. Optional; defaults to the SDK's cap.

      Duration.ZERO is settable and means what gax means by it: a cap of zero, which clamps every retry delay to none. A setting this connector forwards to the SDK stays settable as the SDK defines it; a positive sub-millisecond value is refused instead, because gax reads this with toMillis() and it would silently become that zero (ADR-0068).

      Parameters:
      retryMaxDelay - the maximum retry delay, at least 1 ms or Duration.ZERO
      Returns:
      this builder

    • retryInitialRpcTimeout

      public PubSubPublisherOptions.Builder retryInitialRpcTimeout(Duration retryInitialRpcTimeout)
      Sets the timeout of the first publish RPC attempt. Optional; defaults to the SDK's timeout.

      Duration.ZERO is settable and means what gax means by it: the call runs indefinitely, until the connection itself ends. A setting this connector forwards to the SDK stays settable as the SDK defines it; a positive sub-millisecond value is refused instead, because gax reads this with toMillis() and it would silently become that zero (ADR-0068).

      Parameters:
      retryInitialRpcTimeout - the initial per-RPC timeout, at least 1 ms or Duration.ZERO
      Returns:
      this builder

    • retryRpcTimeoutMultiplier

      public PubSubPublisherOptions.Builder retryRpcTimeoutMultiplier(double retryRpcTimeoutMultiplier)
      Sets the factor the per-RPC timeout grows by per attempt. Optional; defaults to the SDK's multiplier.
      Parameters:
      retryRpcTimeoutMultiplier - the timeout multiplier, at least 1.0
      Returns:
      this builder

    • retryMaxRpcTimeout

      public PubSubPublisherOptions.Builder retryMaxRpcTimeout(Duration retryMaxRpcTimeout)
      Caps the timeout of a publish RPC attempt. Optional; defaults to the SDK's cap.

      Duration.ZERO is settable and means what gax means by it: a cap of zero, which lets every call run indefinitely. A setting this connector forwards to the SDK stays settable as the SDK defines it; a positive sub-millisecond value is refused instead, because gax reads this with toMillis() and it would silently become that zero (ADR-0068).

      Parameters:
      retryMaxRpcTimeout - the maximum per-RPC timeout, at least 1 ms or Duration.ZERO
      Returns:
      this builder

    • retryMaxAttempts

      public PubSubPublisherOptions.Builder retryMaxAttempts(int retryMaxAttempts)
      Caps the publish attempts. Optional; defaults to the SDK's behavior of bounding retries only by the total timeout (0 means the same).

      Cannot be combined with enableMessageOrdering(true); see retryTotalTimeout(Duration).

      Parameters:
      retryMaxAttempts - the maximum attempts, non-negative
      Returns:
      this builder

    • enableMessageOrdering

      public PubSubPublisherOptions.Builder enableMessageOrdering(boolean enableMessageOrdering)
      Sets whether publishers honor message ordering keys. Defaults to false; the writer rejects messages carrying an ordering key while this is disabled.

      Enabling it costs the publish retry budget: build() rejects an explicit retryTotalTimeout(Duration) or retryMaxAttempts(int) beside it, because the SDK publisher would replace both.

      Parameters:
      enableMessageOrdering - whether to enable message ordering
      Returns:
      this builder

    • maxInFlightMessages

      public PubSubPublisherOptions.Builder maxInFlightMessages(int maxInFlightMessages)
      Caps the writer's unacknowledged publishes; a write at the cap yields to the task mailbox until completions bring the count back down. Defaults to 1000.
      Parameters:
      maxInFlightMessages - the in-flight cap, positive
      Returns:
      this builder

    • maxInFlightBytes

      public PubSubPublisherOptions.Builder maxInFlightBytes(long maxInFlightBytes)
      Caps the total PubsubMessage.getSerializedSize() of the writer's unacknowledged publishes, bounding sink memory where the message count cannot: Pub/Sub allows 10 MiB per message, so maxInFlightMessages(int) alone leaves the retained payload unbounded. Defaults to 64 MiB per writer subtask — with the default message cap of 1000 that binds only above ~64 KiB per message, so small-message pipelines keep today's behavior.

      Sizing: this value, times the sink subtasks sharing a TaskManager, must fit that TaskManager's heap budget. The counter measures serialized protobuf size, which under-counts actual JVM retention — leave headroom.

      Like the message cap, a write at the cap yields to the task mailbox rather than blocking the task thread, and it applies with enableMessageOrdering(boolean) enabled. A message larger than the cap is still published rather than rejected, exceeding the cap until it completes. Pass Long.MAX_VALUE to bound by message count only.

      Parameters:
      maxInFlightBytes - the in-flight byte cap, positive
      Returns:
      this builder

    • recoveryInitialBackoff

      public PubSubPublisherOptions.Builder recoveryInitialBackoff(Duration recoveryInitialBackoff)
      Sets the first backoff of the topic auto-creation recovery (republishing after creating a missing topic). Defaults to 500 ms.
      Parameters:
      recoveryInitialBackoff - the first backoff, at least 1 ms
      Returns:
      this builder

    • recoveryMaxBackoff

      public PubSubPublisherOptions.Builder recoveryMaxBackoff(Duration recoveryMaxBackoff)
      Caps the backoff of the topic auto-creation recovery. Defaults to 10 s.
      Parameters:
      recoveryMaxBackoff - the backoff cap, at least 1 ms and at least the initial backoff
      Returns:
      this builder

    • recoveryMaxAttempts

      public PubSubPublisherOptions.Builder recoveryMaxAttempts(int recoveryMaxAttempts)
      Caps the republish attempts of the topic auto-creation recovery. Defaults to 10.
      Parameters:
      recoveryMaxAttempts - the maximum attempts, positive
      Returns:
      this builder

    • publishProgressTimeout

      public PubSubPublisherOptions.Builder publishProgressTimeout(Duration publishProgressTimeout)
      Sets how long the writer may wait with no publish completing before it fails. Defaults to 600 seconds.

      This bounds a stall, not a slow topic: the budget restarts at every completion, so a topic that keeps answering — however slowly, and however long the wait in total — never spends it, while a publisher that has stopped resolving anything at all fails the job once. It covers every publish-completion wait the writer makes on the task thread: the in-flight admission gate in write, and drains at a checkpoint, before a capacity eviction, during failure repair, or during per-message isolation. Which path a stalled sink is parked in depends on the work that first needs a completion.

      Without enableMessageOrdering this rarely fires: a publish gives up at retryTotalTimeout (600 s by default), which fails the job by itself. With ordering the SDK retries a publish without limit, so nothing inside the sink ends an outage but this. Outside it, Flink's own execution.checkpointing.timeout still fails the job at its default — later, and naming nothing about Pub/Sub — but only while execution.checkpointing.tolerable-failed-checkpoints is 0. Raise that and this budget is the only thing left (issue #333).

      Expiry fails the job and drops nothing: the sink is at-least-once and stores nothing in Flink state, so the records behind the unresolved publishes are replayed from the last completed checkpoint. The cost of that is real and is the reason this is a knob: a disturbance that outlasts the budget now restarts the job where the SDK's retries used to absorb it, and a persistent one becomes a restart loop.

      Parameters:
      publishProgressTimeout - the no-progress budget, positive and at most Duration.ofNanos(Long.MAX_VALUE)
      Returns:
      this builder

    • shutdownTimeout

      public PubSubPublisherOptions.Builder shutdownTimeout(Duration shutdownTimeout)
      Sets how long one publisher release waits for shutdown. Defaults to 30 seconds. A capacity eviction can spend this budget in write, an idle eviction in a successful non-terminal flush, and final teardown in close.

      The budget is measured from the moment the writer asks the publisher to shut down, and every publisher selected by one release is asked before any is waited on, so that release costs this once however many publishers it covers. Keep it under Flink's task.cancellation.timeout (180 s by default) for final close; for running eviction it is also the maximum task-thread stall caused by one release.

      If a running-task release gives up while publisher shutdown work or resources are still alive, the writer fails before opening a replacement. The pinned SDK can otherwise strand its shutdown waiter after an ordering-key cascade, and continuing destination churn would accumulate abandoned threads and transport resources despite the active-publisher cap.

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

    • maxActivePublishers

      public PubSubPublisherOptions.Builder maxActivePublishers(int maxActivePublishers)
      Caps the publishers retained by one writer subtask. Defaults to 100. When a new destination reaches the cap, the writer releases the least-recently-used clean publisher; if every publisher still has work, it drains them first so no message or ordering repair is discarded. The replacement is not opened until the released publisher finishes its bounded shutdown; if shutdown overruns, the write fails rather than accumulating abandoned publisher resources.
      Parameters:
      maxActivePublishers - the active publisher cap, positive
      Returns:
      this builder

    • destinationIdleTimeout

      public PubSubPublisherOptions.Builder destinationIdleTimeout(Duration destinationIdleTimeout)
      Sets how long a destination may go unused before a successful non-terminal flush releases its publisher. Defaults to PubSubPublisherOptions.DEFAULT_DESTINATION_IDLE_TIMEOUT. A later record recreates the publisher transparently.
      Parameters:
      destinationIdleTimeout - the idle timeout, positive and at most Duration.ofNanos(Long.MAX_VALUE)
      Returns:
      this builder

    • perDestinationMetrics

      public PubSubPublisherOptions.Builder perDestinationMetrics(boolean perDestinationMetrics)
      Registers per-topic recordsSend and sendErrors counters beside the writer's totals. Defaults to false.

      Off by default because Flink cannot unregister a metric: with per-record destinations the topic set is unbounded, so every topic the job ever writes to keeps a row in the metric registry for the lifetime of the task. Switch it on for a sink whose destinations are few and known.

      Parameters:
      perDestinationMetrics - whether to register per-topic counters
      Returns:
      this builder

    • maxConsecutiveRejections

      public PubSubPublisherOptions.Builder maxConsecutiveRejections(int maxConsecutiveRejections)
      Sets how many consecutive confirmed rejections fail the job. Defaults to 100; PubSubPublisherOptions.UNBOUNDED (-1) never fails it.

      This bound only matters beside a dropping failedMessageHandler — under the default failJob() the first confirmed rejection fails the job anyway. A dropping policy is a decision to keep running through anomalous records, and the repair's isolation pass pays one solo publish per rejection to isolate each from the good messages batched with it. When every message is being refused, that is no longer a stream with anomalies but a broken pipeline degraded to unbatched publishes under a green job — so once this many confirmed rejections arrive in a row, with not one successfully published message between them, the job fails with a message naming this option. Any successful publish resets the count: an occasional bad record can never accumulate into a failure, however long the job runs.

      Only rejections the isolation pass has confirmed solo count; records the serializer rejects do not, since they say nothing about the service's view of the stream. The count is per writer subtask, across its destinations — a success on any topic resets it.

      Parameters:
      maxConsecutiveRejections - the bound, positive or PubSubPublisherOptions.UNBOUNDED
      Returns:
      this builder

    • build

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