Add outbox when sending data through Kafka. #34

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)

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 {
+}

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<KafkaStickyNoteProducer> kafkaStickyNoteProducers;
+    private List<OutboxService> 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<KafkaStickyNoteProducer> kafkaStickyNoteProducers) {
-        this.kafkaStickyNoteProducers = kafkaStickyNoteProducers;
+    public void setOutboxService(List<OutboxService> outboxServices) {
+        this.outboxServices = outboxServices;
     }
 }

springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/AvroKafkaStickyNoteProducer.java → 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<Long, org.softwarecave.springbootnote.avro.StickyNote> kafkaProducer;
+    private KafkaProducer<Long, SpecificRecord> 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 <T extends SpecificRecord> Future<RecordMetadata> sendToKafka(Outbox value, Class<T> avroClass) {
+        try {
+            var avroObject = fromBytes(value.getPayloadBytes(), avroClass);
+            var producerRecord = new ProducerRecord<Long, SpecificRecord>(noteTopic, value.getAggregateId(), avroObject);
+            return kafkaProducer.send(producerRecord);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
     }
 
 
-
 }

springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/JsonKafkaStickyNoteProducer.java → 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<Long, String> 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<RecordMetadata> sendToKafka(Outbox value) {
+        var producerRecord = new ProducerRecord<>(noteTopic, value.getAggregateId(), value.getPayloadString());
+        return kafkaProducer.send(producerRecord);
     }
 
 }

springboot-note/src/main/java/org/softwarecave/springbootnote/notification/kafka/converter/AvroStickyNoteConverter.java → 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);

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<? extends SpecificRecord> avroClass;
+
+    AggregateType(Class<? extends SpecificRecord> avroClass) {
+        this.avroClass = avroClass;
+    }
+
+}

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
+}

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;
+}

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
+}

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);
+    }
+}

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<Outbox, Long> {
+    Page<Outbox> findByStatus(Status status, Pageable pageable);
+}