---------------- Moreover one question I have: Is there a plan for the future, that KafkaSource will provide an API for pause/resume consumption (by utilizing the pause and resume methods of underneath kafkaConsumer (apache kafka client)) in order to be able to pause consumption or resume consumption? This is handy for business use-cases, where you want to dictate the KafkaSource consumption based on business predicates.

I would like to understand the business case for that, because at a glance it conflicts with the entire streaming execution mode

Well, let's assume we have 2 flink jobs running, and somehow the one is the master job and the other is the standby-running/slave job - the distributed consensus is provided from another framework, eg: sofa-jraft or something similar. Each flink job has more than one kafkaSource (2 in our case) to consume from. BUT at any point of time, only one should be picked-consumed based on some business logic. So for example: kafkaSource1 kafkaSource2 \ / {bizPred->yes} {bizPred->no} \ / so resume so pause So this is where for example, if I could wire a Predicate<Boolean> to the kafkaSource and based on the outcome to kafkaConsumer.pause or resume would be handy, instead of me copying the sources and modifying, and adding that maintainance burden to my back as well. I hope it makes sense, if not, I can PM you as I have signed some NDAs and cannot describe with details the mechanics.

I don't think that's something that should be solved within the Flink job itself, but the bizPred should talk to the Flink REST/K8S operator to either stop a job (and therefore the consumer) or start a new one. That's the entire point of the JobManager being the orchestrator for a Flink application

yes we have used the Flink REST API, but we do not want to stop the whole job, as the mental model is not 100% the job, but more like the task. Is there any opportunity we can stop the task somehow?

I don't think we want that as a generic feature. You can try to hack it in Flink yourself, but I don't think this is a use case that we should support in the framework

well, as I said this is some exotic stuff - not my idea, all those you mentioned are handled, by having restart policy exponential and region failover, and some other custom logic, so recovery is ok, etc. So in few words when the bizPred is not satisfied we fail on purpose and due to region failover the affected tasks are getting restarted as it should (run in new mode). Anyway, I thought from API point of view if we could wire in the KafkaSourceBuilder a predicate as I mentioned above it would help, and maybe it was useful in general, from your experience you do not believe, as is a custom need, right?

I would define this as a custom need, yes