diff --git a/README.md b/README.md index c45e36d..c3548dd 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ SpringBoot Playground is a multi-module Maven project that showcases a note-taki - **Event-Driven Architecture** using Apache Kafka - **Microservices** communication via message streaming +- **Outbox** pattern when communicating with Apache Kafka - **AI-Powered Features** using OpenAI and AWS Bedrock - **Stream Processing** with Kafka Streams - **Multi-Database Support** (PostgreSQL, MongoDB, OpenSearch) diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/config/outbox/SchedulerConfig.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/config/outbox/SchedulerConfig.java new file mode 100644 index 0000000..61425c5 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/config/outbox/SchedulerConfig.java @@ -0,0 +1,9 @@ +package org.softwarecave.springbootnote.config.outbox; + +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.annotation.EnableScheduling; + +@Configuration +@EnableScheduling +public class SchedulerConfig { +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/StickyNoteRecorderAspect.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/StickyNoteRecorderAspect.java index 7111e83..d05dfb0 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/StickyNoteRecorderAspect.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/StickyNoteRecorderAspect.java @@ -1,22 +1,24 @@ package org.softwarecave.springbootnote.notification; +import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.aspectj.lang.JoinPoint; import org.aspectj.lang.annotation.AfterReturning; import org.aspectj.lang.annotation.Aspect; import org.softwarecave.springbootnote.note.model.StickyNote; -import org.softwarecave.springbootnote.notification.kafka.KafkaStickyNoteProducer; +import org.softwarecave.springbootnote.outbox.service.queue.OutboxService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import java.util.List; @Component +@RequiredArgsConstructor @Aspect @Slf4j public class StickyNoteRecorderAspect { - private List kafkaStickyNoteProducers; + private List outboxServices; @AfterReturning(pointcut = "@annotation(rec)", returning = "returnValue") @@ -31,16 +33,19 @@ public void addStickyNote(JoinPoint joinPoint, Recordable rec, Object returnValu } } + sendToOutbox((StickyNote) returnValue); + } - if (kafkaStickyNoteProducers != null && !kafkaStickyNoteProducers.isEmpty()) { - kafkaStickyNoteProducers.forEach(e -> e.sendToKafka((StickyNote) returnValue)); + private void sendToOutbox(StickyNote stickyNote) { + if (outboxServices != null && !outboxServices.isEmpty()) { + outboxServices.forEach(outboxService -> outboxService.send(stickyNote)); } else { - log.warn("No KafkaStickyNoteProducers are available, skipping Kafka notification"); + log.warn("No outbox services are available, skipping Kafka notification"); } } @Autowired(required = false) - public void setKafkaStickyNoteProducer(List kafkaStickyNoteProducers) { - this.kafkaStickyNoteProducers = kafkaStickyNoteProducers; + public void setOutboxService(List outboxServices) { + this.outboxServices = outboxServices; } } diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/KafkaStickyNoteProducer.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/KafkaStickyNoteProducer.java deleted file mode 100644 index e1a4439..0000000 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/KafkaStickyNoteProducer.java +++ /dev/null @@ -1,7 +0,0 @@ -package org.softwarecave.springbootnote.notification.kafka; - -import org.softwarecave.springbootnote.note.model.StickyNote; - -public interface KafkaStickyNoteProducer { - void sendToKafka(StickyNote value); -} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/AvroKafkaStickyNoteProducer.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaAvroProducer.java similarity index 57% rename from springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/AvroKafkaStickyNoteProducer.java rename to springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaAvroProducer.java index ed42222..48a60d8 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/AvroKafkaStickyNoteProducer.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaAvroProducer.java @@ -1,50 +1,54 @@ -package org.softwarecave.springbootnote.notification.kafka; +package org.softwarecave.springbootnote.outbox.kafka; import io.confluent.kafka.serializers.KafkaAvroSerializer; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; +import org.apache.avro.specific.SpecificRecord; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.serialization.LongSerializer; -import org.softwarecave.springbootnote.note.model.StickyNote; -import org.softwarecave.springbootnote.notification.kafka.converter.AvroStickyNoteConverter; +import org.softwarecave.springbootnote.note.web.converter.StickyNoteConverter; +import org.softwarecave.springbootnote.outbox.model.Outbox; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; import org.springframework.stereotype.Component; import java.util.Properties; +import java.util.concurrent.Future; + +import static org.softwarecave.springbootnote.outbox.tools.AvroTools.fromBytes; @Slf4j @Component @ConditionalOnBooleanProperty(prefix = "app.kafka.avro", name = "enabled", havingValue = true) -public class AvroKafkaStickyNoteProducer implements KafkaStickyNoteProducer { +public class KafkaAvroProducer { private final String bootstrapServers; private final String noteTopic; private final String schemaRegistryUrl; - private final AvroStickyNoteConverter converter; - private KafkaProducer kafkaProducer; + private KafkaProducer kafkaProducer; - public AvroKafkaStickyNoteProducer(@Value("${app.kafka.bootstrap-servers}") String bootstrapServers, - @Value("${app.kafka.avro.stickynote-topic}") String noteTopic, - @Value("${app.kafka.schema-registry-url}") String schemaRegistryUrl) { + public KafkaAvroProducer( + StickyNoteConverter stickyNoteConverter, + @Value("${app.kafka.bootstrap-servers}") String bootstrapServers, + @Value("${app.kafka.avro.stickynote-topic}") String noteTopic, + @Value("${app.kafka.schema-registry-url}") String schemaRegistryUrl) { this.bootstrapServers = bootstrapServers; this.noteTopic = noteTopic; this.schemaRegistryUrl = schemaRegistryUrl; - this.converter = new AvroStickyNoteConverter(); } @PostConstruct public void init() { Properties properties = new Properties(); properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); - properties.setProperty("schema.registry.url", schemaRegistryUrl); properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName()); properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName()); + properties.setProperty("schema.registry.url", schemaRegistryUrl); properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "50"); properties.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip"); @@ -59,22 +63,15 @@ public void destroy() { log.info("Closed Kafka producer"); } - @Override - public void sendToKafka(StickyNote value) { - var stickyNoteAvro = converter.convertToAvro(value); - - Long key = stickyNoteAvro.getId(); - var producerRecord = new ProducerRecord<>(noteTopic, key, stickyNoteAvro); - kafkaProducer.send(producerRecord, (RecordMetadata metadata, Exception exception) -> { - if (exception != null) { - log.error("Failed to send message to Kafka", exception); - } else { - log.info("Message sent to Kafka topic {} partition {} with offset {}", - metadata.topic(), metadata.partition(), metadata.offset()); - } - }); + public Future sendToKafka(Outbox value, Class avroClass) { + try { + var avroObject = fromBytes(value.getPayloadBytes(), avroClass); + var producerRecord = new ProducerRecord(noteTopic, value.getAggregateId(), avroObject); + return kafkaProducer.send(producerRecord); + } catch (Exception e) { + throw new RuntimeException(e); + } } - } diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/JsonKafkaStickyNoteProducer.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaJsonProducer.java similarity index 55% rename from springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/JsonKafkaStickyNoteProducer.java rename to springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaJsonProducer.java index a6dbe60..5cbd4d7 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/JsonKafkaStickyNoteProducer.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaJsonProducer.java @@ -1,4 +1,4 @@ -package org.softwarecave.springbootnote.notification.kafka; +package org.softwarecave.springbootnote.outbox.kafka; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; @@ -9,37 +9,31 @@ import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.serialization.LongSerializer; import org.apache.kafka.common.serialization.StringSerializer; -import org.softwarecave.springbootnote.note.model.StickyNote; -import org.softwarecave.springbootnote.note.web.StickyNoteDTO; import org.softwarecave.springbootnote.note.web.converter.StickyNoteConverter; +import org.softwarecave.springbootnote.outbox.model.Outbox; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; import org.springframework.stereotype.Component; -import tools.jackson.databind.json.JsonMapper; import java.util.Properties; +import java.util.concurrent.Future; @Slf4j @Component @ConditionalOnBooleanProperty(prefix = "app.kafka.json", name = "enabled", havingValue = true) -public class JsonKafkaStickyNoteProducer implements KafkaStickyNoteProducer { +public class KafkaJsonProducer { - private final JsonMapper jsonMapper; private final String bootstrapServers; private final String noteTopic; private KafkaProducer kafkaProducer; - private final StickyNoteConverter stickyNoteConverter; - public JsonKafkaStickyNoteProducer(JsonMapper jsonMapper, - StickyNoteConverter stickyNoteConverter, - @Value("${app.kafka.bootstrap-servers}") String bootstrapServers, - @Value("${app.kafka.json.stickynote-topic}") String noteTopic) { - this.jsonMapper = jsonMapper; + public KafkaJsonProducer( + StickyNoteConverter stickyNoteConverter, + @Value("${app.kafka.bootstrap-servers}") String bootstrapServers, + @Value("${app.kafka.json.stickynote-topic}") String noteTopic) { this.bootstrapServers = bootstrapServers; this.noteTopic = noteTopic; - - this.stickyNoteConverter = stickyNoteConverter; } @PostConstruct @@ -62,20 +56,9 @@ public void destroy() { log.info("Closed Kafka producer"); } - @Override - public void sendToKafka(StickyNote value) { - StickyNoteDTO dto = stickyNoteConverter.convertToDTO(value); - Long key = dto.getId(); - String jsonValue = jsonMapper.writeValueAsString(dto); - var producerRecord = new ProducerRecord<>(noteTopic, key, jsonValue); - kafkaProducer.send(producerRecord, (RecordMetadata metadata, Exception exception) -> { - if (exception != null) { - log.error("Failed to send message to Kafka", exception); - } else { - log.info("Message sent to Kafka topic {} partition {} with offset {}", - metadata.topic(), metadata.partition(), metadata.offset()); - } - }); + public Future sendToKafka(Outbox value) { + var producerRecord = new ProducerRecord<>(noteTopic, value.getAggregateId(), value.getPayloadString()); + return kafkaProducer.send(producerRecord); } } diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/converter/AvroStickyNoteConverter.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverter.java similarity index 94% rename from springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/converter/AvroStickyNoteConverter.java rename to springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverter.java index 1de8d3d..1564344 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/converter/AvroStickyNoteConverter.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverter.java @@ -1,14 +1,17 @@ -package org.softwarecave.springbootnote.notification.kafka.converter; +package org.softwarecave.springbootnote.outbox.kafka.converter; import lombok.extern.slf4j.Slf4j; import org.softwarecave.springbootnote.note.model.StickyNote; import org.softwarecave.springbootnote.tag.model.Tag; +import org.springframework.stereotype.Component; import java.time.ZoneOffset; import java.util.List; +@Component @Slf4j public class AvroStickyNoteConverter { + public org.softwarecave.springbootnote.avro.StickyNote convertToAvro(StickyNote value) { var linksAvro = convertLinksToAvro(value); var tagsAvro = convertTagsToAvro(value); diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/AggregateType.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/AggregateType.java new file mode 100644 index 0000000..7a0e073 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/AggregateType.java @@ -0,0 +1,17 @@ +package org.softwarecave.springbootnote.outbox.model; + +import lombok.Getter; +import org.apache.avro.specific.SpecificRecord; +import org.softwarecave.springbootnote.avro.StickyNote; + +@Getter +public enum AggregateType { + STICKY_NOTE(StickyNote.class); + + private final Class avroClass; + + AggregateType(Class avroClass) { + this.avroClass = avroClass; + } + +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/MessageType.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/MessageType.java new file mode 100644 index 0000000..98df14b --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/MessageType.java @@ -0,0 +1,6 @@ +package org.softwarecave.springbootnote.outbox.model; + +public enum MessageType { + JSON, + AVRO +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/Outbox.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/Outbox.java new file mode 100644 index 0000000..3190e3b --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/Outbox.java @@ -0,0 +1,57 @@ +package org.softwarecave.springbootnote.outbox.model; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.EnumType; +import jakarta.persistence.Enumerated; +import jakarta.persistence.GeneratedValue; +import jakarta.persistence.GenerationType; +import jakarta.persistence.Id; +import jakarta.persistence.SequenceGenerator; +import jakarta.persistence.Table; +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; +import lombok.ToString; + +import java.time.ZonedDateTime; + +@Entity +@Table(name = "outbox") +@NoArgsConstructor +@AllArgsConstructor +@ToString +@Getter +@Setter +public class Outbox { + + @Id + @GeneratedValue(strategy = GenerationType.SEQUENCE) + @SequenceGenerator(name = "outbox_seq", allocationSize = 1) + private Long id; + + @Column(name = "aggregate_type") + @Enumerated(EnumType.STRING) + private AggregateType aggregateType; + + @Column(name = "aggregate_id") + private Long aggregateId; + + @Column(name = "message_type") + @Enumerated(EnumType.STRING) + private MessageType messageType; + + @Column(name = "created_date") + private ZonedDateTime createdDate; + + @Column(name = "payload_bytes") + private byte[] payloadBytes; + + @Column(name = "payload_string") + private String payloadString; + + @Column(name = "status") + @Enumerated(EnumType.STRING) + private Status status; +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/Status.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/Status.java new file mode 100644 index 0000000..64669f0 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/Status.java @@ -0,0 +1,6 @@ +package org.softwarecave.springbootnote.outbox.model; + +public enum Status { + NEW, + SENT +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/InvalidOutboxDataException.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/InvalidOutboxDataException.java new file mode 100644 index 0000000..0a4898d --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/InvalidOutboxDataException.java @@ -0,0 +1,11 @@ +package org.softwarecave.springbootnote.outbox.service; + +public class InvalidOutboxDataException extends RuntimeException { + public InvalidOutboxDataException(String message) { + super(message); + } + + public InvalidOutboxDataException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxRepository.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxRepository.java new file mode 100644 index 0000000..7759a43 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxRepository.java @@ -0,0 +1,11 @@ +package org.softwarecave.springbootnote.outbox.service; + +import org.softwarecave.springbootnote.outbox.model.Outbox; +import org.softwarecave.springbootnote.outbox.model.Status; +import org.springframework.data.domain.Page; +import org.springframework.data.domain.Pageable; +import org.springframework.data.jpa.repository.JpaRepository; + +public interface OutboxRepository extends JpaRepository { + Page findByStatus(Status status, Pageable pageable); +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcher.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcher.java new file mode 100644 index 0000000..1ef72a1 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcher.java @@ -0,0 +1,98 @@ +package org.softwarecave.springbootnote.outbox.service.dispatch; + +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.softwarecave.springbootnote.outbox.model.MessageType; +import org.softwarecave.springbootnote.outbox.model.Outbox; +import org.softwarecave.springbootnote.outbox.model.Status; +import org.softwarecave.springbootnote.outbox.service.InvalidOutboxDataException; +import org.softwarecave.springbootnote.outbox.service.OutboxRepository; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.domain.Page; +import org.springframework.data.domain.PageRequest; +import org.springframework.data.domain.Sort; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +@Service +@Slf4j +public class OutboxDispatcher { + + private final OutboxRepository outboxRepository; + private final Map dispatcherStrategies; + + public OutboxDispatcher(OutboxRepository outboxRepository) { + this.outboxRepository = outboxRepository; + this.dispatcherStrategies = new HashMap<>(); + } + + @Autowired(required = false) + public void setDispatcherStrategies(List dispatcherStrategies) { + + for (var dispatcherStrategy : dispatcherStrategies) { + var prevValue = this.dispatcherStrategies.putIfAbsent(dispatcherStrategy.getMessageType(), dispatcherStrategy); + if (prevValue != null) { + throw new InvalidOutboxDataException("There are conflicting Outbox dispatchers strategies with the same message type"); + } + } + } + + @Scheduled(fixedDelayString = "${app.outbox.sender.delay}", timeUnit = TimeUnit.MILLISECONDS) + public void process() { + var entryList = outboxRepository.findByStatus(Status.NEW, + PageRequest.of(0, 100, Sort.by(Sort.Order.asc("createdDate")))); + log.info("Fetched {} entries from outbox to process", entryList.getContent().size()); + + var futureList = sendToKafka(entryList); + + waitForKafkaAcks(futureList, entryList); + + } + + private void waitForKafkaAcks(ArrayList> futureList, Page entryList) { + for (int i = 0; i < futureList.size(); i++) { + var future = futureList.get(i); + try { + RecordMetadata recordMetadata = future.get(); + var entry = entryList.getContent().get(i); + + updateStatusAsSent(entry); + } catch (InterruptedException | ExecutionException e) { + log.error("Failed sending the message from outbox ", e); + } + } + } + + private void updateStatusAsSent(Outbox entry) { + log.info("Set the status of outbox entry {} to SENT", entry.getPayloadString()); + entry.setStatus(Status.SENT); + outboxRepository.save(entry); + } + + private ArrayList> sendToKafka(Page entryList) { + var futureList = new ArrayList>(); + for (var entry : entryList) { + futureList.add(sendToKafka(entry)); + } + return futureList; + } + + private Future sendToKafka(Outbox entry) { + MessageType messageType = entry.getMessageType(); + var dispatcherStrategy = dispatcherStrategies.get(messageType); + if (dispatcherStrategy != null) { + return dispatcherStrategy.send(entry); + } else { + throw new InvalidOutboxDataException("Unrecognized message type " + messageType); + } + } + +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherAvroStrategy.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherAvroStrategy.java new file mode 100644 index 0000000..510731b --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherAvroStrategy.java @@ -0,0 +1,46 @@ +package org.softwarecave.springbootnote.outbox.service.dispatch; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.avro.specific.SpecificRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.softwarecave.springbootnote.outbox.kafka.KafkaAvroProducer; +import org.softwarecave.springbootnote.outbox.model.AggregateType; +import org.softwarecave.springbootnote.outbox.model.MessageType; +import org.softwarecave.springbootnote.outbox.model.Outbox; +import org.softwarecave.springbootnote.outbox.service.InvalidOutboxDataException; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; + +import java.util.concurrent.Future; + +@RequiredArgsConstructor +@Service +@Slf4j +public class OutboxDispatcherAvroStrategy implements OutboxDispatcherStrategy { + private KafkaAvroProducer kafkaAvroProducer; + + @Autowired(required = false) + public void setKafkaAvroProducer(KafkaAvroProducer kafkaAvroProducer) { + this.kafkaAvroProducer = kafkaAvroProducer; + } + + @Override + public Future send(Outbox outbox) { + return kafkaAvroProducer.sendToKafka(outbox, getAvroClass(outbox)); + } + + @Override + public MessageType getMessageType() { + return MessageType.AVRO; + } + + private Class getAvroClass(Outbox entry) { + AggregateType aggregateType = entry.getAggregateType(); + if (aggregateType != null) { + return aggregateType.getAvroClass(); + } else { + throw new InvalidOutboxDataException("Unrecognized aggregate type " + aggregateType); + } + } +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherJsonStrategy.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherJsonStrategy.java new file mode 100644 index 0000000..3000bcd --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherJsonStrategy.java @@ -0,0 +1,35 @@ +package org.softwarecave.springbootnote.outbox.service.dispatch; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.softwarecave.springbootnote.outbox.kafka.KafkaJsonProducer; +import org.softwarecave.springbootnote.outbox.model.MessageType; +import org.softwarecave.springbootnote.outbox.model.Outbox; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; + +import java.util.concurrent.Future; + +@RequiredArgsConstructor +@Service +@Slf4j +public class OutboxDispatcherJsonStrategy implements OutboxDispatcherStrategy { + + private KafkaJsonProducer kafkaJsonProducer; + + @Autowired(required = false) + public void setKafkaJsonProducer(KafkaJsonProducer kafkaJsonProducer) { + this.kafkaJsonProducer = kafkaJsonProducer; + } + + @Override + public Future send(Outbox outbox) { + return kafkaJsonProducer.sendToKafka(outbox); + } + + @Override + public MessageType getMessageType() { + return MessageType.JSON; + } +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherStrategy.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherStrategy.java new file mode 100644 index 0000000..80a1175 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherStrategy.java @@ -0,0 +1,13 @@ +package org.softwarecave.springbootnote.outbox.service.dispatch; + +import org.apache.kafka.clients.producer.RecordMetadata; +import org.softwarecave.springbootnote.outbox.model.MessageType; +import org.softwarecave.springbootnote.outbox.model.Outbox; + +import java.util.concurrent.Future; + +public interface OutboxDispatcherStrategy { + Future send(Outbox outbox); + + MessageType getMessageType(); +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/AvroOutboxService.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/AvroOutboxService.java new file mode 100644 index 0000000..28706d2 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/AvroOutboxService.java @@ -0,0 +1,44 @@ +package org.softwarecave.springbootnote.outbox.service.queue; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.softwarecave.springbootnote.note.model.StickyNote; +import org.softwarecave.springbootnote.outbox.kafka.converter.AvroStickyNoteConverter; +import org.softwarecave.springbootnote.outbox.model.AggregateType; +import org.softwarecave.springbootnote.outbox.model.MessageType; +import org.softwarecave.springbootnote.outbox.model.Outbox; +import org.softwarecave.springbootnote.outbox.model.Status; +import org.softwarecave.springbootnote.outbox.service.OutboxRepository; +import org.softwarecave.springbootnote.outbox.tools.AvroTools; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.time.ZonedDateTime; + +@Service +@RequiredArgsConstructor +@Slf4j +@ConditionalOnBooleanProperty(prefix = "app.kafka.avro", name = "enabled", havingValue = true) +public class AvroOutboxService implements OutboxService { + + private final OutboxRepository outboxRepository; + private final AvroStickyNoteConverter avroStickyNoteConverter; + + @Transactional + @Override + public void send(StickyNote stickyNote) { + var avroStickyNote = avroStickyNoteConverter.convertToAvro(stickyNote); + byte[] payloadBinary = AvroTools.convertToBytes(avroStickyNote); + + save(stickyNote.getId(), payloadBinary, AggregateType.STICKY_NOTE); + } + + private void save(Long id, byte[] payloadBinary, AggregateType aggregateType) { + Outbox outbox = new Outbox(null, aggregateType, id, + MessageType.AVRO, ZonedDateTime.now(), payloadBinary, null, Status.NEW); + + log.info("Saving to outbox: {}", outbox); + outboxRepository.save(outbox); + } +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/JsonOutboxService.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/JsonOutboxService.java new file mode 100644 index 0000000..1d09bbd --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/JsonOutboxService.java @@ -0,0 +1,47 @@ +package org.softwarecave.springbootnote.outbox.service.queue; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.softwarecave.springbootnote.note.model.StickyNote; +import org.softwarecave.springbootnote.note.web.StickyNoteDTO; +import org.softwarecave.springbootnote.note.web.converter.StickyNoteConverter; +import org.softwarecave.springbootnote.outbox.model.AggregateType; +import org.softwarecave.springbootnote.outbox.model.MessageType; +import org.softwarecave.springbootnote.outbox.model.Outbox; +import org.softwarecave.springbootnote.outbox.model.Status; +import org.softwarecave.springbootnote.outbox.service.OutboxRepository; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import tools.jackson.databind.json.JsonMapper; + +import java.time.ZonedDateTime; + +@Service +@RequiredArgsConstructor +@Slf4j +@ConditionalOnBooleanProperty(prefix = "app.kafka.json", name = "enabled", havingValue = true) +public class JsonOutboxService implements OutboxService { + + private final OutboxRepository outboxRepository; + private final JsonMapper jsonMapper; + private final StickyNoteConverter stickyNoteConverter; + + @Transactional + @Override + public void send(StickyNote stickyNote) { + StickyNoteDTO dto = stickyNoteConverter.convertToDTO(stickyNote); + String payloadString = jsonMapper.writeValueAsString(dto); + + save(stickyNote.getId(), payloadString, AggregateType.STICKY_NOTE); + } + + private void save(Long id, String payloadString, AggregateType aggregateType) { + Outbox outbox = new Outbox(null, aggregateType, id, + MessageType.JSON, ZonedDateTime.now(), null, payloadString, Status.NEW); + + log.info("Saving to outbox: {}", outbox); + outboxRepository.save(outbox); + } + +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/OutboxService.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/OutboxService.java new file mode 100644 index 0000000..089327f --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/OutboxService.java @@ -0,0 +1,7 @@ +package org.softwarecave.springbootnote.outbox.service.queue; + +import org.softwarecave.springbootnote.note.model.StickyNote; + +public interface OutboxService { + void send(StickyNote stickyNote); +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/tools/AvroTools.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/tools/AvroTools.java new file mode 100644 index 0000000..f6847e3 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/tools/AvroTools.java @@ -0,0 +1,41 @@ +package org.softwarecave.springbootnote.outbox.tools; + + +import lombok.extern.slf4j.Slf4j; +import org.apache.avro.io.BinaryDecoder; +import org.apache.avro.io.BinaryEncoder; +import org.apache.avro.io.DatumReader; +import org.apache.avro.io.DecoderFactory; +import org.apache.avro.io.EncoderFactory; +import org.apache.avro.specific.SpecificDatumReader; +import org.apache.avro.specific.SpecificDatumWriter; +import org.apache.avro.specific.SpecificRecord; +import org.softwarecave.springbootnote.outbox.service.InvalidOutboxDataException; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; + +@Slf4j +public class AvroTools { + + public static byte[] convertToBytes(T record) { + try { + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + SpecificDatumWriter writer = new SpecificDatumWriter<>(record.getSchema()); + BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(outputStream, null); + + writer.write(record, encoder); + encoder.flush(); + + return outputStream.toByteArray(); + } catch (IOException e) { + throw new InvalidOutboxDataException("Failed to convert Avro record into bytes", e); + } + } + + public static T fromBytes(byte[] data, Class avroClass) throws Exception { + DatumReader reader = new SpecificDatumReader<>(avroClass); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + return reader.read(null, decoder); + } +} diff --git a/springboot-note/src/main/resources/application.yaml b/springboot-note/src/main/resources/application.yaml index 783446d..bf7084d 100644 --- a/springboot-note/src/main/resources/application.yaml +++ b/springboot-note/src/main/resources/application.yaml @@ -51,3 +51,6 @@ app: avro: enabled: true stickynote-topic: playground.stickynote.avro + outbox: + sender: + delay: 1000 diff --git a/springboot-note/src/main/resources/db/migration/V1.3__AddOutbox.sql b/springboot-note/src/main/resources/db/migration/V1.3__AddOutbox.sql new file mode 100644 index 0000000..0b58172 --- /dev/null +++ b/springboot-note/src/main/resources/db/migration/V1.3__AddOutbox.sql @@ -0,0 +1,13 @@ +CREATE TABLE outbox ( + id BIGINT PRIMARY KEY, + aggregate_type VARCHAR(64) NOT NULL, + aggregate_id BIGINT NOT NULL, + message_type VARCHAR(64) NOT NULL, + created_date TIMESTAMP WITH TIME ZONE NOT NULL, + payload_bytes BYTEA, + payload_string TEXT, + status VARCHAR(64) NOT NULL, + CHECK(payload_string IS NOT NULL OR payload_bytes IS NOT NULL) +); +CREATE SEQUENCE outbox_seq START WITH 1 INCREMENT BY 1; + diff --git a/springboot-note/src/test/java/org/softwarecave/springbootnote/SpringbootNoteApplicationTests.java b/springboot-note/src/test/java/org/softwarecave/springbootnote/SpringbootNoteApplicationTests.java index b1af049..ffe4edf 100644 --- a/springboot-note/src/test/java/org/softwarecave/springbootnote/SpringbootNoteApplicationTests.java +++ b/springboot-note/src/test/java/org/softwarecave/springbootnote/SpringbootNoteApplicationTests.java @@ -6,8 +6,8 @@ @SpringBootTest class SpringbootNoteApplicationTests { - @Test - void contextLoads() { - } + @Test + void contextLoads() { + } } diff --git a/springboot-note/src/test/java/org/softwarecave/springbootnote/note/StickyNoteServiceIntegrationTest.java b/springboot-note/src/test/java/org/softwarecave/springbootnote/note/StickyNoteServiceIntegrationTest.java index fa0e0e3..37636e9 100644 --- a/springboot-note/src/test/java/org/softwarecave/springbootnote/note/StickyNoteServiceIntegrationTest.java +++ b/springboot-note/src/test/java/org/softwarecave/springbootnote/note/StickyNoteServiceIntegrationTest.java @@ -16,7 +16,6 @@ import java.time.LocalDateTime; import java.util.ArrayList; import java.util.List; -import java.util.Optional; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertFalse; diff --git a/springboot-note/src/test/java/org/softwarecave/springbootnote/note/web/StickyNoteControllerTest.java b/springboot-note/src/test/java/org/softwarecave/springbootnote/note/web/StickyNoteControllerTest.java index 9e944bd..f207ad1 100644 --- a/springboot-note/src/test/java/org/softwarecave/springbootnote/note/web/StickyNoteControllerTest.java +++ b/springboot-note/src/test/java/org/softwarecave/springbootnote/note/web/StickyNoteControllerTest.java @@ -14,7 +14,6 @@ import org.springframework.http.MediaType; import org.springframework.security.test.context.support.WithMockUser; import org.springframework.test.context.bean.override.mockito.MockitoBean; -import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; import org.springframework.test.web.servlet.MockMvc; import tools.jackson.databind.json.JsonMapper; diff --git a/springboot-note/src/test/java/org/softwarecave/springbootnote/notification/kafka/converter/AvroStickyNoteConverterTest.java b/springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverterTest.java similarity index 98% rename from springboot-note/src/test/java/org/softwarecave/springbootnote/notification/kafka/converter/AvroStickyNoteConverterTest.java rename to springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverterTest.java index 316ef54..48e96da 100644 --- a/springboot-note/src/test/java/org/softwarecave/springbootnote/notification/kafka/converter/AvroStickyNoteConverterTest.java +++ b/springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverterTest.java @@ -1,4 +1,4 @@ -package org.softwarecave.springbootnote.notification.kafka.converter; +package org.softwarecave.springbootnote.outbox.kafka.converter; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; diff --git a/springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherAvroStrategyTest.java b/springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherAvroStrategyTest.java new file mode 100644 index 0000000..54a568e --- /dev/null +++ b/springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherAvroStrategyTest.java @@ -0,0 +1,92 @@ +package org.softwarecave.springbootnote.outbox.service.dispatch; + +import org.apache.kafka.clients.producer.RecordMetadata; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.softwarecave.springbootnote.avro.StickyNote; +import org.softwarecave.springbootnote.outbox.kafka.KafkaAvroProducer; +import org.softwarecave.springbootnote.outbox.model.AggregateType; +import org.softwarecave.springbootnote.outbox.model.MessageType; +import org.softwarecave.springbootnote.outbox.model.Outbox; +import org.softwarecave.springbootnote.outbox.model.Status; +import org.softwarecave.springbootnote.outbox.service.InvalidOutboxDataException; + +import java.time.ZonedDateTime; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Future; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class OutboxDispatcherAvroStrategyTest { + + @Mock + private KafkaAvroProducer kafkaAvroProducer; + + @InjectMocks + private OutboxDispatcherAvroStrategy strategy; + + @Test + void testGetMessageType() { + assertThat(strategy.getMessageType()).isEqualTo(MessageType.AVRO); + } + + @Test + void testSendWithValidStickyNoteOutbox() { + Outbox outbox = createOutboxWithStickyNote(); + Future mockFuture = CompletableFuture.completedFuture(mock(RecordMetadata.class)); + + when(kafkaAvroProducer.sendToKafka(eq(outbox), eq(StickyNote.class))) + .thenReturn(mockFuture); + + Future result = strategy.send(outbox); + + assertThat(result).isEqualTo(mockFuture); + verify(kafkaAvroProducer).sendToKafka(outbox, StickyNote.class); + } + + @Test + void testSendWithNullAggregateType() { + Outbox outbox = new Outbox(1L, AggregateType.STICKY_NOTE, 100L, MessageType.AVRO, ZonedDateTime.now(), new byte[]{1, 2, 3}, null, Status.NEW); + outbox.setAggregateType(null); + + assertThatThrownBy(() -> strategy.send(outbox)) + .isInstanceOf(InvalidOutboxDataException.class) + .hasMessageContaining("Unrecognized aggregate type"); + } + + @Test + void testSendWithDifferentPayloadBytes() { + Outbox outbox1 = createOutboxWithStickyNote(); + outbox1.setPayloadBytes(new byte[]{1, 2, 3, 4, 5}); + + Outbox outbox2 = createOutboxWithStickyNote(); + outbox2.setId(2L); + outbox2.setPayloadBytes(new byte[]{10, 20, 30}); + + Future mockFuture = CompletableFuture.completedFuture(mock(RecordMetadata.class)); + when(kafkaAvroProducer.sendToKafka(any(Outbox.class), eq(StickyNote.class))) + .thenReturn(mockFuture); + + strategy.send(outbox1); + strategy.send(outbox2); + + verify(kafkaAvroProducer).sendToKafka(outbox1, StickyNote.class); + verify(kafkaAvroProducer).sendToKafka(outbox2, StickyNote.class); + } + + private Outbox createOutboxWithStickyNote() { + return new Outbox(1L, AggregateType.STICKY_NOTE, 100L, + MessageType.AVRO, ZonedDateTime.now(), + new byte[]{1, 2, 3}, null, Status.NEW); + } +} diff --git a/springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherJsonStrategyTest.java b/springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherJsonStrategyTest.java new file mode 100644 index 0000000..50cb2b9 --- /dev/null +++ b/springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherJsonStrategyTest.java @@ -0,0 +1,102 @@ +package org.softwarecave.springbootnote.outbox.service.dispatch; + +import org.apache.kafka.clients.producer.RecordMetadata; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.softwarecave.springbootnote.outbox.kafka.KafkaJsonProducer; +import org.softwarecave.springbootnote.outbox.model.AggregateType; +import org.softwarecave.springbootnote.outbox.model.MessageType; +import org.softwarecave.springbootnote.outbox.model.Outbox; +import org.softwarecave.springbootnote.outbox.model.Status; + +import java.time.ZonedDateTime; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Future; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class OutboxDispatcherJsonStrategyTest { + + @Mock + private KafkaJsonProducer kafkaJsonProducer; + + @InjectMocks + private OutboxDispatcherJsonStrategy strategy; + + @Test + void testGetMessageType() { + assertThat(strategy.getMessageType()).isEqualTo(MessageType.JSON); + } + + @Test + void testSendCallsKafkaJsonProducer() { + Outbox outbox = createOutbox(); + + Future mockFuture = CompletableFuture.completedFuture(mock(RecordMetadata.class)); + when(kafkaJsonProducer.sendToKafka(any(Outbox.class))).thenReturn(mockFuture); + + strategy.send(outbox); + + verify(kafkaJsonProducer).sendToKafka(any(Outbox.class)); + } + + @Test + void testMultipleSendOperations() { + Outbox outbox1 = createOutbox(); + Outbox outbox2 = createOutbox(); + outbox2.setId(2L); + + Future mockFuture1 = CompletableFuture.completedFuture(mock(RecordMetadata.class)); + Future mockFuture2 = CompletableFuture.completedFuture(mock(RecordMetadata.class)); + + when(kafkaJsonProducer.sendToKafka(outbox1)).thenReturn(mockFuture1); + when(kafkaJsonProducer.sendToKafka(outbox2)).thenReturn(mockFuture2); + + Future result1 = strategy.send(outbox1); + Future result2 = strategy.send(outbox2); + + assertThat(result1).isEqualTo(mockFuture1); + assertThat(result2).isEqualTo(mockFuture2); + verify(kafkaJsonProducer).sendToKafka(outbox1); + verify(kafkaJsonProducer).sendToKafka(outbox2); + } + + + @Test + void testSendWithDifferentPayloads() { + Outbox outboxWithJsonPayload = createOutbox(); + outboxWithJsonPayload.setPayloadString("{\"key\": \"value\"}"); + + Outbox outboxWithComplexPayload = createOutbox(); + outboxWithComplexPayload.setPayloadString("{\"nested\": {\"data\": [1, 2, 3]}}"); + + Future mockFuture = CompletableFuture.completedFuture(mock(RecordMetadata.class)); + when(kafkaJsonProducer.sendToKafka(any(Outbox.class))).thenReturn(mockFuture); + + strategy.send(outboxWithJsonPayload); + strategy.send(outboxWithComplexPayload); + + verify(kafkaJsonProducer).sendToKafka(outboxWithJsonPayload); + verify(kafkaJsonProducer).sendToKafka(outboxWithComplexPayload); + } + + private Outbox createOutbox() { + var outbox = new Outbox(); + outbox.setId(1L); + outbox.setAggregateType(AggregateType.STICKY_NOTE); + outbox.setAggregateId(100L); + outbox.setMessageType(MessageType.JSON); + outbox.setStatus(Status.NEW); + outbox.setCreatedDate(ZonedDateTime.now()); + outbox.setPayloadString("{\"test\": \"data\"}"); + return outbox; + } +} diff --git a/springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherTest.java b/springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherTest.java new file mode 100644 index 0000000..3380305 --- /dev/null +++ b/springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherTest.java @@ -0,0 +1,185 @@ +package org.softwarecave.springbootnote.outbox.service.dispatch; + +import org.apache.kafka.clients.producer.RecordMetadata; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.softwarecave.springbootnote.outbox.model.AggregateType; +import org.softwarecave.springbootnote.outbox.model.MessageType; +import org.softwarecave.springbootnote.outbox.model.Outbox; +import org.softwarecave.springbootnote.outbox.model.Status; +import org.softwarecave.springbootnote.outbox.service.InvalidOutboxDataException; +import org.softwarecave.springbootnote.outbox.service.OutboxRepository; +import org.springframework.data.domain.PageImpl; + +import java.time.ZonedDateTime; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class OutboxDispatcherTest { + + @Mock + private OutboxRepository outboxRepository; + + @Mock + private OutboxDispatcherStrategy jsonStrategy; + + @Mock + private OutboxDispatcherStrategy avroStrategy; + + @InjectMocks + private OutboxDispatcher outboxDispatcher; + + @BeforeEach + void setUp() { + // Marked as lenient because there is no point repeating parts of this code in every test method + lenient().when(jsonStrategy.getMessageType()).thenReturn(MessageType.JSON); + lenient().when(avroStrategy.getMessageType()).thenReturn(MessageType.AVRO); + } + + @Test + void testSetDispatcherStrategies_ConflictingMessageTypes() { + var conflictingStrategy = mock(OutboxDispatcherStrategy.class); + when(conflictingStrategy.getMessageType()).thenReturn(MessageType.JSON); + + assertThatThrownBy(() -> outboxDispatcher.setDispatcherStrategies(List.of(jsonStrategy, conflictingStrategy))) + .isInstanceOf(InvalidOutboxDataException.class) + .hasMessageContaining("conflicting Outbox dispatchers strategies"); + } + + @Test + void testProcessWithNewOutboxEntries() throws Exception { + // given + outboxDispatcher.setDispatcherStrategies(List.of(jsonStrategy, avroStrategy)); + + var outbox1 = createOutbox(1L, MessageType.JSON); + var outbox2 = createOutbox(2L, MessageType.AVRO); + + when(outboxRepository.findByStatus(eq(Status.NEW), any())) + .thenReturn(new PageImpl(List.of(outbox1, outbox2))); + + var future1 = CompletableFuture.completedFuture(mock(RecordMetadata.class)); + var future2 = CompletableFuture.completedFuture(mock(RecordMetadata.class)); + + when(jsonStrategy.send(outbox1)).thenReturn(future1); + when(avroStrategy.send(outbox2)).thenReturn(future2); + + // when + outboxDispatcher.process(); + + // then + var captor = ArgumentCaptor.forClass(Outbox.class); + verify(outboxRepository, times(2)).save(captor.capture()); + + List savedOutboxes = captor.getAllValues(); + assertThat(savedOutboxes).hasSize(2); + assertThat(savedOutboxes).allMatch(outbox -> outbox.getStatus() == Status.SENT); + } + + @Test + void testProcessWithEmptyOutboxEntries() { + // given + outboxDispatcher.setDispatcherStrategies(List.of(jsonStrategy, avroStrategy)); + + when(outboxRepository.findByStatus(eq(Status.NEW), any())) + .thenReturn(new PageImpl(List.of())); + + // when + outboxDispatcher.process(); + + // then + verify(outboxRepository, never()).save(any()); + } + + @Test + void testProcessWithKafkaFailure() throws Exception { + // given + outboxDispatcher.setDispatcherStrategies(List.of(jsonStrategy)); + + var outbox = createOutbox(1L, MessageType.JSON); + + when(outboxRepository.findByStatus(eq(Status.NEW), any())) + .thenReturn(new PageImpl(List.of(outbox))); + + var future = CompletableFuture.failedFuture(new RuntimeException("Kafka error")); + when(jsonStrategy.send(outbox)).thenReturn(future); + + // when + outboxDispatcher.process(); + + // then + verify(outboxRepository, never()).save(any()); + } + + @Test + void testSendToKafkaWithUnrecognizedMessageType() { + outboxDispatcher.setDispatcherStrategies(List.of(jsonStrategy)); + + var outbox = createOutbox(1L, MessageType.AVRO); + + assertThatThrownBy(() -> { + when(outboxRepository.findByStatus(eq(Status.NEW), any())) + .thenReturn(new PageImpl(List.of(outbox))); + outboxDispatcher.process(); + }).isInstanceOf(InvalidOutboxDataException.class) + .hasMessageContaining("Unrecognized message type"); + } + + @Test + void testProcessMultipleEntriesWithMixedResults() throws Exception { + // given + outboxDispatcher.setDispatcherStrategies(List.of(jsonStrategy, avroStrategy)); + + var outbox1 = createOutbox(1L, MessageType.JSON); + var outbox2 = createOutbox(2L, MessageType.JSON); + + when(outboxRepository.findByStatus(eq(Status.NEW), any())) + .thenReturn(new PageImpl(List.of(outbox1, outbox2))); + + var future1 = CompletableFuture.completedFuture(mock(RecordMetadata.class)); + when(jsonStrategy.send(outbox1)).thenReturn(future1); + + var future2 = CompletableFuture.failedFuture(new RuntimeException("Kafka error")); + when(jsonStrategy.send(outbox2)).thenReturn(future2); + + // when + outboxDispatcher.process(); + + // then + var captor = ArgumentCaptor.forClass(Outbox.class); + verify(outboxRepository).save(captor.capture()); + + List savedOutboxes = captor.getAllValues(); + assertThat(savedOutboxes).hasSize(1); + assertThat(savedOutboxes.getFirst().getId()).isEqualTo(1L); + assertThat(savedOutboxes.getFirst().getStatus()).isEqualTo(Status.SENT); + } + + private Outbox createOutbox(Long id, MessageType messageType) { + var outbox = new Outbox(); + outbox.setId(id); + outbox.setAggregateType(AggregateType.STICKY_NOTE); + outbox.setAggregateId(100L); + outbox.setMessageType(messageType); + outbox.setStatus(Status.NEW); + outbox.setCreatedDate(ZonedDateTime.now()); + outbox.setPayloadString("{\"test\": \"data\"}"); + return outbox; + } +} diff --git a/springboot-note/src/test/resources/application.yaml b/springboot-note/src/test/resources/application.yaml index b6b8e1c..d61277c 100644 --- a/springboot-note/src/test/resources/application.yaml +++ b/springboot-note/src/test/resources/application.yaml @@ -11,3 +11,8 @@ spring: ddl-auto: create-drop flyway: enabled: false + +app: + outbox: + sender: + delay: 10000000