fix: make it run

Author: Angular2Guy

database-to-rest/pom.xml

@@ -16,7 +16,7 @@
 	<description>Pipeline from Db to Rest Endpoint</description>
 	<properties>
 		<java.version>24</java.version>
-		<debezium.version>3.0.5.Final</debezium.version>
+		<debezium.version>3.3.2.Final</debezium.version>
 	</properties>
 	<dependencies>
 		<dependency>

database-to-rest/src/main/java/de/xxx/dbtorest/kafka/KafkaConfig.java

@@ -33,12 +33,14 @@
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
 import org.springframework.kafka.transaction.KafkaTransactionManager;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
 
 import java.util.LinkedHashMap;
 import java.util.Map;
 
 @Configuration
 @EnableKafka
+@EnableTransactionManagement
 public class KafkaConfig {
     private static final Logger log = LoggerFactory.getLogger(KafkaConfig.class);
     public static final String ORDERPRODUCT_TOPIC = "orderproduct-topic";

database-to-rest/src/main/java/de/xxx/dbtorest/kafka/KafkaProducer.java

@@ -12,14 +12,14 @@
  */
 package de.xxx.dbtorest.kafka;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import de.xxx.dbtorest.model.DbChangeDto;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Component;
+import tools.jackson.databind.json.JsonMapper;
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -28,10 +28,10 @@
 public class KafkaProducer {
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducer.class);
     private final KafkaTemplate<String, String> kafkaTemplate;
-    private final ObjectMapper objectMapper;
+    private final JsonMapper objectMapper;
     private final AdminClient adminClient;
 
-    public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate, AdminClient adminClient, ObjectMapper objectMapper) {
+    public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate, AdminClient adminClient, JsonMapper objectMapper) {
         this.adminClient = adminClient;
         this.kafkaTemplate = kafkaTemplate;
         this.objectMapper = objectMapper;

database-to-rest/src/main/java/de/xxx/dbtorest/sink/DbChangeSinkService.java

@@ -22,27 +22,25 @@
 import org.springframework.http.MediaType;
 import org.springframework.stereotype.Service;
 import org.springframework.web.client.RestClient;
+import tools.jackson.databind.json.JsonMapper;
+
+import java.util.Optional;
 
 
 @Service
 public class DbChangeSinkService {
     private static final Logger LOG = LoggerFactory.getLogger(DbChangeSinkService.class);
-    private final ObjectMapper objectMapper;
+    private final JsonMapper objectMapper;
     private final RestClient restClient = RestClient.create();
     @Value("${rest.endpoint.url}")
     private String restEndpointUrl;
 
-    public DbChangeSinkService(ObjectMapper objectMapper) {
+    public DbChangeSinkService(JsonMapper objectMapper) {
         this.objectMapper = objectMapper;
     }
 
     public void handleDbChange(DbChangeDto dbChangeDto) {
-        Wrapper wrapper;
-        try {
-            wrapper = this.objectMapper.readValue(dbChangeDto.value(), Wrapper.class);
-        } catch (JsonProcessingException e) {
-            throw new RuntimeException(e);
-        }
+        var wrapper = Optional.ofNullable( dbChangeDto.value()).stream().map(value -> this.objectMapper.readValue(value, Wrapper.class)).findFirst().orElse(new Wrapper(null, null));
         LOG.info("DbChange received: {}", dbChangeDto.toString());
         this.restClient.post().uri(this.restEndpointUrl.trim() + "/rest/orderproduct").contentType(MediaType.APPLICATION_JSON).body(wrapper).retrieve().toBodilessEntity();
     }

event-to-file/src/main/java/de/xxx/eventtofile/kafka/KafkaConfig.java

@@ -31,12 +31,14 @@
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
 import org.springframework.kafka.transaction.KafkaTransactionManager;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
 
 import java.util.LinkedHashMap;
 import java.util.Map;
 
 @Configuration
 @EnableKafka
+@EnableTransactionManagement
 public class KafkaConfig {
     private static final Logger log = LoggerFactory.getLogger(KafkaConfig.class);
     public static final String FLIGHT_SOURCE_TOPIC = "flight-source-topic";

soap-to-db/pom.xml

@@ -60,7 +60,11 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-webservices</artifactId>
         </dependency>
-
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+            <scope>runtime</scope>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-actuator-test</artifactId>
@@ -86,11 +90,11 @@
             <artifactId>spring-boot-starter-webservices-test</artifactId>
             <scope>test</scope>
         </dependency>
-        <!--
 		<dependency>
 			<groupId>wsdl4j</groupId>
 			<artifactId>wsdl4j</artifactId>
 		</dependency>
+        <!--
 		<dependency>
 			<groupId>org.springframework.kafka</groupId>
 			<artifactId>spring-kafka</artifactId>

soap-to-db/src/main/java/de/xxx/soaptodb/kafka/KafkaConfig.java

@@ -34,12 +34,14 @@
 import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
 import org.springframework.kafka.transaction.KafkaTransactionManager;
 import org.springframework.orm.jpa.JpaTransactionManager;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
 
 import java.util.LinkedHashMap;
 import java.util.Map;
 
 @Configuration
 @EnableKafka
+@EnableTransactionManagement
 public class KafkaConfig {
     private static final Logger log = LoggerFactory.getLogger(KafkaConfig.class);
     public static final String COUNTRY_TOPIC = "country-topic";

source-sink/pom.xml

@@ -151,6 +151,7 @@
                     </args>
                     <compilerPlugins>
                         <plugin>spring</plugin>
+                        <plugin>jpa</plugin>
                     </compilerPlugins>
                 </configuration>
                 <dependencies>

source-sink/src/main/kotlin/de/xxx/sourcesink/adapter/event/KafkaConfig.kt

@@ -22,10 +22,12 @@ import org.springframework.context.annotation.Configuration
 import org.springframework.kafka.annotation.EnableKafka
 import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.kafka.core.ProducerFactory
+import org.springframework.transaction.annotation.EnableTransactionManagement
 
 
 @Configuration
 @EnableKafka
+@EnableTransactionManagement
 class KafkaConfig(val producerFactory: ProducerFactory<String, String>) {
     companion object {
         @JvmStatic

source-sink/src/main/kotlin/de/xxx/sourcesink/adapter/event/KafkaProducer.kt

@@ -12,18 +12,17 @@ limitations under the License.
  */
 package de.xxx.sourcesink.adapter.event
 
-import com.fasterxml.jackson.databind.ObjectMapper
 import de.xxx.sourcesink.domain.model.FlightDto
-import org.apache.kafka.clients.admin.AdminClient
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.stereotype.Component
+import tools.jackson.databind.json.JsonMapper
 import java.util.concurrent.TimeUnit
 
 @Component
 class KafkaProducer(val kafkaTemplate: KafkaTemplate<String, String>,
-                    val objectMapper: ObjectMapper) {
+                    val objectMapper: JsonMapper) {
     private val LOGGER: Logger = LoggerFactory.getLogger(KafkaProducer::class.java)
 
     fun sendFlightMsg(flightDto: FlightDto) {