Merge pull request #600 from swisspost/develop

develop to master to trigger release
swisspost · Jun 20, 2024 · 9ccff43 · 9ccff43
2 parents b2aa5ac + 2610f64
commit 9ccff43
Show file tree

Hide file tree

Showing 30 changed files with 71 additions and 46 deletions.
diff --git a/gateleen-cache/pom.xml b/gateleen-cache/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-cache</artifactId>

diff --git a/gateleen-core/pom.xml b/gateleen-core/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-core</artifactId>

diff --git a/gateleen-delegate/pom.xml b/gateleen-delegate/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-delegate</artifactId>

diff --git a/gateleen-delta/pom.xml b/gateleen-delta/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-delta</artifactId>

diff --git a/gateleen-expansion/pom.xml b/gateleen-expansion/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-expansion</artifactId>

diff --git a/gateleen-hook-js/pom.xml b/gateleen-hook-js/pom.xml
@@ -4,7 +4,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
     <artifactId>gateleen-hook-js</artifactId>
     <packaging>jar</packaging>

diff --git a/gateleen-hook/pom.xml b/gateleen-hook/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-hook</artifactId>

diff --git a/gateleen-kafka/pom.xml b/gateleen-kafka/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-kafka</artifactId>

diff --git a/gateleen-kafka/src/main/java/org/swisspush/gateleen/kafka/KafkaProducerRecordBuilder.java b/gateleen-kafka/src/main/java/org/swisspush/gateleen/kafka/KafkaProducerRecordBuilder.java
@@ -15,7 +15,6 @@
 import java.util.List;
 
 import static java.lang.System.currentTimeMillis;
-import static java.lang.Thread.currentThread;
 import static org.slf4j.LoggerFactory.getLogger;
 
 /**
@@ -52,33 +51,31 @@ class KafkaProducerRecordBuilder {
      * @throws ValidationException when the payload is not valid (missing properties, wrong types, etc.)
      */
     Future<List<KafkaProducerRecord<String, String>>> buildRecordsAsync(String topic, Buffer payload) {
-        return Future.<Void>succeededFuture().compose((Void v) -> {
+        return Future.<Void>succeededFuture().compose((Void v) -> vertx.executeBlocking(() -> {
+            long beginEpchMs = currentTimeMillis();
             JsonObject payloadObj;
             try {
                 payloadObj = new JsonObject(payload);
             } catch (DecodeException de) {
-                return Future.failedFuture(new ValidationException("Error while parsing payload", de));
+                throw new ValidationException("Error while parsing payload", de);
             }
             JsonArray recordsArray;
             try {
                 recordsArray = payloadObj.getJsonArray(RECORDS);
             } catch (ClassCastException cce) {
-                return Future.failedFuture(new ValidationException("Property '" + RECORDS + "' must be of type JsonArray holding JsonObject objects"));
+                throw new ValidationException("Property '" + RECORDS + "' must be of type JsonArray holding JsonObject objects");
             }
             if (recordsArray == null) {
-                return Future.failedFuture(new ValidationException("Missing 'records' array"));
+                throw new ValidationException("Missing 'records' array");
             }
-            return vertx.executeBlocking(() -> {
-                long beginEpchMs = currentTimeMillis();
-                List<KafkaProducerRecord<String, String>> kafkaProducerRecords = new ArrayList<>(recordsArray.size());
-                for (int i = 0; i < recordsArray.size(); i++) {
-                    kafkaProducerRecords.add(fromRecordJsonObject(topic, recordsArray.getJsonObject(i)));
-                }
-                long durationMs = currentTimeMillis() - beginEpchMs;
-                log.debug("Serializing JSON did block thread for {}ms", durationMs);
-                return kafkaProducerRecords;
-            });
-        });
+            List<KafkaProducerRecord<String, String>> kafkaProducerRecords = new ArrayList<>(recordsArray.size());
+            for (int i = 0; i < recordsArray.size(); i++) {
+                kafkaProducerRecords.add(fromRecordJsonObject(topic, recordsArray.getJsonObject(i)));
+            }
+            long durationMs = currentTimeMillis() - beginEpchMs;
+            log.debug("Parsing and Serializing JSON did block thread for {}ms", durationMs);
+            return kafkaProducerRecords;
+        }));
     }
 
     /** @deprecated Use {@link #buildRecordsAsync(String, Buffer)}. */

diff --git a/gateleen-logging/pom.xml b/gateleen-logging/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-logging</artifactId>

diff --git a/gateleen-merge/pom.xml b/gateleen-merge/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-merge</artifactId>

diff --git a/gateleen-monitoring/pom.xml b/gateleen-monitoring/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-monitoring</artifactId>

diff --git a/gateleen-packing/pom.xml b/gateleen-packing/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-packing</artifactId>

diff --git a/gateleen-player/pom.xml b/gateleen-player/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-player</artifactId>

diff --git a/gateleen-playground/pom.xml b/gateleen-playground/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-playground</artifactId>

diff --git a/gateleen-qos/pom.xml b/gateleen-qos/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-qos</artifactId>

diff --git a/gateleen-queue/pom.xml b/gateleen-queue/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-queue</artifactId>

diff --git a/gateleen-routing/pom.xml b/gateleen-routing/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-routing</artifactId>

diff --git a/gateleen-routing/src/main/java/org/swisspush/gateleen/routing/Forwarder.java b/gateleen-routing/src/main/java/org/swisspush/gateleen/routing/Forwarder.java
@@ -503,7 +503,7 @@ private Handler<AsyncResult<HttpClientResponse>> getAsyncHttpClientResponseHandl
 
             final LoggingWriteStream loggingWriteStream = new LoggingWriteStream(req.response(), loggingHandler, false);
             final Pump pump = Pump.pump(cRes, loggingWriteStream);
-            cRes.endHandler(v -> {
+            Handler<Void> cResEndHandler = v -> {
                 try {
                     req.response().end();
 
@@ -516,7 +516,14 @@ private Handler<AsyncResult<HttpClientResponse>> getAsyncHttpClientResponseHandl
                     // ignore because maybe already closed
                 }
                 vertx.runOnContext(event -> loggingHandler.log());
-            });
+            };
+            try {
+                cRes.endHandler(cResEndHandler);
+            } catch (IllegalStateException ex) {
+                log.warn("cRes.endHandler() failed", ex);
+                respondError(req, StatusCode.INTERNAL_SERVER_ERROR);
+                return;
+            }
             pump.start();
 
             Runnable unpump = () -> {

diff --git a/gateleen-routing/src/main/java/org/swisspush/gateleen/routing/Router.java b/gateleen-routing/src/main/java/org/swisspush/gateleen/routing/Router.java
@@ -15,6 +15,7 @@
 import org.slf4j.LoggerFactory;
 import org.swisspush.gateleen.core.configuration.ConfigurationResourceManager;
 import org.swisspush.gateleen.core.configuration.ConfigurationResourceObserver;
+import org.swisspush.gateleen.core.exception.GateleenExceptionFactory;
 import org.swisspush.gateleen.core.http.HttpClientFactory;
 import org.swisspush.gateleen.core.http.RequestLoggerFactory;
 import org.swisspush.gateleen.core.logging.LoggableResource;
@@ -56,6 +57,7 @@ public class Router implements Refreshable, LoggableResource, ConfigurationResou
     private final String rulesUri;
     private final String userProfileUri;
     private final String serverUri;
+    private final GateleenExceptionFactory exceptionFactory;
     private io.vertx.ext.web.Router router;
     private final LoggingResourceManager loggingResourceManager;
     private final LogAppenderRepository logAppenderRepository;
@@ -117,6 +119,7 @@ public static RouterBuilder builder() {
            HttpClientFactory httpClientFactory,
            int routeMultiplier,
            @Nullable OAuthProvider oAuthProvider,
+           GateleenExceptionFactory exceptionFactory,
            Handler<Void>... doneHandlers) {
         this.storage = storage;
         this.properties = properties;
@@ -135,6 +138,7 @@ public static RouterBuilder builder() {
         this.doneHandlers = doneHandlers;
         this.routeMultiplier = routeMultiplier;
         this.oAuthProvider = oAuthProvider;
+        this.exceptionFactory =  exceptionFactory;
 
         if (oAuthProvider != null) {
             this.oAuthStrategy = new OAuthStrategy(oAuthProvider);
@@ -324,7 +328,7 @@ private void createForwarders(List<Rule> rules, io.vertx.ext.web.Router newRoute
                         vertx.eventBus());
             } else if (rule.getStorage() != null) {
                 forwarder = new StorageForwarder(vertx.eventBus(), rule, loggingResourceManager, logAppenderRepository,
-                        monitoringHandler);
+                        monitoringHandler, exceptionFactory);
             } else if (rule.getScheme().equals("local")) {
                 forwarder = new Forwarder(vertx, selfClient, rule, this.storage, loggingResourceManager, logAppenderRepository,
                         monitoringHandler, userProfileUri, authStrategy);

diff --git a/gateleen-routing/src/main/java/org/swisspush/gateleen/routing/RouterBuilder.java b/gateleen-routing/src/main/java/org/swisspush/gateleen/routing/RouterBuilder.java
@@ -7,6 +7,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.swisspush.gateleen.core.configuration.ConfigurationResourceManager;
+import org.swisspush.gateleen.core.exception.GateleenExceptionFactory;
 import org.swisspush.gateleen.core.http.HttpClientFactory;
 import org.swisspush.gateleen.core.storage.ResourceStorage;
 import org.swisspush.gateleen.logging.LogAppenderRepository;
@@ -52,6 +53,7 @@ public class RouterBuilder {
     private int routeMultiplier = Router.DEFAULT_ROUTER_MULTIPLIER;
 
     private OAuthProvider oAuthProvider;
+    private GateleenExceptionFactory exceptionFactory;
 
     RouterBuilder() {
         // PackagePrivate, as clients should use "Router.builder()" and not this class here directly.
@@ -67,6 +69,10 @@ public Router build() {
             defaultRouteTypes = all();
         }
 
+        if (this.exceptionFactory == null) {
+            this.exceptionFactory = GateleenExceptionFactory.newGateleenThriftyExceptionFactory();
+        }
+
         Handler<Void>[] doneHandlersArray;
         if (doneHandlers == null || doneHandlers.isEmpty()) {
             logger.debug("No doneHandlers specified.");
@@ -102,6 +108,7 @@ public Router build() {
                 httpClientFactory,
                 routeMultiplier,
                 oAuthProvider,
+                exceptionFactory,
                 doneHandlersArray
         );
         if (resourceLoggingEnabled) {
@@ -273,4 +280,9 @@ public RouterBuilder withRouteMultiplier(int routeMultiplier) {
         this.routeMultiplier = routeMultiplier;
         return this;
     }
+
+    public RouterBuilder withExceptionFactory(GateleenExceptionFactory exceptionFactory) {
+        this.exceptionFactory = exceptionFactory;
+        return this;
+    }
 }
diff --git a/gateleen-routing/src/main/java/org/swisspush/gateleen/routing/StorageForwarder.java b/gateleen-routing/src/main/java/org/swisspush/gateleen/routing/StorageForwarder.java
@@ -15,6 +15,7 @@
 import io.vertx.ext.web.RoutingContext;
 import org.slf4j.Logger;
 import org.swisspush.gateleen.core.cors.CORSHandler;
+import org.swisspush.gateleen.core.exception.GateleenExceptionFactory;
 import org.swisspush.gateleen.core.http.HeaderFunctions;
 import org.swisspush.gateleen.core.http.HttpRequest;
 import org.swisspush.gateleen.core.http.RequestLoggerFactory;
@@ -40,13 +41,17 @@ public class StorageForwarder extends AbstractForwarder {
     private Pattern urlPattern;
     private String address;
     private CORSHandler corsHandler;
+    private GateleenExceptionFactory gateleenExceptionFactory;
 
-    public StorageForwarder(EventBus eventBus, Rule rule, LoggingResourceManager loggingResourceManager, LogAppenderRepository logAppenderRepository, MonitoringHandler monitoringHandler) {
+    public StorageForwarder(EventBus eventBus, Rule rule, LoggingResourceManager loggingResourceManager,
+                            LogAppenderRepository logAppenderRepository, MonitoringHandler monitoringHandler,
+                            GateleenExceptionFactory gateleenExceptionFactory) {
         super(rule, loggingResourceManager, logAppenderRepository, monitoringHandler);
         this.eventBus = eventBus;
         this.address = Address.storageAddress() + "-" + rule.getStorage();
         urlPattern = Pattern.compile(rule.getUrlPattern());
         corsHandler = new CORSHandler();
+        this.gateleenExceptionFactory = gateleenExceptionFactory;
     }
 
     @Override
@@ -97,7 +102,7 @@ public void handle(final RoutingContext ctx) {
                     response.setStatusCode(StatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
                     response.setStatusMessage(statusMessage);
                     response.end();
-                    log.error("Storage request failed", result.cause());
+                    log.error(statusMessage, gateleenExceptionFactory.newException(result.cause()));
                 } else {
                     Buffer buffer = result.result().body();
                     int headerLength = buffer.getInt(0);

diff --git a/gateleen-runconfig/pom.xml b/gateleen-runconfig/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-runconfig</artifactId>

diff --git a/gateleen-scheduler/pom.xml b/gateleen-scheduler/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-scheduler</artifactId>

diff --git a/gateleen-security/pom.xml b/gateleen-security/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-security</artifactId>

diff --git a/gateleen-test/pom.xml b/gateleen-test/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
     <artifactId>gateleen-test</artifactId>
     <packaging>jar</packaging>

diff --git a/gateleen-testhelper/pom.xml b/gateleen-testhelper/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.8-SNAPSHOT</version>
+        <version>2.1.9-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-testhelper</artifactId>