Using Rqueue with ReadFrom.REPLICA_PREFERRED fails

[tuneyou]

Hello,
When using ReadFrom.REPLICA_PREFERRED, it keeps trying to write to slave server, instead of just reading from it.

[sonus21]

Hi @tuneyou
Can you please provide some more details Rqueue version and Redis driver details?

Rqueue relies on Spring heavily for read/write, need to see what's causing this issue.

Also what's the error you're getting?
Can you please also use Rqueue version 2.0.0 instead of 1.x?

[tuneyou]

Of Course :)
I am using the latest version, 2.0.0.

Lettuce Conf:

@bean
public RedisConnectionFactory redisConnectionFactory() {
RedisStandaloneConfiguration serverConfig = new RedisStandaloneConfiguration(redisMasterHost, redisMasterPort);
serverConfig.setPassword(RedisPassword.of(redisPassword));
LettuceClientConfiguration lettuceClientConfiguration = LettuceClientConfiguration.builder().readFrom(ReadFrom.REPLICA_PREFERRED).build();
return new LettuceConnectionFactory(serverConfig, lettuceClientConfiguration);
}

Redis configuration: Master-slave standalone

Error:

org.springframework.data.redis.RedisSystemException: Error in execution; nested exception is io.lettuce.core.RedisCommandExecutionException: ERR Error running script (call to f_cabaad845bb1fbf876872b4bb21f919f895cd53e): @user_script:14: @user_script: 14: -READONLY You can't write against a read only slave.
at org.springframework.data.redis.connection.lettuce.LettuceExceptionConverter.convert(LettuceExceptionConverter.java:54) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.connection.lettuce.LettuceExceptionConverter.convert(LettuceExceptionConverter.java:52) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.connection.lettuce.LettuceExceptionConverter.convert(LettuceExceptionConverter.java:41) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.PassThroughExceptionTranslationStrategy.translate(PassThroughExceptionTranslationStrategy.java:44) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.FallbackExceptionTranslationStrategy.translate(FallbackExceptionTranslationStrategy.java:42) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.connection.lettuce.LettuceConnection.convertLettuceAccessException(LettuceConnection.java:270) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.connection.lettuce.LettuceScriptingCommands.convertLettuceAccessException(LettuceScriptingCommands.java:236) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.connection.lettuce.LettuceScriptingCommands.evalSha(LettuceScriptingCommands.java:195) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.connection.DefaultedRedisConnection.evalSha(DefaultedRedisConnection.java:1440) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:na]
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
at java.base/java.lang.reflect.Method.invoke(Method.java:566) ~[na:na]
at org.springframework.data.redis.core.CloseSuppressingInvocationHandler.invoke(CloseSuppressingInvocationHandler.java:61) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at com.sun.proxy.$Proxy148.evalSha(Unknown Source) ~[na:na]
at org.springframework.data.redis.core.script.DefaultScriptExecutor.eval(DefaultScriptExecutor.java:77) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.core.script.DefaultScriptExecutor.lambda$execute$0(DefaultScriptExecutor.java:68) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:228) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:188) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:175) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.core.script.DefaultScriptExecutor.execute(DefaultScriptExecutor.java:58) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at org.springframework.data.redis.core.script.DefaultScriptExecutor.execute(DefaultScriptExecutor.java:52) ~[spring-data-redis-2.2.6.RELEASE.jar!/:2.2.6.RELEASE]
at com.github.sonus21.rqueue.core.RqueueMessageTemplateImpl.pop(RqueueMessageTemplateImpl.java:60) ~[rqueue-2.0.0-RELEASE.jar!/:na]
at com.github.sonus21.rqueue.listener.RqueueMessagePoller.getMessage(RqueueMessagePoller.java:55) ~[rqueue-2.0.0-RELEASE.jar!/:na]
at com.github.sonus21.rqueue.listener.RqueueMessagePoller.poll(RqueueMessagePoller.java:109) ~[rqueue-2.0.0-RELEASE.jar!/:na]
at com.github.sonus21.rqueue.listener.DefaultRqueuePoller.start(DefaultRqueuePoller.java:60) ~[rqueue-2.0.0-RELEASE.jar!/:na]
at com.github.sonus21.rqueue.listener.MessageContainerBase.run(MessageContainerBase.java:90) ~[rqueue-2.0.0-RELEASE.jar!/:na]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na]
at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
Caused by: io.lettuce.core.RedisCommandExecutionException: ERR Error running script (call to f_cabaad845bb1fbf876872b4bb21f919f895cd53e): @user_script:14: @user_script: 14: -READONLY You can't write against a read only slave.

[sonus21]

Hmm, I was looking into this. Can you use master preferred? I see it will bring performance penalty the system, this problem is coming due to the underlying code in the spring-data that assumes that script has to be run on slave nodes for some reason, will dig into details why that's the case.

You can use two different redis connection, if you really want to use read preferred for your application, as of now I have to say Rqueue requires Master only.

[tuneyou]

Thank you very much for your prompt responses.
Your dev is great, I've chose it from various options had, also pub sub of redis wich didn't satisfy what I requested (it send message to all the listeners, and not processing only once)
For now, I will do so, as I still work on the functionality of what I need, but on the long term it won't be possible without supporting all redis servers, as this is how the rest of the system works :)

Thank you very much!

[sonus21]

also pub sub of redis wich didn't satisfy what I requested (it send message to all the listeners, and not processing only once)

Can you provide more details about this? I would like to see what's causing this to happen, if visibilityTimeout is too small then only this is possible, in that case you should increase visibility timeout.

On the other note, Rqueue very rarely reads from, due to atomic operation, most of the times it reads and writes in the same call.

[tuneyou]

BTW,
I've tried using multiple connections, and defining on Rqueue only master, but from some reason (which I don't have time to spend on :) ) it wasn't accepted by spring and build failed.

also pub sub of redis wich didn't satisfy what I requested (it send message to all the listeners, and not processing only once)

Can you provide more details about this? I would like to see what's causing this to happen, if visibilityTimeout is too small then only this is possible, in that case you should increase visibility timeout.

On the other note, Rqueue very rarely reads from, due to atomic operation, most of the times it reads and writes in the same call.

My need is that each request will be handled only once, it does happen when using only 1 server, but on N servers, there are N consumers for same message, and according to redis logic, it sends the message to everyone, thus, this solution wasn't good for me.
I didn't want to build new logic for it, although I can use also kafka, but this will be overkill, and handling more systems.
Your solution, Rqueue, looks perfect for me, it is easy to implement, and using existing service (redis)
If I understood correctly, it is also persistent, right?

[sonus21]

Your solution, Rqueue, looks perfect for me, it is easy to implement, and using existing service (redis)
If I understood correctly, it is also persistent, right?

As long as your Redis is alive, data will be there, it won't be lost. If Redis dies and you didn't have enable disk persistent data would be lost. So it relies on the Redis, if data persist there then it will be able to handle all fallbacks.

Also I didn't understand one of your point, if you've cluster setup then you can write to only master nodes not to all. So there's no chance of utilising entire Redis setup.

[tuneyou]

Yes, it automatically writes only to master, but prefer reading from slave.
But when using Rqueue, it doesn't detect this definition, and tries to write to slave.
All other redis writes are working as expected, writing to master, and reading from replica as long as it is available.

[sonus21]

There seems to be some issue, look at this ticket.

https://jira.spring.io/browse/DATAREDIS-908

[tuneyou]

Interesting... what do you suggest besides putting MATER PREFERRED.
Do you have any other idea?

[sonus21]

I would suggest to use MASTER_PREFERRED and provide an additional connection factory to Rqueue to deal with this problem. Other part of the applications can continue to use REPLICA_PREFERRED.

I don't have any other workaround apart from going back doing some transaction stuff and changing again entire code. That would also require master connection. From the number of read/writes perspective, Rqueue does almost same number of read and writes, so I'm not gaining much thing even by providing REPLICA_PREFERRED.

[tuneyou]

I've tried this, but the code wasn't compiled.
I use Rqueue with this way:

@Bean
    public SimpleRqueueListenerContainerFactory simpleRqueueListenerContainerFactory(RqueueMessageHandler rqueueMessageHandler){
        SimpleRqueueListenerContainerFactory factory = new SimpleRqueueListenerContainerFactory();
        factory.setRedisConnectionFactory(redisConnectionFactory());
        factory.setMaxNumWorkers(10);
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setThreadNamePrefix("taskExecutor");
        threadPoolTaskExecutor.setCorePoolSize(10);
        threadPoolTaskExecutor.setMaxPoolSize(50);
        threadPoolTaskExecutor.setQueueCapacity(10000);
        threadPoolTaskExecutor.afterPropertiesSet();
        factory.setTaskExecutor(threadPoolTaskExecutor);
        factory.setRqueueMessageHandler(rqueueMessageHandler);
        return factory;
    }

@Bean
    public RedisConnectionFactory redisConnectionFactory() {
        RedisStandaloneConfiguration serverConfig = new RedisStandaloneConfiguration(redisMasterHost, redisMasterPort);
        serverConfig.setPassword(RedisPassword.of(redisPassword));
        return new LettuceConnectionFactory(serverConfig);
    }

[sonus21]

What's the compilation error? It should compile perfectly.

[tuneyou]

when I change it to this:

@bean
public RedisConnectionFactory redisConnectionFactory() {
RedisStandaloneConfiguration serverConfig = new RedisStandaloneConfiguration(redisMasterHost, redisMasterPort);
serverConfig.setPassword(RedisPassword.of(redisPassword));
LettuceClientConfiguration lettuceClientConfiguration = LettuceClientConfiguration.builder().readFrom(ReadFrom.REPLICA_PREFERRED).build();
return new LettuceConnectionFactory(serverConfig, lettuceClientConfiguration);
}

@Bean
public SimpleRqueueListenerContainerFactory simpleRqueueListenerContainerFactory(RqueueMessageHandler rqueueMessageHandler){
    SimpleRqueueListenerContainerFactory factory = new SimpleRqueueListenerContainerFactory();
    RedisStandaloneConfiguration serverConfig = new RedisStandaloneConfiguration(redisMasterHost, redisMasterPort);
    serverConfig.setPassword(RedisPassword.of(redisPassword));
    factory.setRedisConnectionFactory(new LettuceConnectionFactory(serverConfig));
    factory.setMaxNumWorkers(10);
    ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
    threadPoolTaskExecutor.setThreadNamePrefix("taskExecutor");
    threadPoolTaskExecutor.setCorePoolSize(10);
    threadPoolTaskExecutor.setMaxPoolSize(50);
    threadPoolTaskExecutor.setQueueCapacity(10000);
    threadPoolTaskExecutor.afterPropertiesSet();
    factory.setTaskExecutor(threadPoolTaskExecutor);
    factory.setRqueueMessageHandler(rqueueMessageHandler);
    return factory;
}

I get:

org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'messageService': Unsatisfied dependency expressed through field 'rqueueMessageSender'; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'rqueueMessageSender' defined in class path resource [com/github/sonus21/rqueue/spring/RqueueListenerConfig.class]: Unsatisfied dependency expressed through method 'rqueueMessageSender' parameter 0; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'rqueueMessageTemplate' defined in class path resource [com/github/sonus21/rqueue/spring/RqueueListenerConfig.class]: Unsatisfied dependency expressed through method 'rqueueMessageTemplate' parameter 0; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'rqueueConfig' defined in class path resource [com/github/sonus21/rqueue/spring/RqueueListenerConfig.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.github.sonus21.rqueue.config.RqueueConfig]: Factory method 'rqueueConfig' threw exception; nested exception is java.lang.NullPointerException
at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor$AutowiredFieldElement.inject(AutowiredAnnotationBeanPostProcessor.java:643) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.annotation.InjectionMetadata.inject(InjectionMetadata.java:130) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor.postProcessProperties(AutowiredAnnotationBeanPostProcessor.java:399) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.populateBean(AbstractAutowireCapableBeanFactory.java:1422) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:594) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:517) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:323) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:222) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:321) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:202) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:882) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:878) ~[spring-context-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:550) ~[spring-context-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:141) ~[spring-boot-2.2.6.RELEASE.jar:2.2.6.RELEASE]
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:747) [spring-boot-2.2.6.RELEASE.jar:2.2.6.RELEASE]
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397) [spring-boot-2.2.6.RELEASE.jar:2.2.6.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:315) [spring-boot-2.2.6.RELEASE.jar:2.2.6.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1226) [spring-boot-2.2.6.RELEASE.jar:2.2.6.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1215) [spring-boot-2.2.6.RELEASE.jar:2.2.6.RELEASE]
Caused by: org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'rqueueMessageSender' defined in class path resource [com/github/sonus21/rqueue/spring/RqueueListenerConfig.class]: Unsatisfied dependency expressed through method 'rqueueMessageSender' parameter 0; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'rqueueMessageTemplate' defined in class path resource [com/github/sonus21/rqueue/spring/RqueueListenerConfig.class]: Unsatisfied dependency expressed through method 'rqueueMessageTemplate' parameter 0; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'rqueueConfig' defined in class path resource [com/github/sonus21/rqueue/spring/RqueueListenerConfig.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.github.sonus21.rqueue.config.RqueueConfig]: Factory method 'rqueueConfig' threw exception; nested exception is java.lang.NullPointerException
at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:798) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:539) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1338) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1177) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:557) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:517) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:323) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:222) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:321) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:202) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.config.DependencyDescriptor.resolveCandidate(DependencyDescriptor.java:276) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.DefaultListableBeanFactory.doResolveDependency(DefaultListableBeanFactory.java:1290) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveDependency(DefaultListableBeanFactory.java:1210) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor$AutowiredFieldElement.inject(AutowiredAnnotationBeanPostProcessor.java:640) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
... 19 common frames omitted
Caused by: org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'rqueueMessageTemplate' defined in class path resource [com/github/sonus21/rqueue/spring/RqueueListenerConfig.class]: Unsatisfied dependency expressed through method 'rqueueMessageTemplate' parameter 0; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'rqueueConfig' defined in class path resource [com/github/sonus21/rqueue/spring/RqueueListenerConfig.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.github.sonus21.rqueue.config.RqueueConfig]: Factory method 'rqueueConfig' threw exception; nested exception is java.lang.NullPointerException
at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:798) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:539) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1338) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1177) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:557) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:517) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:323) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:222) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:321) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:202) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.config.DependencyDescriptor.resolveCandidate(DependencyDescriptor.java:276) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.DefaultListableBeanFactory.doResolveDependency(DefaultListableBeanFactory.java:1290) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveDependency(DefaultListableBeanFactory.java:1210) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.ConstructorResolver.resolveAutowiredArgument(ConstructorResolver.java:885) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:789) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
... 32 common frames omitted
Caused by: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'rqueueConfig' defined in class path resource [com/github/sonus21/rqueue/spring/RqueueListenerConfig.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.github.sonus21.rqueue.config.RqueueConfig]: Factory method 'rqueueConfig' threw exception; nested exception is java.lang.NullPointerException
at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:656) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:636) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1338) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1177) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:557) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:517) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:323) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:222) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:321) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:202) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.config.DependencyDescriptor.resolveCandidate(DependencyDescriptor.java:276) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.DefaultListableBeanFactory.doResolveDependency(DefaultListableBeanFactory.java:1290) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveDependency(DefaultListableBeanFactory.java:1210) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.ConstructorResolver.resolveAutowiredArgument(ConstructorResolver.java:885) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:789) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
... 46 common frames omitted
Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.github.sonus21.rqueue.config.RqueueConfig]: Factory method 'rqueueConfig' threw exception; nested exception is java.lang.NullPointerException
at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:185) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:651) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
... 60 common frames omitted
Caused by: java.lang.NullPointerException: null
at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory$SharedConnection.getNativeConnection(LettuceConnectionFactory.java:1197) ~[spring-data-redis-2.2.6.RELEASE.jar:2.2.6.RELEASE]
at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory$SharedConnection.getConnection(LettuceConnectionFactory.java:1178) ~[spring-data-redis-2.2.6.RELEASE.jar:2.2.6.RELEASE]
at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory.getSharedConnection(LettuceConnectionFactory.java:942) ~[spring-data-redis-2.2.6.RELEASE.jar:2.2.6.RELEASE]
at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory.getConnection(LettuceConnectionFactory.java:353) ~[spring-data-redis-2.2.6.RELEASE.jar:2.2.6.RELEASE]
at org.springframework.data.redis.core.RedisConnectionUtils.doGetConnection(RedisConnectionUtils.java:134) ~[spring-data-redis-2.2.6.RELEASE.jar:2.2.6.RELEASE]
at org.springframework.data.redis.core.RedisConnectionUtils.getConnection(RedisConnectionUtils.java:97) ~[spring-data-redis-2.2.6.RELEASE.jar:2.2.6.RELEASE]
at org.springframework.data.redis.core.RedisConnectionUtils.getConnection(RedisConnectionUtils.java:84) ~[spring-data-redis-2.2.6.RELEASE.jar:2.2.6.RELEASE]
at com.github.sonus21.rqueue.utils.RedisUtils.updateAndGetVersion(RedisUtils.java:68) ~[rqueue-2.0.0-RELEASE.jar:na]
at com.github.sonus21.rqueue.config.RqueueListenerBaseConfig.rqueueConfig(RqueueListenerBaseConfig.java:80) ~[rqueue-2.0.0-RELEASE.jar:na]
at com.github.sonus21.rqueue.spring.RqueueListenerConfig$$EnhancerBySpringCGLIB$$bc1ab31.CGLIB$rqueueConfig$15() ~[rqueue-spring-2.0.0-RELEASE.jar:na]
at com.github.sonus21.rqueue.spring.RqueueListenerConfig$$EnhancerBySpringCGLIB$$bc1ab31$$FastClassBySpringCGLIB$$c55cd5ba.invoke() ~[rqueue-spring-2.0.0-RELEASE.jar:na]
at org.springframework.cglib.proxy.MethodProxy.invokeSuper(MethodProxy.java:244) ~[spring-core-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at org.springframework.context.annotation.ConfigurationClassEnhancer$BeanMethodInterceptor.intercept(ConfigurationClassEnhancer.java:331) ~[spring-context-5.2.5.RELEASE.jar:5.2.5.RELEASE]
at com.github.sonus21.rqueue.spring.RqueueListenerConfig$$EnhancerBySpringCGLIB$$bc1ab31.rqueueConfig() ~[rqueue-spring-2.0.0-RELEASE.jar:na]
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:na]
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
at java.base/java.lang.reflect.Method.invoke(Method.java:564) ~[na:na]
at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:154) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
... 61 common frames omitted

[sonus21]

@Bean
public SimpleRqueueListenerContainerFactory simpleRqueueListenerContainerFactory(RqueueMessageHandler rqueueMessageHandler){
    SimpleRqueueListenerContainerFactory factory = new SimpleRqueueListenerContainerFactory();
    RedisStandaloneConfiguration serverConfig = new RedisStandaloneConfiguration(redisMasterHost, redisMasterPort);
    serverConfig.setPassword(RedisPassword.of(redisPassword));
    factory.setRedisConnectionFactory(new LettuceConnectionFactory(serverConfig));
    factory.setMaxNumWorkers(10);
    ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
    threadPoolTaskExecutor.setThreadNamePrefix("taskExecutor");
    threadPoolTaskExecutor.setCorePoolSize(10);
    threadPoolTaskExecutor.setMaxPoolSize(50);
    threadPoolTaskExecutor.setQueueCapacity(10000);
    threadPoolTaskExecutor.afterPropertiesSet();
    factory.setTaskExecutor(threadPoolTaskExecutor);
    factory.setRqueueMessageHandler(rqueueMessageHandler);
    return factory;
}

Lettuce connection factory is not created correctly, you need to call afterPropertiesSet method.
At the same time, if you really want to use redis cluster, then you should provide all redis node details right?

[tuneyou]

Hello Sonus,
How are you?
What does setMaxNumWorkers do, and what's the relations with threads pool size?

Thanks :)

[sonus21]

Hello
I'm doing good, I hope you're doing good too.

What does setMaxNumWorkers do, and what's the relations with threads pool size?

When you're using custom executor then you should set this number as (thread pool max size-number of queue size), given you're not sharing this executor with other executors, if you're sharing the same executor across multiple tasks, then you can set this upto some maximum numbers.

What would happen if I set this to very high value?
Task would be rejected by executor unless queue size is non-zero, when queue size is non-zero then it can create duplicate message consumption problem, since you've already polled but you have not processed the same message.

The maxNumWokers tells how many workers you want to run in parallel for all listeners, for example if you have 3 listeners and you have set this as 10 then all 3 listeners would be running combined 10 jobs.

If you don't set this then it assumes that you want to use default size that's 2 threads per listeners.

[tuneyou]

Currently this is my code:

@bean
public SimpleRqueueListenerContainerFactory simpleRqueueListenerContainerFactory(RqueueMessageHandler rqueueMessageHandler){
SimpleRqueueListenerContainerFactory factory = new SimpleRqueueListenerContainerFactory();
RedisStandaloneConfiguration serverConfig = new RedisStandaloneConfiguration(redisMasterHost, redisMasterPort);
serverConfig.setPassword(RedisPassword.of(redisPassword));
LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(serverConfig);
lettuceConnectionFactory.afterPropertiesSet();
factory.setRedisConnectionFactory(lettuceConnectionFactory);
//factory.setRedisConnectionFactory(redisConnectionFactory());
factory.setMaxNumWorkers(50);
ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
threadPoolTaskExecutor.setThreadNamePrefix("taskExecutor");
threadPoolTaskExecutor.setCorePoolSize(20);
threadPoolTaskExecutor.setMaxPoolSize(100);
threadPoolTaskExecutor.setQueueCapacity(100);
threadPoolTaskExecutor.afterPropertiesSet();
factory.setTaskExecutor(threadPoolTaskExecutor);
factory.setRqueueMessageHandler(rqueueMessageHandler);
return factory;
}

I still didn't understand what's the relation between those paramters.
It looks like only setMaxNumWorkers affects the number of running threads.
Can you try to explain bit more about this?
My goal is to use as much as needed threads, as long as the machine will be able to handle this of course.

Thanks :)

[sonus21]

factory.setMaxNumWorkers(50);

This tells how many workers should be running at any point of time, so you could be running 50 threads at some point of times.

threadPoolTaskExecutor.setCorePoolSize(20);

This tells that we should always have 20 threads ready for workers.

threadPoolTaskExecutor.setMaxPoolSize(100);

This tells that we can create maximum 100 threads, and as the execution completed they are recycled for re-usage and some of them would be discarded to reach CorePoolSize

threadPoolTaskExecutor.setQueueCapacity(100);

This number says even though we have 100 runnings threads we can have 100 task waiting in the queue, that can be dequeued and executed by the existing thread as soon as they complete the task.

Now coming back to factory.setMaxNumWorkers(50); as you have set number of threads to 100 but MaxNumWorkers as 50, only 50 of those threads would be used it will never reach the limit that's 100 here.

How many queues do you have?
Let's say you have N queues then you can set

MaxNumWorkers = 100 (MaxPoolSize) + 100(QueueCapacity) - N

But I have worries, really you should be running 100 threads for your use-case.

[tuneyou]

Perfect, thank you sonus!

[sonus21]

Would you mind, if I can add your web/company name in the new section Used by? You can also raise a PR for the same.

[tuneyou]

Of course you can :)

…

On Fri, May 29, 2020, 16:50 Sonu Kumar ***@***.***> wrote: Would you mind, if I can add your web/company name in the new section Used by? You can also raise a PR for the same. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#19 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AHBAKJV5K7URXJ3G63YX7IDRT64RTANCNFSM4M66IQYA> .

[sonus21]

Thanks, I have added :)

[Harsha-changejar]

Hello Sonus,

I am trying to do the same, but on bootup the application throws the below error.

Caused by: org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'simpleRqueueListenerContainerFactory' defined in class path resource [com/api/changejar/config/redis/RedisStandaloneConfig.class]: Unsatisfied dependency expressed through method 'simpleRqueueListenerContainerFactory' parameter 0; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'rqueueMessageHandler' defined in class path resource [com/github/sonus21/rqueue/spring/boot/RqueueListenerAutoConfig.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.github.sonus21.rqueue.listener.RqueueMessageHandler]: Circular reference involving containing bean 'com.github.sonus21.rqueue.spring.boot.RqueueListenerAutoConfig' - consider declaring the factory method as static for independence from its containing instance. Factory method 'rqueueMessageHandler' threw exception; nested exception is java.lang.NullPointerException

... 57 common frames omitted

Caused by: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'rqueueMessageHandler' defined in class path resource [com/github/sonus21/rqueue/spring/boot/RqueueListenerAutoConfig.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.github.sonus21.rqueue.listener.RqueueMessageHandler]: Circular reference involving containing bean 'com.github.sonus21.rqueue.spring.boot.RqueueListenerAutoConfig' - consider declaring the factory method as static for independence from its containing instance. Factory method 'rqueueMessageHandler' threw exception; nested exception is java.lang.NullPointerException
at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:658) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:486) ~[spring-beans-5.3.3.jar:5.3.3]
Caused by: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'rqueueMessageHandler' defined in class path resource [com/github/sonus21/rqueue/spring/boot/RqueueListenerAutoConfig.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.github.sonus21.rqueue.listener.RqueueMessageHandler]: Circular reference involving containing bean 'com.github.sonus21.rqueue.spring.boot.RqueueListenerAutoConfig' - consider declaring the factory method as static for independence from its containing instance. Factory method 'rqueueMessageHandler' threw exception; nested exception is java.lang.NullPointerException

at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1336) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1179) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:571) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:531) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:335) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:234) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:333) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:208) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.config.DependencyDescriptor.resolveCandidate(DependencyDescriptor.java:276) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.DefaultListableBeanFactory.doResolveDependency(DefaultListableBeanFactory.java:1380) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveDependency(DefaultListableBeanFactory.java:1300) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.ConstructorResolver.resolveAutowiredArgument(ConstructorResolver.java:887) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:791) ~[spring-beans-5.3.3.jar:5.3.3]
... 70 common frames omitted

Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.github.sonus21.rqueue.listener.RqueueMessageHandler]: Circular reference involving containing bean 'com.github.sonus21.rqueue.spring.boot.RqueueListenerAutoConfig' - consider declaring the factory method as static for independence from its containing instance. Factory method 'rqueueMessageHandler' threw exception; nested exception is java.lang.NullPointerException
at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:185) ~[spring-beans-5.3.3.jar:5.3.3]
at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:653) ~[spring-beans-5.3.3.jar:5.3.3]
... 84 common frames omitted
Caused by: java.lang.NullPointerException: null
at java.base/java.util.concurrent.ConcurrentHashMap.putVal(ConcurrentHashMap.java:1011) ~[na:na]
at java.base/java.util.concurrent.ConcurrentHashMap.putIfAbsent(ConcurrentHashMap.java:1541) ~[na:na]
at java.base/java.lang.ClassLoader.getClassLoadingLock(ClassLoader.java:667) ~[na:na]
at java.base/jdk.internal.loader.BuiltinClassLoader.loadClassOrNull(BuiltinClassLoader.java:591) ~[na:na]
at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:579) ~[na:na]
at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178) ~[na:na]
at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521) ~[na:na]
Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.github.sonus21.rqueue.listener.RqueueMessageHandler]: Circular reference involving containing bean 'com.github.sonus21.rqueue.spring.boot.RqueueListenerAutoConfig' - consider declaring the factory method as static for independence from its containing instance. Factory method 'rqueueMessageHandler' threw exception; nested exception is java.lang.NullPointerException

Caused by: java.lang.NullPointerException: null