The configuration for User Registration is as follows:

@SpringBootApplication
@Import(RepoConfig.class)
public class SpringBootUserRegistrationApplication {
    ...

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

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

}

In the preceding configuration, an Apache Kafka topic by the name resetPasswordRequests is created as a Spring Bean with a number of replications 10 and replication factor of 2. The resetPasswordRequests topic is responsible for sending the message to the consumer.

The kafkaTemplate Spring Bean is created with an instance of KafkaTemplate to enable communication.

Furthermore, the following AsyncService is used to produce reset password requests:

@Service
@Transactional(readOnly = true)
public class AsyncService {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final UserService userService;
    private final ObjectMapper objectMapper;
    private final PasswordEncoder passwordEncoder;


    public AsyncService(KafkaTemplate<String, String> kafkaTemplate, UserService userService, ObjectMapper objectMapper, PasswordEncoder passwordEncoder) {
        this.kafkaTemplate = kafkaTemplate;
        this.userService = userService;
        this.objectMapper = objectMapper;
        this.passwordEncoder = passwordEncoder;
    }

    @Transactional(rollbackFor = Exception.class)
    public void sendResetPassword(String email) throws IOException {
        User user = userService.getByEmail(email);

        if (user != null) {
            ResetPasswordRequest resetPasswordRequest = new ResetPasswordRequest();
            resetPasswordRequest.setEmail(user.getEmail());

            RandomStringGenerator generator = new RandomStringGenerator.Builder()
                    .withinRange('a', 'z').build();
            String newPassword = generator.generate(10);
            resetPasswordRequest.setNewPassword(newPassword);
            resetPasswordRequest.setUsername(user.getUsername());
            ProducerRecord<String, String> record = new ProducerRecord<>("resetPasswordRequests", objectMapper.writeValueAsString(resetPasswordRequest));
            kafkaTemplate.send(record);

            user.setPassword(passwordEncoder.encode(newPassword));
            userService.save(user);

        }
    }
}

The sendResetPassword method will find the User by email, and if found will generate a 10 digit random password, send a JSON message to the resetPasswordRequests Apache Kafka topic, and update the user password with the generated password before saving User.