fix(): use same JDBC mapper (#2915)

* fix(): use same JDBC mapper * cleanup * fix(test): change wiremock unit test port --------- Co-authored-by: Ludovic DEHON <[email protected]>
kestra-io · Jan 23, 2024 · 7463224 · 7463224
1 parent 668380e
commit 7463224
Show file tree

Hide file tree

Showing 11 changed files with 23 additions and 20 deletions.
diff --git a/jdbc/src/main/java/io/kestra/jdbc/AbstractJdbcRepository.java b/jdbc/src/main/java/io/kestra/jdbc/AbstractJdbcRepository.java
@@ -1,11 +1,11 @@
 package io.kestra.jdbc;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.exceptions.DeserializationException;
 import io.kestra.core.models.executions.metrics.MetricAggregation;
 import io.kestra.core.queues.QueueService;
 import io.kestra.core.repositories.ArrayListTotal;
-import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.utils.IdUtils;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.data.model.Pageable;
@@ -31,6 +31,8 @@
 import java.util.stream.IntStream;
 
 public abstract class AbstractJdbcRepository<T> {
+    protected static final ObjectMapper MAPPER = JdbcMapper.of();
+
     protected final QueueService queueService;
 
     protected final Class<T> cls;
@@ -163,7 +165,7 @@ public <R extends Record> Instant getDate(R record, String groupByType) {
 
     public T deserialize(String record) {
         try {
-            return JacksonMapper.ofJson().readValue(record, cls);
+            return MAPPER.readValue(record, cls);
         } catch (IOException e) {
             throw new DeserializationException(e, record);
         }

diff --git a/jdbc/src/main/java/io/kestra/jdbc/JdbcMapper.java b/jdbc/src/main/java/io/kestra/jdbc/JdbcMapper.java
@@ -21,7 +21,7 @@ public abstract class JdbcMapper {
 
     public static ObjectMapper of() {
         if (MAPPER == null) {
-            MAPPER = JacksonMapper.ofJson().copy();
+            MAPPER = JacksonMapper.ofJson(false).copy();
 
             final SimpleModule module = new SimpleModule();
             module.addSerializer(Instant.class, new JsonSerializer<>() {

diff --git a/jdbc/src/main/java/io/kestra/jdbc/JdbcWorkerTriggerResultQueueService.java b/jdbc/src/main/java/io/kestra/jdbc/JdbcWorkerTriggerResultQueueService.java
@@ -23,7 +23,7 @@
 @Singleton
 @Slf4j
 public class JdbcWorkerTriggerResultQueueService {
-    private final static ObjectMapper MAPPER = JacksonMapper.ofJson();
+    private final static ObjectMapper MAPPER = JdbcMapper.of();
 
     private final JdbcQueue<WorkerTriggerResult> workerTriggerResultQueue;
     @Inject

diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSubflowExecutionStorage.java b/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSubflowExecutionStorage.java
@@ -5,7 +5,7 @@
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.runners.SubflowExecution;
-import io.kestra.core.serializers.JacksonMapper;
+import io.kestra.jdbc.JdbcMapper;
 import io.kestra.jdbc.repository.AbstractJdbcRepository;
 import org.jooq.DSLContext;
 import org.jooq.Field;
@@ -18,7 +18,7 @@
 import java.util.Optional;
 
 public abstract class AbstractJdbcSubflowExecutionStorage extends AbstractJdbcRepository {
-    private final static ObjectMapper MAPPER = JacksonMapper.ofJson();
+    private final static ObjectMapper MAPPER = JdbcMapper.of();
     protected io.kestra.jdbc.AbstractJdbcRepository<SubflowExecution<?>> jdbcRepository;
 
     @SuppressWarnings({"unchecked", "rawtypes"})

diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -31,6 +31,7 @@
 import io.kestra.core.topologies.FlowTopologyService;
 import io.kestra.core.utils.Await;
 import io.kestra.core.utils.Either;
+import io.kestra.jdbc.JdbcMapper;
 import io.kestra.jdbc.repository.AbstractJdbcExecutionRepository;
 import io.kestra.jdbc.repository.AbstractJdbcFlowTopologyRepository;
 import io.kestra.jdbc.repository.AbstractJdbcWorkerInstanceRepository;
@@ -58,7 +59,7 @@
 @JdbcRunnerEnabled
 @Slf4j
 public class JdbcExecutor implements ExecutorInterface {
-    private static final ObjectMapper MAPPER = JacksonMapper.ofJson();
+    private static final ObjectMapper MAPPER = JdbcMapper.of();;
 
     private final ScheduledExecutorService schedulerDelay = Executors.newSingleThreadScheduledExecutor();
 

diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java
@@ -7,11 +7,11 @@
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.queues.QueueService;
-import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.utils.Either;
 import io.kestra.core.utils.ExecutorsUtils;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.jdbc.JdbcConfiguration;
+import io.kestra.jdbc.JdbcMapper;
 import io.kestra.jdbc.JooqDSLContextWrapper;
 import io.kestra.jdbc.repository.AbstractJdbcRepository;
 import io.micronaut.context.ApplicationContext;
@@ -42,7 +42,7 @@
 
 @Slf4j
 public abstract class JdbcQueue<T> implements QueueInterface<T> {
-    protected static final ObjectMapper mapper = JacksonMapper.ofJson();
+    protected static final ObjectMapper MAPPER = JdbcMapper.of();
 
     private static ExecutorService poolExecutor;
 
@@ -86,7 +86,7 @@ protected Map<Field<Object>, Object> produceFields(String consumerGroup, String
         Map<Field<Object>, Object> fields = new HashMap<>();
         fields.put(AbstractJdbcRepository.field("type"), this.cls.getName());
         fields.put(AbstractJdbcRepository.field("key"), key != null ? key : IdUtils.create());
-        fields.put(AbstractJdbcRepository.field("value"), JSONB.valueOf(mapper.writeValueAsString(message)));
+        fields.put(AbstractJdbcRepository.field("value"), JSONB.valueOf(MAPPER.writeValueAsString(message)));
 
         if (consumerGroup != null) {
             fields.put(AbstractJdbcRepository.field("consumer_group"), consumerGroup);
@@ -289,7 +289,7 @@ protected List<Either<T, DeserializationException>> map(Result<Record> fetch) {
         return fetch
             .map(record -> {
                 try {
-                    return Either.left(JacksonMapper.ofJson().readValue(record.get("value", String.class), cls));
+                    return Either.left(MAPPER.readValue(record.get("value", String.class), cls));
                 } catch (JsonProcessingException e) {
                     return Either.right(new DeserializationException(e, record.get("value", String.class)));
                 }

diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/BlueprintControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/BlueprintControllerTest.java
@@ -22,7 +22,7 @@
 import static org.hamcrest.Matchers.*;
 
 @MicronautTest
-@WireMockTest(httpPort = 8081)
+@WireMockTest(httpPort = 28181)
 class BlueprintControllerTest {
     @Inject
     @Client("/")

diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/EditorControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/EditorControllerTest.java
@@ -22,7 +22,7 @@
 import static org.hamcrest.Matchers.is;
 
 // For this controller tests, we replace every Marketplace URLs with http://localhost:8081/{previous-host} to target Wiremock server
-@WireMockTest(httpPort = 8081)
+@WireMockTest(httpPort = 28181)
 class EditorControllerTest extends JdbcH2ControllerTest {
     @Inject
     @Client("/")
@@ -36,12 +36,12 @@ private MarketplaceRequestMapper marketplaceRequestMapper() {
         return new MarketplaceRequestMapper() {
             @Override
             public String url(MarketplaceRequestType type) {
-                return type.getUrl().replaceFirst("https?://", "http://localhost:8081/");
+                return type.getUrl().replaceFirst("https?://", "http://localhost:28181/");
             }
 
             @Override
             public String resourceBaseUrl(String publisher) {
-                return super.resourceBaseUrl(publisher).replaceFirst("https?://", "http://localhost:8081/");
+                return super.resourceBaseUrl(publisher).replaceFirst("https?://", "http://localhost:28181/");
             }
         };
     }
@@ -123,7 +123,7 @@ void getResource_ReplaceUrlsForExtensionPath() throws IOException {
 
     private MappingBuilder hasGoodHeaders(MappingBuilder mappingBuilder) {
         return mappingBuilder.withHeader("Origin", equalTo("http://localhost:8080"))
-            .withHeader("Host", equalTo("localhost:8081"))
+            .withHeader("Host", equalTo("localhost:28181"))
             .withHeader("Cookie", absent())
             .withHeader("Access-Control-Allow-Origin", absent());
     }

diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
@@ -446,7 +446,7 @@ void downloadFile() throws TimeoutException {
             FileMetas.class
         ).blockingFirst();
 
-        assertThat(metas.getSize(), equalTo(3002L));
+        assertThat(metas.getSize(), greaterThanOrEqualTo(3003L));
 
         String newExecutionId = IdUtils.create();
 

diff --git a/webserver/src/test/resources/__files/extension.json b/webserver/src/test/resources/__files/extension.json
@@ -1,4 +1,4 @@
 [
-    "http://localhost:8081/my-publisher.vscode-unpkg.net/my-publisher/my-extension/1.0.0/extension/package.json",
-    "http://localhost:8081/my-publisher.vscode-unpkg.net/my-publisher/my-extension/1.0.0/extension/package.nls.json"
+    "http://localhost:28181/my-publisher.vscode-unpkg.net/my-publisher/my-extension/1.0.0/extension/package.json",
+    "http://localhost:28181/my-publisher.vscode-unpkg.net/my-publisher/my-extension/1.0.0/extension/package.nls.json"
 ]
diff --git a/webserver/src/test/resources/application-test.yml b/webserver/src/test/resources/application-test.yml
@@ -8,7 +8,7 @@ micronaut:
   http:
     services:
       api:
-        url: http://localhost:8081
+        url: http://localhost:28181
   server:
     cors:
       enabled: true