I have Kafka producer and consumer servers, when I try to send a message I get following exception:

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.mail.sender.service.senders.GmailConfirmationSenderService.confirmationMessageListener(com.mail.sender.dto.request.AccountRequest)]
Bean [com.mail.sender.service.senders.GmailConfirmationSenderService@24787445]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [com.mail.sender.dto.request.AccountRequest] for GenericMessage [payload={"email":"[email protected]","username":"asdasd-mjeesh","confirmationTokenDetails":{"token":"3fd3c1ee-20ec-420b-8ee9-11d22cd7598e","createdAt":[2016,1,25,21,34,55],"expiredAt":[2023,1,8,18,19,6,661473300]}}, headers={kafka_offset=12, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@4008ea0f, kafka_timestampType=CREATE_TIME, kafka_receivedPartitionId=0, kafka_receivedTopic=mail_confirmation_message, kafka_receivedTimestamp=1673193848466, __TypeId__=[B@68a9f1ab, kafka_groupId=account_confirmation_group_id}]

Config for my producer server

@Configuration
public class KafkaProducerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServersUrl;

    public Map<String, Object> producerConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServersUrl);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        props.put(JsonSerializer.TYPE_MAPPINGS, "accountRequest:com.confirmation_token.model.dto.request.outgoing.AccountRequest");
        return props;
    }

    @Bean
    public ProducerFactory<String, AccountRequest> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfig());
    }

    @Bean
    public KafkaTemplate<String, AccountRequest> kafkaTemplate(ProducerFactory<String, AccountRequest> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }
}

How I send the messages

@Service
@Slf4j
@RequiredArgsConstructor
public class EmailConfirmationSenderServiceCommunication implements ConfirmationSender {
    private final KafkaTemplate<String, AccountRequest> kafkaTemplate;

    @Override
    public void sendConfirmationToken(AccountRequest accountRequest) {
        kafkaTemplate.send("mail_confirmation_message", accountRequest);
        log.info("Confirmation token={} has been send", accountRequest);
    }
}

Models on producer side

@Builder
@Data
@AllArgsConstructor
@NoArgsConstructor
public class AccountRequest {
    private String email;
    private String username;
    private ConfirmationTokenDetailsRequest confirmationTokenDetails;
}

@Builder
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ConfirmationTokenDetailsRequest {
    private String token;
    private LocalDateTime createdAt;
    private LocalDateTime expiredAt;
}

Config for my consumer server

@Configuration
public class KafkaConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServerUrl;

    public Map<String, Object> consumerConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServerUrl);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(JsonDeserializer.TYPE_MAPPINGS, "accountRequest:com.mail.sender.dto.request.AccountRequest");
        return props;
    }

    @Bean
    public ConsumerFactory<String, AccountRequest> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfig());
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, AccountRequest>> listenerContainerFactory(
            ConsumerFactory<String, AccountRequest> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, AccountRequest> listenerContainerFactory =
                new ConcurrentKafkaListenerContainerFactory<>();
        listenerContainerFactory.setConsumerFactory(consumerFactory);
        return listenerContainerFactory;
    }
}

Config for topic

@Configuration
public class KafkaTopicConfig {

    @Value("${kafka.topic.names.account.confirmation}")
    private String accountMailConfirmationTopicName;

    @Bean
    public NewTopic accountConfirmationTopic() {
        return TopicBuilder
                .name(accountMailConfirmationTopicName)
                .build();
    }
}

How I try to listen the messages

@Service
@Slf4j
@RequiredArgsConstructor
public class GmailConfirmationSenderService implements EmailSender<AccountRequest> {
    private final String accountConfirmationTemplate;
    private final ApplicationModelValidator applicationModelValidator;
    private final JavaMailSender mailSender;

    @Value("${confirmation.link.template}")
    private String confirmationLink;

    @KafkaListener(topics = "mail_confirmation_message", groupId = "account_confirmation_group_id")
    public void confirmationMessageListener(AccountRequest accountRequest) {
        String validationViolations = applicationModelValidator.validate(accountRequest);
        if (!validationViolations.isBlank()) {
            log.error(validationViolations);
            throw new ModelValidationException(validationViolations);
        }

        String userConfirmationLink = String.format(
                accountConfirmationTemplate,
                accountRequest.getUsername(),
                confirmationLink   accountRequest.getConfirmationTokenDetails().getToken());
        this.send(accountRequest, userConfirmationLink);
    }
    
    ...
}

Models on consumer side

@Builder
@Data
@AllArgsConstructor
@NoArgsConstructor
public class AccountRequest {
    @Email
    private String email;

    @Length(min = 8, max = 40)
    private String username;

    @NotNull
    @Valid
    private ConfirmationTokenDetailsRequest confirmationTokenDetails;
}

@Builder
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ConfirmationTokenDetailsRequest {
    @NotBlank
    private String token;

    @PastOrPresent
    private LocalDateTime createdAt;

    @Future
    private LocalDateTime expiredAt;
}

*In the consumer's models are annotations for validation, but they don't affect the process

CodePudding user response：

Your listener is using the default (String) deserializer (probably from Boot's auto configured factory).

listenerContainerFactory - since you are using a non-standard name for the factory, you need to specify it on the @KafkaListener (or change the factory bean name to kafkaListenerContainerFactory).