Can Spring-kafka support dynamic Pause and resume with inject some condition

[xzhan]

User case:
consume server poll data from Kafka and write data into Persistent level, but while consumer failure with Persistent level know issue (network/disk/maintain) how can we dynamic Pause it ? and while it recovery , how can we base on some condition do Resume?

another User Case:

on the peak time , we have one APPlication it have two ability

• support Rest API
• support consume
  so on Peak time we want to pause consume to not impact Rest API service , after Peak time we want dynamic Resume our service

Can we do it well with Spring-Kafka? and I have read the doc https://docs.spring.io/spring-kafka/reference/html/#pause-resume

so i am thinking how can we do more integration with it

@garyrussell @artembilan

[artembilan]

I'm not sure what is on so dynamic with those registry.getListenerContainer("pause.resume").pause(); and registry.getListenerContainer("pause.resume").resume(); calls from your code?

Your condition is only valid for your use-case: the other projects might rely on a logic which far away from just plain condition.
Therefore it is fully legit to have a generic pause() and resume() API on the listener container for any possible implementations in the target projects.

Please, elaborate what else could be done in the framework to make this feature easier?

[xzhan]

Please, elaborate what else could be done in the framework to make this feature easier?

--
@artembilan my point is can we define interface for Ratelimit then like this
@component
public class KafkaManager {

private final Ratelimit ratelimit;
private final KafkaListenerEndpointRegistry registry;

public KafkaManager(Ratelimit ratelimit, KafkaListenerEndpointRegistry registry) {
    this.ratelimit = ratelimit;
    this.registry = registry;
}

public void pause(String containerId) {
    if(ratelimit.isNeedRatelimit()) {
        registry.getListenerContainer(containerId).pause();
    }
}

public void resume(String containerId) {
    if(!ratelimit.isNeedRatelimit()) {
        registry.getListenerContainer(containerId).resume();
    }
}

}

can we do it on spring-framework?

[artembilan]

Probably no: this doesn't feet either way to the framework and feels like some custom feature which simply could be implemented in the target project according to its requirements.
Plus your pause/resume API doesn't seem reasonable seeing its logic based on the internal condition and still call with know container id from the outside.
Your current code does not bring too much benefit over that registry.getListenerContainer(containerId).pause()/resume() API which simply can be used from your conditions as we stated in the beginning of this discussion.

See Resilience4j RateLimiter for your conditional use-case.

[xzhan]

-cc @artembilan and @garyrussell what's your idea?

[artembilan]

See my reply on your other comment: -1 from me for this request to be implemented somehow in the framework.
We have already exposed good enough API for pause/resume which simply can be used from whatever custom condition may be implemented in the target project.

[xzhan]

thx @artembilan , I agree with u , that's target project logical. I am draft one design base on each time poll records max size to do some rich logical to Pause/resume.That would be more nature for peak time concept on Consumer side .Will share it to u if u interesting on it