In FLINK-21132 we disable endInput calls when stopping with savepoint. However as discussed in FLINK-21133, stop with savepoint with drain (stop-with-savepoint --drain), should be calling endOfInput().
is caused by
FLINK-21132BoundedOneInput.endInput is called when taking synchronous savepoint
FLINK-21133FLIP-27 Source does not work with synchronous savepoint