Hi All, after finishing my pull request that should fix the problems with the synchronisation of checkpoints and element emission (the reason for the faulty results of the exactly-once tests) I discovered that the Kafka source does not deal well with being interrupted. We recently changed the SourceFunction to the reachedEnd()/next() interface, with the contract that the source must be interruptible to be able to perform checkpoints. Now this doesn't seem to work with Kafka. I added another Source interface in my PR (https://github.com/apache/flink/pull/742). This is similar to the old interface of run()/cancel(), with the addition that the source must acquire a lock before updating state and emitting elements. The update of state and the emission of elements must happen in the same synchronized block to ensure consistency. This seems to solve the problem but now we have two source interfaces.
The question is now. What do you think about the two interfaces? Should we keep both? Remove one? Cheers, Aljoscha
