reactor-kafka

UserService.java

@Service
@Slf4j
@AllArgsConstructor
public class UserService {
    private final KafkaProducer kafkaProducer;

    private static final String USER_TOPIC = "user-create";
  
    public Mono<String> userCreate(Mono<UserDto> username){
        return username.flatMap(name ->
                kafkaProducer.kafkaSender()
                        .send(USER_TOPIC, name.getUsername())
                        .doOnNext(voidSenderResult -> log.info("{} sent to {} topic", name.getUsername(), USER_TOPIC))
                        .doOnNext(voidSenderResult -> log.info("{}", voidSenderResult.recordMetadata()))
                        .thenReturn("It was sent successfully")
        );
    }
}