Added Mapstruct changes

build.gradle

+import com.commercehub.gradle.plugin.avro.GenerateAvroJavaTask
 plugins {
 	id 'org.springframework.boot' version '2.2.6.RELEASE'
 	id 'io.spring.dependency-management' version '1.0.9.RELEASE'
 	id 'java'
+	id 'com.commercehub.gradle.plugin.avro' version '0.9.1'
 }
 
 group = 'com.safeway.epe'
@@ -19,15 +21,26 @@ configurations {
 }
 
 repositories {
+	gradlePluginPortal()
 	mavenCentral()
+	maven { url "http://packages.confluent.io/maven/" }
+	maven { url "https://plugins.gradle.org/m2/"}
 }
 
 dependencies {
 	implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
 	implementation 'org.springframework.boot:spring-boot-starter-web'
+	implementation 'org.mapstruct:mapstruct:1.3.1.Final'
+	implementation 'io.confluent:kafka-avro-serializer:5.4.0'
+	compile (group: 'io.confluent', name: 'kafka-schema-registry', version: '5.4.0'){
+		exclude group:'org.slf4j', module: 'slf4j-log4j12'
+	}
+	compile "org.apache.avro:avro:1.9.1"
+	compile group: 'org.apache.kafka', name: 'kafka-clients', version: '5.4.1-ccs'
 	compileOnly 'org.projectlombok:lombok'
 	developmentOnly 'org.springframework.boot:spring-boot-devtools'
 	runtimeOnly 'org.postgresql:postgresql'
+	annotationProcessor 'org.mapstruct:mapstruct-processor:1.3.1.Final'
 	annotationProcessor 'org.projectlombok:lombok'
 	compile 'com.vladmihalcea:hibernate-types-52:2.0.0'
 	testImplementation('org.springframework.boot:spring-boot-starter-test') {
@@ -38,3 +51,21 @@ dependencies {
 test {
 	useJUnitPlatform()
 }
+avro {
+	createSetters = false
+	fieldVisibility = "PRIVATE"
+	stringType = "String"
+	outputCharacterEncoding = "UTF-8"
+
+}
+
+task generateAvro(type: GenerateAvroJavaTask) {
+	source("src/main/resources/avro/")
+	outputDir = file("build/generated/java/main")
+}
+
+sourceSets {
+	main {
+		java.srcDirs += generateAvro.outputs
+	}
+}

src/main/java/com/safeway/epe/Config/ConfigProps.java

+package com.safeway.epe.Config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+@Data
+@Configuration
+@ConfigurationProperties(prefix = "kafka.producer")
+public class ConfigProps {
+
+    private String offerTransactionRecordTopic;
+}

src/main/java/com/safeway/epe/controller/TransactionController.java

 package com.safeway.epe.controller;
 
+import com.safeway.epe.domain.Offsets;
 import com.safeway.epe.domain.TransactionRecorder;
 import com.safeway.epe.service.TransactionService;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -16,15 +17,9 @@ public class TransactionController
     @Autowired
     private TransactionService service;
 
-    /*@GetMapping("transactions")
-    public ResponseEntity<List<TransactionController>> getAllTransactions()
-    {
-          return service.getAllTransactions();
-    }*/
-
     @GetMapping("transaction/{uuid}")
-    public ResponseEntity<TransactionRecorder> getTransaction(@PathVariable("uuid") String uuid)
+    public ResponseEntity<List<Offsets>> getTransaction(@PathVariable("uuid") String uuid)
     {
-        return service.getTransactionById(uuid);
+        return service.produceMessage(uuid);
     }
 }

src/main/java/com/safeway/epe/domain/ConsumerPayload.java

+package com.safeway.epe.domain;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class ConsumerPayload {
+    @JsonProperty("key_schema_id")
+    private String keySchemaId;
+    @JsonProperty("value_schema_id")
+    private String valueSchemaId;
+    @JsonProperty("offsets")
+    private List<Offsets> offsetsList;
+}
\ No newline at end of file

src/main/java/com/safeway/epe/domain/Offsets.java

+package com.safeway.epe.domain;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class Offsets {
+    @JsonProperty("partition")
+    private String partition;
+    @JsonProperty("offset")
+    private String offset;
+}
+
+
+

src/main/java/com/safeway/epe/domain/Record.java

+package com.safeway.epe.domain;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class Record {
+
+    @JsonProperty("key")
+    private String key;
+    @JsonProperty("value")
+    private TransactionRecorder transactionRecorder;
+
+}
\ No newline at end of file

src/main/java/com/safeway/epe/domain/TransactionPayload.java

+package com.safeway.epe.domain;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class TransactionPayload {
+    @JsonProperty("records")
+    private List<Record> records;
+}
\ No newline at end of file

src/main/java/com/safeway/epe/domain/TransactionRecorder.java

@@ -33,13 +33,13 @@ public class TransactionRecorder
     @Column(name = "uuid")
     @JsonProperty("uuid")
     UUID uuid;
-    @Type(type = "jsonb")
-    @JsonProperty("offertransactionresponse")
+//    @Type(type = "jsonb")
+//    @JsonProperty("offertransactionresponse")
     @Column(name = "offertransactionresponse")
     String offerTransactionResponse;
 
-    @Type(type="jsonb")
-    @JsonProperty("offers")
+//    @Type(type="jsonb")
+//    @JsonProperty("offers")
     @Column(name="offers")
     String offers;
 

src/main/java/com/safeway/epe/mapper/TransactionPayloadMapper.java

+package com.safeway.epe.mapper;
+
+import com.safeway.epe.domain.Record;
+import com.safeway.epe.domain.TransactionPayload;
+import org.mapstruct.Mapper;
+import org.mapstruct.Mapping;
+import org.mapstruct.Mappings;
+import org.mapstruct.Named;
+import org.mapstruct.factory.Mappers;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Mapper
+public interface TransactionPayloadMapper {
+    TransactionPayloadMapper INSTANCE = Mappers.getMapper(TransactionPayloadMapper.class);
+
+    @Mappings({
+            @Mapping(source = "record", target = "records", qualifiedByName = "transactionCreation")
+             })
+    public TransactionPayload transactionPayloadMapper(Record record);
+
+    @Named("transactionCreation")
+    default List<Record> transactionConversion(Record record){
+        List<Record> recordList = new ArrayList<>();
+        recordList.add(record);
+        return recordList;
+    }
+}

src/main/java/com/safeway/epe/mapper/TransactionRecordMapper.java

+package com.safeway.epe.mapper;
+
+
+import com.safeway.epe.domain.Record;
+import com.safeway.epe.domain.TransactionRecorder;
+import org.mapstruct.*;
+import org.mapstruct.factory.Mappers;
+
+@Mapper(nullValueCheckStrategy = NullValueCheckStrategy.ALWAYS)
+public interface TransactionRecordMapper {
+
+    TransactionRecordMapper INSTANCE =Mappers.getMapper(TransactionRecordMapper.class);
+
+    @Mappings({
+            @Mapping(source = "transactionRecorder", target = "transactionRecorder", qualifiedByName = "recordCreation"),
+            @Mapping(target = "key", qualifiedByName = "recordCreation")
+            })
+    public Record transactionRecordMapper(String key, TransactionRecorder transactionRecorder);
+
+    @Named("recordCreation")
+    default Record recordConvertion(String key, TransactionRecorder transactionRecorder) {
+        return new Record(key, transactionRecorder);
+    }
+
+}

src/main/java/com/safeway/epe/service/ProducerService.java

+package com.safeway.epe.service;
+
+import com.safeway.epe.domain.Offsets;
+import com.safeway.epe.domain.TransactionPayload;
+import com.safeway.epe.domain.TransactionRecorder;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Component
+public class ProducerService {
+
+    public List<Offsets> produce(TransactionPayload request, String topic, TransactionRecorder transactionRecorder, boolean isReprocesed){
+
+        return new ArrayList<>();
+    }
+}

src/main/java/com/safeway/epe/service/TransactionService.java

 package com.safeway.epe.service;
 
 import com.safeway.epe.controller.TransactionController;
+import com.safeway.epe.domain.Offsets;
 import com.safeway.epe.domain.TransactionRecorder;
 import org.springframework.http.ResponseEntity;
 
@@ -8,6 +9,6 @@ import java.util.List;
 
 public interface TransactionService
 {
-      //ResponseEntity<List<TransactionController>> getAllTransactions();
-      ResponseEntity<TransactionRecorder> getTransactionById(String uuid);
+
+      ResponseEntity<List<Offsets>> produceMessage(String uuid);
 }

src/main/java/com/safeway/epe/service/TransactionServiceImpl.java

 package com.safeway.epe.service;
 
+import com.safeway.epe.Config.ConfigProps;
 import com.safeway.epe.controller.TransactionController;
+import com.safeway.epe.domain.Offsets;
+import com.safeway.epe.domain.Record;
+import com.safeway.epe.domain.TransactionPayload;
 import com.safeway.epe.domain.TransactionRecorder;
+import com.safeway.epe.mapper.TransactionPayloadMapper;
+import com.safeway.epe.mapper.TransactionRecordMapper;
 import com.safeway.epe.repository.TransactionRepository;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
@@ -16,16 +22,27 @@ public class TransactionServiceImpl implements  TransactionService
     @Autowired
     TransactionRepository repository;
 
-    /*@Override
-    public ResponseEntity<List<TransactionController>> getAllTransactions() {
-        List<TransactionRecorder> transactions = new ArrayList<TransactionRecorder>();
-        repository.findAll().forEach(transactions::add);
-        return new ResponseEntity<List<TransactionController>>(HttpStatus.OK);
-    }*/
+    @Autowired
+    ConfigProps configProps;
+
+    @Autowired
+    ProducerService producerService;
 
     @Override
-    public ResponseEntity<TransactionRecorder> getTransactionById(String uuid) {
+    public ResponseEntity<List<Offsets>> produceMessage(String uuid) {
         Optional<TransactionRecorder> optionalTransaction = repository.findById(UUID.fromString(uuid));
-        return ResponseEntity.status(HttpStatus.FOUND).body(optionalTransaction.get());
+        if (optionalTransaction.isPresent()) {
+            return sendData(optionalTransaction.get(),false);
+        }
+        return ResponseEntity.status(HttpStatus.NOT_FOUND).body(Arrays.asList(new Offsets()));
     }
+
+    private ResponseEntity<List<Offsets>> sendData(TransactionRecorder transactionRecorder, boolean isReprocesed) {
+        Record record = TransactionRecordMapper.INSTANCE.recordConvertion(transactionRecorder.getUuid().toString(), transactionRecorder);
+        TransactionPayload transactionPayload = TransactionPayloadMapper.INSTANCE.transactionPayloadMapper(record);
+        System.out.println(configProps.getOfferTransactionRecordTopic());
+//        return ResponseEntity.status(HttpStatus.NOT_FOUND).body(Arrays.asList(new Offsets()));
+       return  ResponseEntity.ok(producerService.produce(transactionPayload,configProps.getOfferTransactionRecordTopic(),transactionRecorder,isReprocesed));
+    }
+
 }

src/main/resources/application.yml

@@ -12,4 +12,12 @@ spring:
       hibernate:
         dialect: org.hibernate.dialect.PostgreSQLDialect
 server:
-  port: 8200
+  port: 8200
\ No newline at end of file
+
+kafka:
+  producer:
+    offer-transaction-record-topic : transactionRecord
+
+schema:
+  api:
+    baseurl:  http://localhost:8082

src/main/resources/avro/transactionRecord.avsc

+{
+  "name": "TransactionRecorder1",
+  "type": "record",
+  "namespace": "com.safeway.epe.domain",
+  "fields": [
+    {
+      "name": "uuid",
+      "type": "string"
+    },
+    {
+      "name": "offertransactionresponse",
+      "type": "string"
+    },
+    {
+      "name": "offers",
+      "type": "string"
+    }
+  ]
+}
\ No newline at end of file