Package io.github.flink.gcp.connector.base.lifecycle

Class BoundedShutdown

java.lang.Object
io.github.flink.gcp.connector.base.lifecycle.BoundedShutdown
All Implemented Interfaces:
AutoCloseable
@Internal public final class BoundedShutdown extends Object implements AutoCloseable
The teardown of one client whose own shutdown cannot be trusted to return: both of its steps on a separate thread, and one deadline that the calling thread's single join is the whole of.

The bound is the point. Written for the google-cloud-pubsub Publisher, whose shutdown() waits on a counter of accepted publishes, uninterruptibly and with no timeout, until it is exactly zero — and which both the Pub/Sub sink's per-topic publishers and PubSubDeadLetterQueue own.

That counter can be left permanently above zero: the failure callback cancels the messages still accumulating in a failed ordering key's un-flushed batch and removes the batch, but decrements only by the size of the batch that was in flight, so those increments are never returned (measured on google-cloud-pubsub 1.152.0; issue #265).

And it can simply take arbitrarily long to reach zero, with nothing defective involved: with enableMessageOrdering the SDK overrides the publisher's retry settings to maxAttempts = Integer.MAX_VALUE and an effectively infinite total timeout, so during an outage the in-flight publishes retry forever and the counter never drains. An ordered sink therefore needs this bound whatever the SDK version, which is why it is not written as a workaround.

A separate thread is therefore the only lever available: the wait cannot be interrupted, and Publisher offers no forcible variant. It is a daemon thread so one that never returns cannot keep a JVM from exiting, and a plain thread rather than an executor because shutdownNow() could not interrupt that wait either — the thread would leak just the same, and the executor would then need a bounded teardown of its own.

The termination wait runs on that thread too, rather than on the calling thread after a successful join, and that placement is load-bearing: gax's BackgroundResourceAggregation.awaitTermination passes the full duration to every resource in turn (its own source carries the TODO subtract time already used up from previous resources), and a publisher nests several — its executor, then the stub's transport channel and watchdog. Awaiting on the calling thread would therefore cost a multiple of the timeout, not the timeout. Here it costs the daemon thread's time and nothing else.

Anything either step throws is captured and rethrown by close() with its own type, because a thread's uncaught exception would otherwise reach only Flink's JVM-wide handler — losing a teardown failure the caller used to report, and, under cluster.uncaught-exception-handling: FAIL, turning it into a TaskManager exit.

The two steps are held as functional values rather than as a client, because the client this was written for cannot be subclassed — Publisher is non-final, but its only constructor is private, which forbids a subclass just as effectively (#324) — so this is the only seam a test can drive.

start() and close() must be called from one thread — the task thread, for both users today. That precondition is what makes thread safe as a plain field, and it is enforced by nothing: two threads racing start() would each see a null thread and run shutdown twice, on two daemon threads. A guard was weighed and left out — the callers are writer teardowns, which Flink runs on the task thread by construction — so a third consumer has to honour it deliberately.

The budget must be expressible in nanoseconds — at most Duration.ofNanos(Long.MAX_VALUE), about 292 years — because nanoseconds are the arithmetic the clock does. The constructor rejects a longer one, rather than leaving Duration.toNanos() to throw ArithmeticException from start(): that would land on a TaskManager during a teardown, where it reaches Flink's teardown path and not a caller's try. Every option setter that feeds a budget here rejects the same value, so a user meets it on the client; this is the backstop for a consumer whose budget is built in code and passes no setter (#334). A budget at that ceiling is a real budget — see remainingNanos(), whose arithmetic looks wrong there and is not.

The threading of the remaining mutable state, stated precisely because the class is shared:

  • deadlineNanos is read by both threads (close() and shutdownAndAwait()), so it is not confined. It is safe as a plain long because its single write happens before Thread.start() and the idempotence guard means it never happens again — publication, not confinement. A write added after the thread starts would be a data race.
  • abandoned is written by the calling thread and read by the shutdown thread with no synchronisation edge between them, so it genuinely needs volatile. terminationIncomplete is written by both threads: the shutdown thread when the client's wait reports live resources, and the calling thread when close() gives up. It is read after close, so it needs volatile too.
  • failure is written by the shutdown thread and read by close() only after thread.isAlive() has returned false, which is itself a happens-before edge (JLS 17.4.5), so a plain field would suffice. It is volatile as belt and braces, since the read is one early-return away from being unordered.

  • Constructor Details

    • BoundedShutdown

      public BoundedShutdown(Runnable shutdown, BoundedShutdown.TerminationWait awaitTermination, String description, @Nullable Runnable release, Duration timeout, LongAdder abandonedCount)
      Creates the teardown. Nothing runs until start() or close().
      Parameters:
      shutdown - the client's own shutdown call, which may never return
      awaitTermination - its bounded wait for the resources behind it
      description - what is being shut down, for the thread name and the give-up warnings. Name the kind as well as the resource, so one client is distinguishable from another in a log line the class itself cannot qualify — the two callers pass "topic my-project/events" and "dead-letter topic my-project/dead-letters"
      release - a resource released in close()'s finally whatever happened, including on the give-up path, or null; the caller's owned transport channel is what this is for. A Runnable rather than an AutoCloseable deliberately: it runs in a finally, where anything it threw would replace the failure being propagated, so this is for a release that does not fail — ManagedChannel.shutdownNow() is the one it was written for. A resource whose release can fail belongs in the caller's own Closers.closeAll(AutoCloseable...) list beside this one instead — the shape PubSubDeadLetterQueue.close() takes.
      timeout - the whole budget, measured from start(); at most Duration.ofNanos(Long.MAX_VALUE), which is checked — a non-positive one is not, and gives up at once, the callers' own setters being where positivity is refused
      abandonedCount - incremented once whenever close() gives up or the client's termination wait reports resources still alive, so the owner can report the residue. Supplied by the caller rather than held here, and that is the design: the count has to outlive the task to be observable at all (measured — a reporter at 10 ms never sees a metric a writer only touches in close()), so it is process-wide wherever it lives. Keeping it here would make one number out of every client this class ever serves, and a metric named for one of them — publisherShutdownsAbandoned — would silently include the rest. The nearest such client is not another connector but the Pub/Sub source, whose subscriber teardown has the same shape. Each owner passing its own keeps the names true by construction

  • Method Details

    • timeout

      @VisibleForTesting public Duration timeout()
      The budget, for a test that checks which one its caller handed over.

    • abandonedCounter

      @VisibleForTesting public LongAdder abandonedCounter()
      The counter this teardown was handed, so a caller's wiring can be asserted by identity rather than by driving a give-up and observing an increment — which is a footrace against the freshly started thread, not a deterministic test. Widened for that reason specifically, per the rule beside timeout()'s justification in the module's detailed guidance.

    • wasIncomplete

      public boolean wasIncomplete()
      Whether shutdown left work or resources alive after its budget.

      Meaningful after close() returns. This is true both when close() stopped waiting for the shutdown thread and when the client's bounded termination wait returned false. A running-task owner uses it to stop before opening a replacement for a client whose resources may still be live; final teardown can keep the existing log-and-report behavior.

    • start

      public void start()
      Starts the client's teardown and the clock, without waiting for either. Idempotent, and deliberately does not restart the clock: a caller owning several clients starts every teardown before it closes any, and a second call resetting the deadline would turn its one timeout back into one per client.

    • close

      public void close() throws Exception
      Specified by:
      close in interface AutoCloseable
      Throws:
      Exception