fix: V3 error response handling

CHANGELOG.md

@@ -2,7 +2,11 @@
 
 ### Features
 
-1. [#352](https://github.com/InfluxCommunity/influxdb3-java/pull/352): Update Apache Flight client dependencies.
+1. [#352](https://github.com/InfluxCommunity/influxdb3-java/pull/352): Upgrade Arrow Flight client dependencies.
+
+### Bug Fixes
+
+1. [#351](https://github.com/InfluxCommunity/influxdb3-java/pull/351): Enterprise/Core structured errors handling.
 
 ### CI
 

pom.xml

@@ -317,6 +317,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <version>${junit-jupiter.version}</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.assertj</groupId>
             <artifactId>assertj-core</artifactId>

src/main/java/com/influxdb/v3/client/internal/RestClient.java

@@ -217,22 +217,12 @@ HttpResponse<String> request(@Nonnull final String path,
 
         int statusCode = response.statusCode();
         if (statusCode < 200 || statusCode >= 300) {
-            String reason = "";
+            String reason;
             String body = response.body();
-            if (!body.isEmpty()) {
-                try {
-                    final JsonNode root = objectMapper.readTree(body);
-                    final List<String> possibilities = List.of("message", "error_message", "error");
-                    for (final String field : possibilities) {
-                        final JsonNode node = root.findValue(field);
-                        if (node != null) {
-                            reason = node.asText();
-                            break;
-                        }
-                    }
-                } catch (JsonProcessingException e) {
-                    LOG.debug("Can't parse msg from response {}", response);
-                }
+            reason = formatErrorMessage(body, response.headers().firstValue("Content-Type").orElse(null));
+
+            if (reason == null) {
+                reason = "";
             }
 
             if (reason.isEmpty()) {
@@ -257,6 +247,103 @@ HttpResponse<String> request(@Nonnull final String path,
         return response;
     }
 
+    @Nullable
+    private String formatErrorMessage(@Nonnull final String body, @Nullable final String contentType) {
+        if (body.isEmpty()) {
+            return null;
+        }
+
+        if (contentType != null
+                && !contentType.isEmpty()
+                && !contentType.regionMatches(true, 0, "application/json", 0, "application/json".length())) {
+            return null;
+        }
+
+        try {
+            final JsonNode root = objectMapper.readTree(body);
+            if (!root.isObject()) {
+                return null;
+            }
+
+            final String rootMessage = errNonEmptyField(root, "message");
+            if (rootMessage != null) {
+                return rootMessage;
+            }
+
+            final String error = errNonEmptyField(root, "error");
+            final JsonNode dataNode = root.get("data");
+
+            // InfluxDB 3 Core/Enterprise write error format:
+            // {"error":"...","data":[{"error_message":"...","line_number":2,"original_line":"..."}]}
+            if (error != null && dataNode != null && dataNode.isArray()) {
+                final StringBuilder message = new StringBuilder(error);
+                boolean hasDetails = false;
+                for (JsonNode item : dataNode) {
+                    final String detail = errFormatDataArrayDetail(item);
+                    if (detail == null) {
+                        continue;
+                    }
+                    if (!hasDetails) {
+                        message.append(':');
+                        hasDetails = true;
+                    }
+                    message.append("\n\t").append(detail);
+                }
+                return message.toString();
+            }
+
+            // Core/Enterprise object format:
+            // {"error":"...","data":{"error_message":"..."}}
+            final String errorMessage = errNonEmptyField(dataNode, "error_message");
+            if (errorMessage != null) {
+                return errorMessage;
+            }
+
+            return error;
+        } catch (JsonProcessingException e) {
+            LOG.debug("Can't parse msg from response body {}", body, e);
+            return null;
+        }
+    }
+
+    @Nullable
+    private String errNonEmptyText(@Nullable final JsonNode node) {
+        if (node == null || node.isNull()) {
+            return null;
+        }
+        final String value = node.asText();
+        return value.isEmpty() ? null : value;
+    }
+
+    @Nullable
+    private String errNonEmptyField(@Nullable final JsonNode object, @Nonnull final String fieldName) {
+        if (object == null || !object.isObject()) {
+            return null;
+        }
+        return errNonEmptyText(object.get(fieldName));
+    }
+
+    @Nullable
+    private String errFormatDataArrayDetail(@Nullable final JsonNode item) {
+        if (!item.isObject()) {
+            return null;
+        }
+
+        final String errorMessage = errNonEmptyField(item, "error_message");
+        if (errorMessage == null) {
+            return null;
+        }
+
+        if (item.hasNonNull("line_number")) {
+            final String originalLine = errNonEmptyField(item, "original_line");
+            if (originalLine != null) {
+                final String lineNumber = item.get("line_number").asText();
+                return "line " + lineNumber + ": " + errorMessage + " (" + originalLine + ")";
+            }
+        }
+        return errorMessage;
+    }
+
     private X509TrustManager getX509TrustManagerFromFile(@Nonnull final String filePath) {
         try {
             KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());

src/test/java/com/influxdb/v3/client/AbstractMockServerTest.java

@@ -65,10 +65,23 @@ protected MockResponse createResponse(final int responseCode,
                                           @Nullable final Map<String, String> headers,
                                           @Nullable final String body) {
 
+      return createResponse(responseCode, "text/csv; charset=utf-8", headers, body);
+    }
+
+    @Nonnull
+    protected MockResponse createResponse(final int responseCode,
+                                          @Nullable final String contentType,
+                                          @Nullable final Map<String, String> headers,
+                                          @Nullable final String body) {
+
       MockResponse.Builder mrb = new MockResponse.Builder();
       mrb.code(responseCode);
-      Map<String, String> effectiveHeaders = new HashMap<>(Map.of("Content-Type", "text/csv; charset=utf-8",
-        "Date", "Tue, 26 Jun 2018 13:15:01 GMT"));
+      Map<String, String> effectiveHeaders = new HashMap<>(Map.of(
+        "Date", "Tue, 26 Jun 2018 13:15:01 GMT"
+      ));
+      if (contentType != null) {
+        effectiveHeaders.put("Content-Type", contentType);
+      }
       if (headers != null) {
         effectiveHeaders.putAll(headers);
       }
@@ -85,9 +98,6 @@ protected MockResponse createResponse(final int responseCode,
     @Nonnull
     protected MockResponse createResponse(final int responseCode) {
 
-        return createResponse(responseCode, Map.of(
-          "Content-Type", "text/csv; charset=utf-8",
-          "Date", "Tue, 26 Jun 2018 13:15:01 GMT"
-        ), null);
+        return createResponse(responseCode, "text/csv; charset=utf-8", null, null);
     }
 }

src/test/java/com/influxdb/v3/client/InfluxDBClientWriteTest.java

@@ -517,7 +517,8 @@ void defaultTags() throws InterruptedException {
     @Test
     public void retryHandled429Test() {
         mockServer.enqueue(createResponse(429,
-          Map.of("retry-after", "42", "content-type", "application/json"),
+          "application/json",
+          Map.of("retry-after", "42"),
           "{ \"message\" : \"Too Many Requests\" }"));
 
         Point point = Point.measurement("mem")