BookEvent.java

package zup.edu.livraria.samples.books;

import java.math.BigDecimal;
import java.util.UUID;

public class BookEvent {
    private UUID id;
    private String title;
    private String description;
    private BigDecimal price;

    public BookEvent(String title, String description, BigDecimal price) {
        this.id = UUID.randomUUID();
        this.title = title;
        this.description = description;
        this.price = price;
    }

    public BookEvent() {
    }

    
    public UUID getId() {
        return id;
    }

    public String getTitle() {
        return title;
    }

    public String getDescription() {
        return description;
    }

    public BigDecimal getPrice() {
        return price;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (!(o instanceof BookEvent)) return false;

        BookEvent bookEvent = (BookEvent) o;

        if (id != null ? !id.equals(bookEvent.id) : bookEvent.id != null) return false;
        if (title != null ? !title.equals(bookEvent.title) : bookEvent.title != null) return false;
        if (description != null ? !description.equals(bookEvent.description) : bookEvent.description != null)
            return false;
        return price != null ? price.equals(bookEvent.price) : bookEvent.price == null;
    }

    @Override
    public int hashCode() {
        int result = id != null ? id.hashCode() : 0;
        result = 31 * result + (title != null ? title.hashCode() : 0);
        result = 31 * result + (description != null ? description.hashCode() : 0);
        result = 31 * result + (price != null ? price.hashCode() : 0);
        return result;
    }

    @Override
    public String toString() {
        return "BookEvent{" +
                "id=" + id +
                ", title='" + title + '\'' +
                ", description='" + description + '\'' +
                ", price=" + price +
                '}';
    }
}

BookProducer.java

@Component
public class BookProducer {

    @Autowired
    private KafkaTemplate<String,BookEvent> template;
    private static final Logger LOGGER = LoggerFactory.getLogger(BookProducer.class);

    @Value("${spring.kafka.producer.topic}" )
    private String topico;


    public void enviar(BookEvent book){
        template.send(topico, book);
        LOGGER.info("book submmit sucess!{}",book);
    }


}

EmbeddedKafka.java

@SpringBootTest
@ActiveProfiles("test")
@EmbeddedKafka(partitions = 1, topics = "books", brokerProperties = {
        "listeners=PLAINTEXT://localhost:9092", "port=9092",
})
class BookProducerTest {
    @Autowired
    private BookProducer producer;
    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;
    private Consumer<String, BookEvent> consumer;

    @BeforeEach
    void setUp() {
        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("book_api", "true", this.embeddedKafka);
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        DefaultKafkaConsumerFactory<String, BookEvent> consumerFactory = new DefaultKafkaConsumerFactory<>(
                consumerProps, new StringDeserializer(), new BookEventDeserializer()
        );
        this.consumer = consumerFactory.createConsumer();
        this.embeddedKafka.consumeFromAllEmbeddedTopics(consumer);

    }


    @Test
    @DisplayName("should contain a message in the books topic")
    void t1() throws Exception {
        BookEvent bookEvent = new BookEvent("Domain Drive Design", "DDD da massa", BigDecimal.TEN);

        producer.enviar(bookEvent);

        ConsumerRecords<String, BookEvent> records = KafkaTestUtils.getRecords(consumer);

        assertThat(records.count())
                .isEqualTo(1);
    }
}