Page History

...

c) The implementation of this alternative approach will likely be harder to maintain. Note that users might want to stop the job based on the offset, de-serialized message, or both. The offset-based criteria should ideally be evaluated before records are de-serialized for performance-reason; and the criteria based on the de-serialized record should be evaluated after the record is de-serialized. Thus these criteria ideally should be evaluated at difference position in the code path. It could be inconvenient to achieve this goal while having these logics in the same class.

2) Let user specify eofRecordEvaluator via StreamExecutionEnvironment::fromSource(...).withEofRecordEvaluator(...)

The advantage of this approach that the feature can be used by all connectors without we having to change the implementation (e.g. KafkaSourceBuilder, PulsarSourceBuilder) of individual connectors.

The disadvantage of this approach is that it requires users to pass some source configuration via StreamExecutionEnvironment::fromSource(...) and some other source configuration via e.g. KafkaSourceBuilder(...). This might create a sense of inconsistency/confusion.

Given that the number of connector users are much more than the number of connector developers, it is probably better to optimize the user experience in this case.