From ff78737e40f4957c1f8caed791f4e762bf8de5db Mon Sep 17 00:00:00 2001 From: Robert Piasecki Date: Wed, 15 Jul 2026 10:22:37 +0100 Subject: [PATCH 1/4] Add outbox when sending data through Kafka. #34 --- .../SpringbootNoteApplication.java | 2 + .../StickyNoteRecorderAspect.java | 17 ++-- .../kafka/KafkaStickyNoteProducer.java | 7 -- .../kafka/KafkaAvroProducer.java} | 56 +++++------ .../kafka/KafkaJsonProducer.java} | 36 +++----- .../converter/AvroStickyNoteConverter.java | 18 +++- .../outbox/model/AggregateType.java | 5 + .../outbox/model/MessageType.java | 6 ++ .../springbootnote/outbox/model/Outbox.java | 58 ++++++++++++ .../springbootnote/outbox/model/Status.java | 6 ++ .../outbox/service/AvroOutboxService.java | 39 ++++++++ .../service/InvalidOutboxDataException.java | 7 ++ .../outbox/service/JsonOutboxService.java | 43 +++++++++ .../outbox/service/OutboxRepository.java | 11 +++ .../outbox/service/OutboxService.java | 7 ++ .../outbox/service/sender/OutboxSender.java | 92 +++++++++++++++++++ .../src/main/resources/application.yaml | 3 + .../db/migration/V1.3__AddOutbox.sql | 13 +++ .../AvroStickyNoteConverterTest.java | 1 + 19 files changed, 361 insertions(+), 66 deletions(-) delete mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/KafkaStickyNoteProducer.java rename springboot-note/src/main/java/org/softwarecave/springbootnote/{notification/kafka/AvroKafkaStickyNoteProducer.java => outbox/kafka/KafkaAvroProducer.java} (55%) rename springboot-note/src/main/java/org/softwarecave/springbootnote/{notification/kafka/JsonKafkaStickyNoteProducer.java => outbox/kafka/KafkaJsonProducer.java} (60%) rename springboot-note/src/main/java/org/softwarecave/springbootnote/{notification => outbox}/kafka/converter/AvroStickyNoteConverter.java (77%) create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/AggregateType.java create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/MessageType.java create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/Outbox.java create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/Status.java create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/AvroOutboxService.java create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/InvalidOutboxDataException.java create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/JsonOutboxService.java create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxRepository.java create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxService.java create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/sender/OutboxSender.java create mode 100644 springboot-note/src/main/resources/db/migration/V1.3__AddOutbox.sql diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/SpringbootNoteApplication.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/SpringbootNoteApplication.java index cce2309..408e58b 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/SpringbootNoteApplication.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/SpringbootNoteApplication.java @@ -2,8 +2,10 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication +@EnableScheduling public class SpringbootNoteApplication { public static void main(String[] args) { 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..74da8b7 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.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") @@ -32,15 +34,16 @@ public void addStickyNote(JoinPoint joinPoint, Recordable rec, Object returnValu } - if (kafkaStickyNoteProducers != null && !kafkaStickyNoteProducers.isEmpty()) { - kafkaStickyNoteProducers.forEach(e -> e.sendToKafka((StickyNote) returnValue)); + if (outboxServices != null && !outboxServices.isEmpty()) { + StickyNote stickyNote = (StickyNote) returnValue; + 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 55% 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..f15a91b 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,56 @@ -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.io.BinaryDecoder; +import org.apache.avro.io.DatumReader; +import org.apache.avro.io.DecoderFactory; +import org.apache.avro.specific.SpecificDatumReader; +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; @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 +65,20 @@ 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); + } } - + 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/java/org/softwarecave/springbootnote/notification/kafka/JsonKafkaStickyNoteProducer.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaJsonProducer.java similarity index 60% 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..e40ed8b 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; @@ -12,34 +12,31 @@ 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 +59,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 77% 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..3f7124f 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,30 @@ -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.io.IOException; +import java.nio.ByteBuffer; import java.time.ZoneOffset; import java.util.List; +@Component @Slf4j public class AvroStickyNoteConverter { + + public ByteBuffer convertToByteBuffer(StickyNote value) { + try { + var stickyNoteAvro = convertToAvro(value); + return stickyNoteAvro.toByteBuffer(); + } catch (IOException e) { + // TODO: + log.error("Failed to convert Sticky Note Avro object to ByteBuffer"); + throw new IllegalArgumentException("Failed to convert Sticky Note Avro object to ByteBuffer"); + } + } + 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..8afbe20 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/AggregateType.java @@ -0,0 +1,5 @@ +package org.softwarecave.springbootnote.outbox.model; + +public enum AggregateType { + STICKY_NOTE +} 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..327642a --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/model/Outbox.java @@ -0,0 +1,58 @@ +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.Lob; +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/AvroOutboxService.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/AvroOutboxService.java new file mode 100644 index 0000000..261ebe6 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/AvroOutboxService.java @@ -0,0 +1,39 @@ +package org.softwarecave.springbootnote.outbox.service; + +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.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.nio.ByteBuffer; +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) { + ByteBuffer payload = avroStickyNoteConverter.convertToByteBuffer(stickyNote); + byte[] payloadBinary = payload.array(); + + Outbox outbox = new Outbox(null, AggregateType.STICKY_NOTE, stickyNote.getId(), + 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/InvalidOutboxDataException.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/InvalidOutboxDataException.java new file mode 100644 index 0000000..b737e5b --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/InvalidOutboxDataException.java @@ -0,0 +1,7 @@ +package org.softwarecave.springbootnote.outbox.service; + +public class InvalidOutboxDataException extends RuntimeException { + public InvalidOutboxDataException(String message) { + super(message); + } +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/JsonOutboxService.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/JsonOutboxService.java new file mode 100644 index 0000000..2d45568 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/JsonOutboxService.java @@ -0,0 +1,43 @@ +package org.softwarecave.springbootnote.outbox.service; + +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.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); + + Outbox outbox = new Outbox(null, AggregateType.STICKY_NOTE, stickyNote.getId(), + 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/OutboxRepository.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxRepository.java new file mode 100644 index 0000000..5b70bdd --- /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.PageRequest; +import org.springframework.data.jpa.repository.JpaRepository; + +public interface OutboxRepository extends JpaRepository { + Page findByStatus(Status status, PageRequest createdDate); +} diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxService.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxService.java new file mode 100644 index 0000000..398f69b --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxService.java @@ -0,0 +1,7 @@ +package org.softwarecave.springbootnote.outbox.service; + +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/service/sender/OutboxSender.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/sender/OutboxSender.java new file mode 100644 index 0000000..939f413 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/sender/OutboxSender.java @@ -0,0 +1,92 @@ +package org.softwarecave.springbootnote.outbox.service.sender; + +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.avro.StickyNote; +import org.softwarecave.springbootnote.outbox.kafka.KafkaAvroProducer; +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 org.softwarecave.springbootnote.outbox.service.InvalidOutboxDataException; +import org.softwarecave.springbootnote.outbox.service.OutboxRepository; +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.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +@Service +@RequiredArgsConstructor +@Slf4j +public class OutboxSender { + + private final OutboxRepository outboxRepository; + private final KafkaJsonProducer kafkaJsonProducer; + private final KafkaAvroProducer kafkaAvroProducer; + + @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(); + return switch (messageType) { + case AVRO -> kafkaAvroProducer.sendToKafka(entry, getAvroClass(entry)); + case JSON -> kafkaJsonProducer.sendToKafka(entry); + case null -> throw new InvalidOutboxDataException("Unrecognized message type " + messageType); + }; + } + + private Class getAvroClass(Outbox entry) { + AggregateType aggregateType = entry.getAggregateType(); + return switch (aggregateType) { + case STICKY_NOTE -> StickyNote.class; + case null -> throw new InvalidOutboxDataException("Unrecognized aggregate type " + aggregateType); + }; + } +} 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/notification/kafka/converter/AvroStickyNoteConverterTest.java b/springboot-note/src/test/java/org/softwarecave/springbootnote/notification/kafka/converter/AvroStickyNoteConverterTest.java index 316ef54..7eb81da 100644 --- a/springboot-note/src/test/java/org/softwarecave/springbootnote/notification/kafka/converter/AvroStickyNoteConverterTest.java +++ b/springboot-note/src/test/java/org/softwarecave/springbootnote/notification/kafka/converter/AvroStickyNoteConverterTest.java @@ -6,6 +6,7 @@ import org.softwarecave.springbootnote.note.model.StickyNoteLink; import org.softwarecave.springbootnote.note.model.StickyNoteTag; import org.softwarecave.springbootnote.note.model.Type; +import org.softwarecave.springbootnote.outbox.kafka.converter.AvroStickyNoteConverter; import org.softwarecave.springbootnote.tag.model.Tag; import java.time.LocalDateTime; From 4b3fb8b7fa80b288e635e77e64d99e2c123ace51 Mon Sep 17 00:00:00 2001 From: Robert Piasecki Date: Wed, 15 Jul 2026 12:10:04 +0200 Subject: [PATCH 2/4] Separated OutboxDispatcher strategies. #34 --- .../StickyNoteRecorderAspect.java | 2 +- .../outbox/kafka/KafkaAvroProducer.java | 7 +-- .../converter/AvroStickyNoteConverter.java | 11 ----- .../service/InvalidOutboxDataException.java | 4 ++ .../OutboxDispatcher.java} | 47 ++++++++++--------- .../OutboxDispatcherAvroStrategy.java | 40 ++++++++++++++++ .../OutboxDispatcherJsonStrategy.java | 28 +++++++++++ .../dispatch/OutboxDispatcherStrategy.java | 13 +++++ .../{ => queue}/AvroOutboxService.java | 9 ++-- .../{ => queue}/JsonOutboxService.java | 3 +- .../service/{ => queue}/OutboxService.java | 2 +- .../outbox/tools/AvroTools.java | 41 ++++++++++++++++ 12 files changed, 161 insertions(+), 46 deletions(-) rename springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/{sender/OutboxSender.java => dispatch/OutboxDispatcher.java} (67%) create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherAvroStrategy.java create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherJsonStrategy.java create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherStrategy.java rename springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/{ => queue}/AvroOutboxService.java (79%) rename springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/{ => queue}/JsonOutboxService.java (92%) rename springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/{ => queue}/OutboxService.java (68%) create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/tools/AvroTools.java 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 74da8b7..e491c10 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 @@ -6,7 +6,7 @@ import org.aspectj.lang.annotation.AfterReturning; import org.aspectj.lang.annotation.Aspect; import org.softwarecave.springbootnote.note.model.StickyNote; -import org.softwarecave.springbootnote.outbox.service.OutboxService; +import org.softwarecave.springbootnote.outbox.service.queue.OutboxService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaAvroProducer.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaAvroProducer.java index f15a91b..28c6b13 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaAvroProducer.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaAvroProducer.java @@ -23,6 +23,8 @@ 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) @@ -75,10 +77,5 @@ public Future sendToKafka(Outbox valu } } - 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/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverter.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverter.java index 3f7124f..039b3c0 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverter.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverter.java @@ -14,17 +14,6 @@ @Slf4j public class AvroStickyNoteConverter { - public ByteBuffer convertToByteBuffer(StickyNote value) { - try { - var stickyNoteAvro = convertToAvro(value); - return stickyNoteAvro.toByteBuffer(); - } catch (IOException e) { - // TODO: - log.error("Failed to convert Sticky Note Avro object to ByteBuffer"); - throw new IllegalArgumentException("Failed to convert Sticky Note Avro object to ByteBuffer"); - } - } - 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/service/InvalidOutboxDataException.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/InvalidOutboxDataException.java index b737e5b..0a4898d 100644 --- 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 @@ -4,4 +4,8 @@ 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/sender/OutboxSender.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcher.java similarity index 67% rename from springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/sender/OutboxSender.java rename to springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcher.java index 939f413..445a3e2 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/sender/OutboxSender.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcher.java @@ -1,13 +1,7 @@ -package org.softwarecave.springbootnote.outbox.service.sender; +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.avro.StickyNote; -import org.softwarecave.springbootnote.outbox.kafka.KafkaAvroProducer; -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; @@ -20,18 +14,31 @@ 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 -@RequiredArgsConstructor @Slf4j -public class OutboxSender { +public class OutboxDispatcher { private final OutboxRepository outboxRepository; - private final KafkaJsonProducer kafkaJsonProducer; - private final KafkaAvroProducer kafkaAvroProducer; + private final Map dispatcherStrategies; + + public OutboxDispatcher(OutboxRepository outboxRepository, + List dispatcherStrategies) { + this.outboxRepository = outboxRepository; + this.dispatcherStrategies = new HashMap<>(); + 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() { @@ -75,18 +82,12 @@ private ArrayList> sendToKafka(Page entryList) { private Future sendToKafka(Outbox entry) { MessageType messageType = entry.getMessageType(); - return switch (messageType) { - case AVRO -> kafkaAvroProducer.sendToKafka(entry, getAvroClass(entry)); - case JSON -> kafkaJsonProducer.sendToKafka(entry); - case null -> throw new InvalidOutboxDataException("Unrecognized message type " + messageType); - }; + var dispatcherStrategy = dispatcherStrategies.get(messageType); + if (dispatcherStrategy != null) { + return dispatcherStrategy.send(entry); + } else { + throw new InvalidOutboxDataException("Unrecognized message type " + messageType); + } } - private Class getAvroClass(Outbox entry) { - AggregateType aggregateType = entry.getAggregateType(); - return switch (aggregateType) { - case STICKY_NOTE -> StickyNote.class; - case null -> throw new InvalidOutboxDataException("Unrecognized aggregate type " + aggregateType); - }; - } } 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..cfb3624 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherAvroStrategy.java @@ -0,0 +1,40 @@ +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.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.service.InvalidOutboxDataException; +import org.springframework.stereotype.Service; + +import java.util.concurrent.Future; + +@RequiredArgsConstructor +@Service +@Slf4j +public class OutboxDispatcherAvroStrategy implements OutboxDispatcherStrategy { + private final 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(); + return switch (aggregateType) { + case STICKY_NOTE -> StickyNote.class; // TODO: encapsulate this + case null -> 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..c42d0c3 --- /dev/null +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherJsonStrategy.java @@ -0,0 +1,28 @@ +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.stereotype.Service; + +import java.util.concurrent.Future; + +@RequiredArgsConstructor +@Service +@Slf4j +public class OutboxDispatcherJsonStrategy implements OutboxDispatcherStrategy { + private final 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/AvroOutboxService.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/AvroOutboxService.java similarity index 79% rename from springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/AvroOutboxService.java rename to springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/AvroOutboxService.java index 261ebe6..8fea422 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/AvroOutboxService.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/AvroOutboxService.java @@ -1,4 +1,4 @@ -package org.softwarecave.springbootnote.outbox.service; +package org.softwarecave.springbootnote.outbox.service.queue; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -8,11 +8,12 @@ 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.nio.ByteBuffer; import java.time.ZonedDateTime; @Service @@ -27,8 +28,8 @@ public class AvroOutboxService implements OutboxService { @Transactional @Override public void send(StickyNote stickyNote) { - ByteBuffer payload = avroStickyNoteConverter.convertToByteBuffer(stickyNote); - byte[] payloadBinary = payload.array(); + var avroStickyNote = avroStickyNoteConverter.convertToAvro(stickyNote); + byte[] payloadBinary = AvroTools.convertToBytes(avroStickyNote); Outbox outbox = new Outbox(null, AggregateType.STICKY_NOTE, stickyNote.getId(), MessageType.AVRO, ZonedDateTime.now(), payloadBinary, null, Status.NEW); diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/JsonOutboxService.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/JsonOutboxService.java similarity index 92% rename from springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/JsonOutboxService.java rename to springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/JsonOutboxService.java index 2d45568..91bd3a5 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/JsonOutboxService.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/JsonOutboxService.java @@ -1,4 +1,4 @@ -package org.softwarecave.springbootnote.outbox.service; +package org.softwarecave.springbootnote.outbox.service.queue; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -9,6 +9,7 @@ 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; diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxService.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/OutboxService.java similarity index 68% rename from springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxService.java rename to springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/OutboxService.java index 398f69b..089327f 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxService.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/queue/OutboxService.java @@ -1,4 +1,4 @@ -package org.softwarecave.springbootnote.outbox.service; +package org.softwarecave.springbootnote.outbox.service.queue; import org.softwarecave.springbootnote.note.model.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); + } +} From f30601f96124617848fc4f3e21944739a5e913f5 Mon Sep 17 00:00:00 2001 From: Robert Piasecki Date: Wed, 15 Jul 2026 19:42:50 +0200 Subject: [PATCH 3/4] Corrected Avro serialization to byte array before saving into outbox. Fixed tests. #34 --- README.md | 1 + .../SpringbootNoteApplication.java | 1 - .../config/outbox/SchedulerConfig.java | 9 + .../StickyNoteRecorderAspect.java | 4 +- .../outbox/model/AggregateType.java | 14 +- .../outbox/service/OutboxRepository.java | 3 +- .../service/dispatch/OutboxDispatcher.java | 9 +- .../OutboxDispatcherAvroStrategy.java | 17 +- .../OutboxDispatcherJsonStrategy.java | 9 +- .../service/queue/AvroOutboxService.java | 6 +- .../service/queue/JsonOutboxService.java | 7 +- .../AvroStickyNoteConverterTest.java | 3 +- .../OutboxDispatcherAvroStrategyTest.java | 92 +++++++++ .../OutboxDispatcherJsonStrategyTest.java | 102 ++++++++++ .../dispatch/OutboxDispatcherTest.java | 185 ++++++++++++++++++ .../src/test/resources/application.yaml | 5 + 16 files changed, 450 insertions(+), 17 deletions(-) create mode 100644 springboot-note/src/main/java/org/softwarecave/springbootnote/config/outbox/SchedulerConfig.java rename springboot-note/src/test/java/org/softwarecave/springbootnote/{notification => outbox}/kafka/converter/AvroStickyNoteConverterTest.java (96%) create mode 100644 springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherAvroStrategyTest.java create mode 100644 springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherJsonStrategyTest.java create mode 100644 springboot-note/src/test/java/org/softwarecave/springbootnote/outbox/service/dispatch/OutboxDispatcherTest.java 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/SpringbootNoteApplication.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/SpringbootNoteApplication.java index 408e58b..c60d3e8 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/SpringbootNoteApplication.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/SpringbootNoteApplication.java @@ -5,7 +5,6 @@ import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication -@EnableScheduling public class SpringbootNoteApplication { public static void main(String[] args) { 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 e491c10..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 @@ -33,9 +33,11 @@ public void addStickyNote(JoinPoint joinPoint, Recordable rec, Object returnValu } } + sendToOutbox((StickyNote) returnValue); + } + private void sendToOutbox(StickyNote stickyNote) { if (outboxServices != null && !outboxServices.isEmpty()) { - StickyNote stickyNote = (StickyNote) returnValue; outboxServices.forEach(outboxService -> outboxService.send(stickyNote)); } else { log.warn("No outbox services are available, skipping Kafka notification"); 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 index 8afbe20..7a0e073 100644 --- 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 @@ -1,5 +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 + 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/service/OutboxRepository.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/service/OutboxRepository.java index 5b70bdd..d0e6e0a 100644 --- 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 @@ -4,8 +4,9 @@ import org.softwarecave.springbootnote.outbox.model.Status; import org.springframework.data.domain.Page; import org.springframework.data.domain.PageRequest; +import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; public interface OutboxRepository extends JpaRepository { - Page findByStatus(Status status, PageRequest createdDate); + 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 index 445a3e2..1ef72a1 100644 --- 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 @@ -7,6 +7,7 @@ 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; @@ -28,10 +29,14 @@ public class OutboxDispatcher { private final OutboxRepository outboxRepository; private final Map dispatcherStrategies; - public OutboxDispatcher(OutboxRepository outboxRepository, - List 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) { 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 index cfb3624..979fc5d 100644 --- 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 @@ -10,6 +10,7 @@ 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; @@ -18,7 +19,12 @@ @Service @Slf4j public class OutboxDispatcherAvroStrategy implements OutboxDispatcherStrategy { - private final KafkaAvroProducer kafkaAvroProducer; + private KafkaAvroProducer kafkaAvroProducer; + + @Autowired(required = false) + public void setKafkaAvroProducer(KafkaAvroProducer kafkaAvroProducer) { + this.kafkaAvroProducer = kafkaAvroProducer; + } @Override public Future send(Outbox outbox) { @@ -32,9 +38,10 @@ public MessageType getMessageType() { private Class getAvroClass(Outbox entry) { AggregateType aggregateType = entry.getAggregateType(); - return switch (aggregateType) { - case STICKY_NOTE -> StickyNote.class; // TODO: encapsulate this - case null -> throw new InvalidOutboxDataException("Unrecognized aggregate type " + aggregateType); - }; + 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 index c42d0c3..3000bcd 100644 --- 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 @@ -6,6 +6,7 @@ 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; @@ -14,7 +15,13 @@ @Service @Slf4j public class OutboxDispatcherJsonStrategy implements OutboxDispatcherStrategy { - private final KafkaJsonProducer kafkaJsonProducer; + + private KafkaJsonProducer kafkaJsonProducer; + + @Autowired(required = false) + public void setKafkaJsonProducer(KafkaJsonProducer kafkaJsonProducer) { + this.kafkaJsonProducer = kafkaJsonProducer; + } @Override public Future send(Outbox outbox) { 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 index 8fea422..28706d2 100644 --- 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 @@ -31,7 +31,11 @@ public void send(StickyNote stickyNote) { var avroStickyNote = avroStickyNoteConverter.convertToAvro(stickyNote); byte[] payloadBinary = AvroTools.convertToBytes(avroStickyNote); - Outbox outbox = new Outbox(null, AggregateType.STICKY_NOTE, stickyNote.getId(), + 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); 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 index 91bd3a5..1d09bbd 100644 --- 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 @@ -30,11 +30,14 @@ public class JsonOutboxService implements OutboxService { @Transactional @Override public void send(StickyNote stickyNote) { - StickyNoteDTO dto = stickyNoteConverter.convertToDTO(stickyNote); String payloadString = jsonMapper.writeValueAsString(dto); - Outbox outbox = new Outbox(null, AggregateType.STICKY_NOTE, stickyNote.getId(), + 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); 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 96% 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 7eb81da..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; @@ -6,7 +6,6 @@ import org.softwarecave.springbootnote.note.model.StickyNoteLink; import org.softwarecave.springbootnote.note.model.StickyNoteTag; import org.softwarecave.springbootnote.note.model.Type; -import org.softwarecave.springbootnote.outbox.kafka.converter.AvroStickyNoteConverter; import org.softwarecave.springbootnote.tag.model.Tag; import java.time.LocalDateTime; 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 From bd2851e45a11569ad737b65e90459af600f31578 Mon Sep 17 00:00:00 2001 From: Robert Piasecki Date: Fri, 17 Jul 2026 08:02:18 +0200 Subject: [PATCH 4/4] Cleanup. #34 --- .../springbootnote/SpringbootNoteApplication.java | 1 - .../springbootnote/outbox/kafka/KafkaAvroProducer.java | 4 ---- .../springbootnote/outbox/kafka/KafkaJsonProducer.java | 9 +++------ .../outbox/kafka/converter/AvroStickyNoteConverter.java | 2 -- .../softwarecave/springbootnote/outbox/model/Outbox.java | 1 - .../springbootnote/outbox/service/OutboxRepository.java | 1 - .../service/dispatch/OutboxDispatcherAvroStrategy.java | 1 - .../springbootnote/SpringbootNoteApplicationTests.java | 6 +++--- .../note/StickyNoteServiceIntegrationTest.java | 1 - .../note/web/StickyNoteControllerTest.java | 1 - 10 files changed, 6 insertions(+), 21 deletions(-) diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/SpringbootNoteApplication.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/SpringbootNoteApplication.java index c60d3e8..cce2309 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/SpringbootNoteApplication.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/SpringbootNoteApplication.java @@ -2,7 +2,6 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication public class SpringbootNoteApplication { diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaAvroProducer.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaAvroProducer.java index 28c6b13..48a60d8 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaAvroProducer.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaAvroProducer.java @@ -4,10 +4,6 @@ import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; -import org.apache.avro.io.BinaryDecoder; -import org.apache.avro.io.DatumReader; -import org.apache.avro.io.DecoderFactory; -import org.apache.avro.specific.SpecificDatumReader; import org.apache.avro.specific.SpecificRecord; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaJsonProducer.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaJsonProducer.java index e40ed8b..5cbd4d7 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaJsonProducer.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/KafkaJsonProducer.java @@ -9,14 +9,11 @@ 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; @@ -32,9 +29,9 @@ public class KafkaJsonProducer { private KafkaProducer kafkaProducer; public KafkaJsonProducer( - StickyNoteConverter stickyNoteConverter, - @Value("${app.kafka.bootstrap-servers}") String bootstrapServers, - @Value("${app.kafka.json.stickynote-topic}") String noteTopic) { + StickyNoteConverter stickyNoteConverter, + @Value("${app.kafka.bootstrap-servers}") String bootstrapServers, + @Value("${app.kafka.json.stickynote-topic}") String noteTopic) { this.bootstrapServers = bootstrapServers; this.noteTopic = noteTopic; } diff --git a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverter.java b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverter.java index 039b3c0..1564344 100644 --- a/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverter.java +++ b/springboot-note/src/main/java/org/softwarecave/springbootnote/outbox/kafka/converter/AvroStickyNoteConverter.java @@ -5,8 +5,6 @@ import org.softwarecave.springbootnote.tag.model.Tag; import org.springframework.stereotype.Component; -import java.io.IOException; -import java.nio.ByteBuffer; import java.time.ZoneOffset; import java.util.List; 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 index 327642a..3190e3b 100644 --- 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 @@ -7,7 +7,6 @@ import jakarta.persistence.GeneratedValue; import jakarta.persistence.GenerationType; import jakarta.persistence.Id; -import jakarta.persistence.Lob; import jakarta.persistence.SequenceGenerator; import jakarta.persistence.Table; import lombok.AllArgsConstructor; 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 index d0e6e0a..7759a43 100644 --- 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 @@ -3,7 +3,6 @@ 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.PageRequest; import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; 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 index 979fc5d..510731b 100644 --- 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 @@ -4,7 +4,6 @@ import lombok.extern.slf4j.Slf4j; import org.apache.avro.specific.SpecificRecord; import org.apache.kafka.clients.producer.RecordMetadata; -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; 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;