Audit operator code for "overload protection"

[frankmcsherry]

At least one operator (#301) has an implementation that risks operators getting wedged if they process inputs more slowly than they are produced upstream. Ideally, operators would ensure they perform a finite amount of work each time they are scheduled, but I suspect that many risk continuing to run as long as input data are available.