new user (e.g. start reading a Pubsub topic associated with that user, or register them in another system, etc). Requirements Overview For every element in a PCollection, the Watch.growthOf(fn) transform will repeatedly invoke a poll function fn: InputT → Set> and emit OutputT's that have not yet been seen into an unbounded PCollection>, at the timestamp they were first seen with. Tempting generalization: also watch for departures from each set, and produce a PCollection>>. But let's not do this: it's impossible to assign event-time timestamps to departures, because we don't have access to the reified "event" that caused an element to be missing from the Set<>: e.g. when watching a glob, we know the creation time of each file, but we don't know the deletion time of files that have been deleted. In some cases we might, but it's outside the scope of this proposal. Termination At some point, different for each InputT, we should stop invoking the poll function. The condition can be internal or external to the poll function itself: The poll function may detect termination in a way specific to the method of polling: A function listing a directory can detect that the directory was marked read-only A function expanding a sharded filepattern XXX-of-YYY can detect that YYY files were already found The same poll function can be used with different "external" termination conditions, specified on the Watch transform itself: Poll each input for at most N seconds Stop polling an input if no new outputs were found in the last N seconds A combination of these Internal termination conditions are part of the body of the poll function (the function returns whether the output is final or not). External termination conditions are state machines whose state is carried over between cycles: Initialized when the InputT arrives React to events "new OutputT discovered" and the implicit passage of time At any moment can be asked "is it time to stop polling yet". In theory such state machines can be combined sequentially (wait for one condition; when it's satisfied, initialize the second and wait for it too) or in parallel (using boolean operators). But for practical purposes it's enough to have a few common conditions, the OR operator, and an API for user-defined ones. Watermarks The poll function produces timestamped elements, and the Watch transform produces an unbounded PCollection, so naturally a runner would be interested in timestamps of future elements in this collection. The poll function should be able to report a watermark on future outputs. E.g., a poll function that expands a sharded filepattern may return the watermark from the timestamp of the last file before a gap in available numbered shards (e.g. if of 10 shards, shards 1, 2, 3, 7 and 8 are available - it should return timestamp of shard 3). Poll frequency We would like to control how long to wait before the next round of polling. Ideally it should be possible to do some kind of exponential backoff, but for a start, a fixed amount configurable at the level of the Watch transform itself should suffice (since it's reasonable to use the same polling function with different frequencies). Proposed API PCollection inputs = …; PCollection> outputs = inputs.apply( Watch.growthOf((InputT input, PollReceiver out) → { … out.put(timestamp, value) … // Returns Watch.PollResult return Watch.outputCanGrow().withWatermark( … ); // or: return Watch.outputIsFinal(); }) .withPollInterval(10 sec) .withTerminationPerInput( // Returns Watch.TerminationStrategy Watch.afterEitherOf( Watch.afterTotalOf(5 min), Watch.afterOutputStableFor(1 min))) .withOutputCoder( … ))