The Email Formatter consumer is responsible for listening for messages coming from the Apache Kafka topic and doing the actual formatting and sending of emails. Consider the following code:

@SpringBootApplication
public class SpringBootEmailFormatter {

    private static final Logger LOGGER = LoggerFactory.getLogger(SpringBootEmailFormatterConsumerApplication.class);

    @Autowired
    private EmailSenderService emailSenderService;

    @KafkaListener(id="resetPasswordRequests", topics = "resetPasswordRequests")
    public String listen(String in) {
        try {
            ResetPasswordRequest resetPasswordRequest = objectMapper().readValue(in, ResetPasswordRequest.class);
            emailSenderService.sendResetPasswordEmail(resetPasswordRequest);
        } catch (IOException e) {
            LOGGER.error("Error while sending Reset Password Email", e);
        }
        return in;
    }

    @Bean
    public NewTopic resetPasswordRequests() {
        return new NewTopic("resetPasswordRequests", 10, (short) 2);
    }

}

In the preceding configuration, one Apache Kafka topic by the name of resetPasswordRequests is created as a Spring Bean with a number of replications 10 and replication factor of 2. The listen method is responsible for consuming requests sent by the producer via the Apache Kafka topic resetPasswordRequests. This method is annotated with @KafkaListener, which points to the resetPasswordRequests topic.

Furthermore, the following EmailSenderService is used to format and send actual emails:

@Service
public class EmailSenderService {

    private static final Logger LOGGER = LoggerFactory.getLogger(EmailTemplateService.class);

    public static final String RESET_PASSWORD_EN_HTML = "reset_password_en.html";

    private final EmailTemplateService emailTemplateService;
    private final JavaMailSender javaMailSender;

    public EmailSenderService(EmailTemplateService emailTemplateService, JavaMailSender javaMailSender) {
        this.emailTemplateService = emailTemplateService;
        this.javaMailSender = javaMailSender;
    }

    public boolean sendResetPasswordEmail(ResetPasswordRequest resetPasswordRequest) {
        boolean sent = false;
        try {
            Map<String, Object> data = new LinkedHashMap<>();
            data.put("USERNAME", resetPasswordRequest.getUsername());
            data.put("NEW_PASSWORD", resetPasswordRequest.getNewPassword());

            String resetPasswordEmailContent = emailTemplateService.getResetPasswordEmail(data, RESET_PASSWORD_EN_HTML);

            MimeMessage mimeMessage = javaMailSender.createMimeMessage();
            MimeMessageHelper mimeMessageHelper = new MimeMessageHelper(mimeMessage);

            mimeMessageHelper.setTo(resetPasswordRequest.getEmail());
            mimeMessageHelper.setText(resetPasswordEmailContent, true);
            mimeMessageHelper.setSubject("Password Reset");

            javaMailSender.send(mimeMessage);
            sent = true;
        } catch (Exception e) {
            LOGGER.error("Error while sending email", e);
        }
        return sent;
    }
}

sendResetPasswordEmail first retrieves the required data for email formatting from ResetPasswordRequest and will create a Map of the data to be submitted to EmailTemplateService.getResetPasswordEmail, along with the template file named reset_password_en.html, which is available inside the src/resources/templates directory. Then, it will use JavaMailSender to actually send the email.