pnowojski commented on issue #8811: [FLINK-12777][network] Support CheckpointBarrierHandler in StreamTwoInputSelectableProcessor URL: https://github.com/apache/flink/pull/8811#issuecomment-507169781 Thanks for the review @StephanEwen. > Is this very much tailored towards two inputs, when it would be good to keep it generic enough for N inputs (think side inputs in the future). I was considering making it more general, but currently we don't have general `NInputStreamTask` or `NInputStreamProcessor` partially for the performance reasons. So: 1. I wanted to be consistent with existing `TwoInput***` classes 2. I didn't want to risk spending time investigating & make sure that performance of `n input` storage would be good enough 3. There is also a drawback of keeping a more general classes/interfaces when they are never used. It confuses code reader, making potential future refactoring/bug fixing/expanding/maintaining more expensive (a good example was `int[] ChannelSelector#selectChannel`) Because of those reasons I thought that it's better to keep it specialised now and maybe revisit this in the future. > Could the barrier handler (or the specific barrier aligner implementation) not trigger the roll over of the buffer storage? This something that I was also considering. The problem is that buffer storage can not be hidden from `CheckpointedInputGate` inside `BarrierHandler` class, because buffers must be stored independently for two inputs (in order to make input selection work). Once this out of the question, the only alternative solution would be to keep `BufferStorage` instance in the `CheckpointedInputGate`, but "automatically" roll over it from `BarrierHandler`. This would unfortunately add another dependency to the `BarrierHandler` (making it more complicated & harder to insatiate/test) and would also add some implicit side effects. `BufferStorage` would be implicitly modified by actions manipulating `BarrierHandler`, which is I think worse compared to the current setup. All in all, I don't like the current setup too much, but I think it's a lesser evil and at least `BarrierHandler` and `BufferStorage` classes are mostly independent of one another. > `InputGate.pollNext()` throws an InterruptedException even though it is a non-blocking operation. I think I will add a java doc above this method: ``` // Please increase the counter if you spent your time investigating // why InterruptedException might be thrown and realised that it can not be removed // // total_people_count = 4 ``` @StephanEwen + @StefanRRichter + @NicoK + @pnowojski = 4 :) The catch is that this method might block if there is some data, but the buffers pool is empty and we are waiting for some buffer to be recycled (back pressure). Joking aside, I think I will write this down in the java doc. > implementation that throws unsupported operation exceptions Done
---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
