Why doesn't Redis call my MessageListener on __keyevent@*__:expired topic events?

Question

I have recently upgraded Spring-Boot to 2.1.4.RELEASE and Spring-Cloud to Greenwich.SR1. My service is running on Java 11. My only dependency to Redis is through spring-boot-starter-data-redis. Although I did the configuration on Redis by setting notify-keyspace-events Ex, yet I don't seem to be able to receive any key expiry events from it. This is the first time I intend to receive such events for timeout purposes. What could have gone wrong?

Please help!

This is my RedisConfiguration:

@Configuration
public class RedisConfiguration {

    @Value("${spring.redis.port}")
    private String port;

    @Value("${spring.redis.host}")
    private String host;

    @Value("${spring.redis.verification-code-topic}")
    private String verificationCodeTopic;

    @Bean
    public RedisConnectionFactory redisConnectionFactory(){
        RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration();
        configuration.setHostName(host);
        configuration.setPort(Integer.valueOf(port));
        return new LettuceConnectionFactory(configuration);
    }

    @Bean
    @Primary
    public RedisTemplate<FundRedisKey, ResetPasswordRequest> resetPasswordRedisTemplate(){
        RedisTemplate<FundRedisKey, ResetPasswordRequest> redisTemplate = new RedisTemplate();
        redisTemplate.setConnectionFactory(redisConnectionFactory());
        redisTemplate.setValueSerializer(resetPasswordRequestSerializer());
        redisTemplate.setKeySerializer(redisKeySerializer());
        return redisTemplate;
    }

    @Bean
    public RedisTemplate<FundRedisKey, VerificationMessage> verificationMessageRedisTemplate(){
        RedisTemplate<FundRedisKey, VerificationMessage> redisTemplate = new RedisTemplate();
        redisTemplate.setConnectionFactory(redisConnectionFactory());
        redisTemplate.setValueSerializer(verificationMessageSerializer());
        redisTemplate.setKeySerializer(redisKeySerializer());
        return redisTemplate;
    }

    @Bean
    @Primary
    public MessageListener verificationCodeMessageListener(){
        return new VerificationCodeSubscriber(verificationMessageSerializer(),
                resetPasswordRedisTemplate());
    }

    @Bean
    public MessageListener resetPasswordTimeoutListener(){
        return new ResetPasswordTimeoutSubscriber(resetPasswordRequestSerializer());
    }

    @Bean
    @Primary
    public MessageListenerAdapter verificationCodeMessageListenerAdapter(){
        return new MessageListenerAdapter(verificationCodeMessageListener());
    }

    @Bean
    public MessageListenerAdapter resetPasswordTimeoutMessageListenerAdapter(){
        return new MessageListenerAdapter(resetPasswordTimeoutListener());
    }

    @Bean
    public ChannelTopic verificationCodeTopic(){
        return new ChannelTopic(verificationCodeTopic);
    }

    @Bean
    @DependsOn(value = "taskExecutor")
    public RedisMessageListenerContainer fundMessageListenerContainer(
            @Qualifier("taskExecutor")Executor executor){
        RedisMessageListenerContainer messageListenerContainer = new RedisMessageListenerContainer();
        messageListenerContainer.setConnectionFactory(redisConnectionFactory());
        messageListenerContainer.addMessageListener(
                verificationCodeMessageListenerAdapter(), verificationCodeTopic());
        messageListenerContainer.addMessageListener(
                resetPasswordTimeoutMessageListenerAdapter(), new PatternTopic("__keyevent@*__:expired"));
        messageListenerContainer.setTaskExecutor(executor);
        return messageListenerContainer;
    }

    @Bean
    public MessagePublisher verificationCodeMessagePublisher(){
        return new VerificationCodePublisher(
                verificationMessageRedisTemplate(), verificationCodeTopic());

    }

    @Bean
    public RedisSerializer verificationMessageSerializer(){
        return new Jackson2JsonRedisSerializer(VerificationMessage.class);
    }

    @Bean
    @Primary
    public RedisSerializer resetPasswordRequestSerializer(){
        return new Jackson2JsonRedisSerializer(ResetPasswordRequest.class);
    }

    @Bean
    public RedisSerializer redisKeySerializer(){
        return new Jackson2JsonRedisSerializer(FundRedisKey.class);
    }
}

And here is my ResetPasswordTimeoutSubscriber:

@Component
public class ResetPasswordTimeoutSubscriber implements MessageListener {

    @Value("${spring.redis.key}")
    private String key;

    private final RedisSerializer messageSerializer;

    public ResetPasswordTimeoutSubscriber(RedisSerializer messageSerializer){
        this.messageSerializer = messageSerializer;
    }

    @Override
    public void onMessage(Message message, byte[] bytes) {

        ResetPasswordRequest resetPasswordRequest =
                (ResetPasswordRequest)messageSerializer.deserialize(message.getBody());

        //TODO Send operation timeout notification



    }
}

Here is my TaskExecutor configuration

@Configuration
@EnableAsync
@EnableScheduling
public class AsyncConfiguration implements AsyncConfigurer, SchedulingConfigurer {

    private final Logger log = LoggerFactory.getLogger(AsyncConfiguration.class);

    private final JHipsterProperties jHipsterProperties;

    public AsyncConfiguration(JHipsterProperties jHipsterProperties) {
        this.jHipsterProperties = jHipsterProperties;
    }

    @Override
    @Bean(name = "taskExecutor")
    public Executor getAsyncExecutor() {
        log.debug("Creating Async Task Executor");
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(jHipsterProperties.getAsync().getCorePoolSize());
        executor.setMaxPoolSize(jHipsterProperties.getAsync().getMaxPoolSize());
        executor.setQueueCapacity(jHipsterProperties.getAsync().getQueueCapacity());
        executor.setThreadNamePrefix("app-1-Executor-");
        return new ExceptionHandlingAsyncTaskExecutor(executor);
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new SimpleAsyncUncaughtExceptionHandler();
    }

    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        taskRegistrar.setScheduler(scheduledTaskExecutor());
    }

    @Bean
    public ScheduledExecutorService scheduledTaskExecutor() {
        return Executors.newScheduledThreadPool(jHipsterProperties.getAsync().getCorePoolSize());
    }
}

Answer 1

I had not put an @Bean on my redisKeySerializer. I had posted the correct answer as question body.