Package io.github.flink.gcp.connector.testutils

Class Drains

java.lang.Object

io.github.flink.gcp.connector.testutils.Drains

@Internal
public final class Drains
extends Object

Deadline-bounded draining of a running job's collect iterator.

Method Summary

Modifier and Type	Method	Description
static <T> List<T>	drainDistinct(org.apache.flink.util.CloseableIterator<T> iterator, int count, Duration timeout, Function<? super T,?> distinguisher)	Drains the iterator until count distinct elements have arrived, the job behind it ends, or timeout passes, returning whatever did arrive in arrival order.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Details

drainDistinct

public static <T> List<T> drainDistinct(org.apache.flink.util.CloseableIterator<T> iterator,
 int count,
 Duration timeout,
 Function<? super T,?> distinguisher)
                                 throws Exception

Drains the iterator until count distinct elements have arrived, the job behind it ends, or timeout passes, returning whatever did arrive in arrival order.

Distinct, and returning a shortfall rather than blocking until the class timeout, for two reasons: the connectors are at-least-once, so a redelivery is legitimate and counting total elements would let one duplicate crowd out an original; and a shortfall must fail the assertion that asked for the elements, with the ones that did arrive in its message, rather than consume the build's whole budget. Callers should therefore assert with containsAll or an in-any-order variant, not an exact multiset.

The iteration runs on its own thread because a deadline consulted between elements is not enough: hasNext() on a collect iterator blocks until an element arrives or the job ends, so a loop over it parks exactly when fewer elements than asked for ever arrive — and JUnit's timeout interrupt is ignored by the blocking iterator. One such shortfall cost a build 38 minutes before this helper existed (issue #150).

The iterator is not closed here: every caller owns it in try-with-resources, and that close is what cancels the job and lets the drain thread (a daemon, so it cannot park the JVM either) unwind after a shortfall.

A job failure surfaces here as it did from a bare hasNext() loop: if the job ends exceptionally before enough elements have arrived, the failure is rethrown as-is.

Parameters:

iterator - the running job's output

count - how many distinct elements to wait for

timeout - how long to wait for them

distinguisher - what makes an element distinct — normally the part a redelivery repeats

Throws:

Exception