diff --git a/src/main/java/com/influxdb/v3/client/InfluxDBPartialWriteException.java b/src/main/java/com/influxdb/v3/client/InfluxDBPartialWriteException.java index 9e7a2494..0ba7ca77 100644 --- a/src/main/java/com/influxdb/v3/client/InfluxDBPartialWriteException.java +++ b/src/main/java/com/influxdb/v3/client/InfluxDBPartialWriteException.java @@ -76,7 +76,7 @@ public static final class LineError { * @param originalLine original line protocol row; may be null if not provided by server */ public LineError(@Nullable final Integer lineNumber, - @Nonnull final String errorMessage, + @Nullable final String errorMessage, @Nullable final String originalLine) { this.lineNumber = lineNumber; this.errorMessage = errorMessage; @@ -94,7 +94,7 @@ public Integer lineNumber() { /** * @return line-level error message */ - @Nonnull + @Nullable public String errorMessage() { return errorMessage; } diff --git a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java index d6fa6466..182003ab 100644 --- a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java +++ b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java @@ -384,7 +384,7 @@ private void writeData(@Nonnull final List data, @Nonnull final WriteOpti headers.putAll(options.headersSafe()); try { - restClient.request(path, HttpMethod.POST, body, queryParams, headers); + restClient.request(path, HttpMethod.POST, body, queryParams, headers, acceptPartial, useV2Api); } catch (InfluxDBApiHttpException e) { if (e.statusCode() == HttpResponseStatus.METHOD_NOT_ALLOWED.code()) { if (useV2Api && "api/v2/write".equals(path)) { diff --git a/src/main/java/com/influxdb/v3/client/internal/RestClient.java b/src/main/java/com/influxdb/v3/client/internal/RestClient.java index 29112515..3b6770ee 100644 --- a/src/main/java/com/influxdb/v3/client/internal/RestClient.java +++ b/src/main/java/com/influxdb/v3/client/internal/RestClient.java @@ -35,8 +35,8 @@ import java.security.cert.CertificateFactory; import java.security.cert.X509Certificate; import java.util.ArrayList; +import java.util.Collections; import java.util.List; -import java.util.Locale; import java.util.Map; import java.util.Optional; import java.util.stream.Stream; @@ -54,6 +54,7 @@ import io.netty.handler.codec.http.HttpMethod; import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.handler.codec.http.QueryStringEncoder; +import org.jspecify.annotations.NonNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -165,7 +166,19 @@ HttpResponse request(@Nonnull final String path, @Nonnull final HttpMethod method, @Nullable final byte[] data, @Nullable final Map queryParams, - @Nullable final Map headers) { + @Nullable final Map headers + ) { + return request(path, method, data, queryParams, headers, false, false); + } + + HttpResponse request(@Nonnull final String path, + @Nonnull final HttpMethod method, + @Nullable final byte[] data, + @Nullable final Map queryParams, + @Nullable final Map headers, + final boolean acceptPartial, + final boolean useV2Api + ) { QueryStringEncoder uriEncoder = new QueryStringEncoder(String.format("%s%s", baseUrl, path)); if (queryParams != null) { @@ -220,162 +233,199 @@ HttpResponse request(@Nonnull final String path, int statusCode = response.statusCode(); if (statusCode < 200 || statusCode >= 300) { - String reason; + String reason = ""; String body = response.body(); String contentType = response.headers().firstValue("Content-Type").orElse(null); - reason = formatErrorMessage(body, contentType); - - if (reason == null) { - reason = ""; - } - - if (reason.isEmpty()) { - reason = Stream.of("X-Platform-Error-Code", "X-Influx-Error", "X-InfluxDb-Error") - .map(name -> response.headers().firstValue(name).orElse(null)) - .filter(message -> message != null && !message.isEmpty()).findFirst() - .orElse(""); - } - if (reason.isEmpty()) { - reason = body; + if ("text/plain".equals(contentType)) { + var message = String.format("HTTP status code: %d; Message: %s", statusCode, body); + throw new InfluxDBApiHttpException(message, response.headers(), response.statusCode()); } - if (reason.isEmpty()) { - reason = HttpResponseStatus.valueOf(statusCode).reasonPhrase(); + JsonNode root; + try { + root = objectMapper.readTree(body); + } catch (JsonProcessingException e) { + var message = String.format("HTTP status code: %d; Message: %s", statusCode, body); + throw new InfluxDBApiHttpException(message, response.headers(), response.statusCode()); } - String message = String.format("HTTP status code: %d; Message: %s", statusCode, reason); - List lineErrors = - parsePartialWriteLineErrors(body, contentType); - if (!lineErrors.isEmpty()) { - throw new InfluxDBPartialWriteException(message, response.headers(), response.statusCode(), lineErrors); + final String rootMessage = errNonEmptyField(root, "message"); + if (rootMessage != null) { + var message = String.format("HTTP status code: %d; Message: %s", statusCode, rootMessage); + throw new InfluxDBApiHttpException(message, response.headers(), response.statusCode()); } - throw new InfluxDBApiHttpException(message, response.headers(), response.statusCode()); - } - - return response; - } - @Nullable - private String formatErrorMessage(@Nonnull final String body, @Nullable final String contentType) { - if (body.isEmpty()) { - return null; - } + if (root == null || root.toString().isEmpty()) { + reason = Stream.of("X-Platform-Error-Code", "X-Influx-Error", "X-InfluxDb-Error") + .map(name -> response.headers().firstValue(name).orElse(null)) + .filter(message -> message != null && !message.isEmpty()).findFirst() + .orElse(""); - if (!errIsJsonLikeContentType(contentType)) { - return null; - } + if (reason.isEmpty()) { + reason = HttpResponseStatus.valueOf(statusCode).reasonPhrase(); + } - try { - final JsonNode root = objectMapper.readTree(body); - if (!root.isObject()) { - return null; + var message = String.format("HTTP status code: %d; Message: %s", statusCode, reason); + throw new InfluxDBApiHttpException(message, response.headers(), response.statusCode()); } - final String rootMessage = errNonEmptyField(root, "message"); - if (rootMessage != null) { - return rootMessage; + final String error = errNonEmptyField(root, "error"); + if (error != null && !error.isEmpty()) { + reason = error; } - 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 (String detail : errFormatDataArrayDetails(dataNode)) { - if (!hasDetails) { - message.append(':'); - hasDetails = true; + if (isV3PartialWriteError(statusCode, path, acceptPartial, useV2Api, root) + ) { + // InfluxDB 3 Core/Enterprise write error format: + // {"error":"...","data":[{"error_message":"...","line_number":2,"original_line": "..."}]} + ParseLineErrorResult result = parsePartialWriteLineErrors(root); + if (result == null) { + var message = String.format("HTTP status code: %d; Message: %s", statusCode, body); + throw new InfluxDBApiHttpException(message, response.headers(), response.statusCode()); + } + var errorMsgDetails = createErrorMsgDetails(result, root); + if (!errorMsgDetails.isEmpty()) { + var sb = new StringBuilder(reason + ":"); + for (String detailError : errorMsgDetails) { + sb.append("\n\t").append(detailError); } - message.append("\n\t").append(detail); + reason = sb.toString(); } - return message.toString(); + + var message = String.format("HTTP status code: %d; Message: %s", statusCode, reason); + throw new InfluxDBPartialWriteException( + message, + response.headers(), + response.statusCode(), + result.lineErrors() + ); } // Core/Enterprise object format: // {"error":"...","data":{"error_message":"..."}} - if (isV3PartialWriteError(error) && dataNode != null && dataNode.isObject()) { - final String errorMessage = errNonEmptyField(dataNode, "error_message"); - return errorMessage == null - ? error - : error + ":\n\t" + errorMessage; + JsonNode dataNode = root.get("data"); + if (dataNode != null && dataNode.isObject()) { + final String lineNumber = Optional.ofNullable(errNonEmptyField(dataNode, "line_number")) + .orElse(""); + final String errorMessage = Optional.ofNullable(errNonEmptyField(dataNode, "error_message")) + .orElse(""); + final String originalLine = Optional.ofNullable(errNonEmptyField(dataNode, "original_line")) + .orElse(""); + if (!errorMessage.isEmpty() && (lineNumber.isEmpty() || !Utils.isInteger(lineNumber))) { + reason = reason + ":" + "\n\t" + errorMessage; + } else if (!errorMessage.isEmpty() && Utils.isInteger(lineNumber) && originalLine.isEmpty()) { + reason = String.format("%s:\n\tline %s: %s", error, lineNumber, errorMessage); + } else if (!errorMessage.isEmpty() && !originalLine.isEmpty()) { + reason = String.format("%s:\n\tline %s: %s (%s)", error, lineNumber, errorMessage, originalLine); + } } - return error; - } catch (JsonProcessingException e) { - LOG.debug("Can't parse msg from response body {}", body, e); - return null; + if (reason.isEmpty()) { + reason = body; + } + + + String message = String.format("HTTP status code: %d; Message: %s", statusCode, reason); + throw new InfluxDBApiHttpException(message, response.headers(), response.statusCode()); } + + return response; } + /** + * Creates a list of detailed error messages based on the provided parsing result and JSON root node. + * The method generates error messages from line-level parsing errors or extracts error details + * from the root node depending on the parsing result. + * + * @param result the parsing result containing details about failed lines; may be null + * @param root the JSON node representing the response data; may be null + * @return a list of generated error message details; an empty list if the input parameters are null + */ @Nonnull - private List parsePartialWriteLineErrors( - @Nonnull final String body, - @Nullable final String contentType) { - if (body.isEmpty()) { - return List.of(); + private @NonNull List createErrorMsgDetails( + @Nullable final ParseLineErrorResult result, + @Nullable final JsonNode root + ) { + if (result == null || root == null) { + return Collections.emptyList(); } - if (!errIsJsonLikeContentType(contentType)) { - return List.of(); - } - - try { - final JsonNode root = objectMapper.readTree(body); - if (!root.isObject()) { - return List.of(); - } - - final String error = errNonEmptyField(root, "error"); - final JsonNode dataNode = root.get("data"); - if (!isV3PartialWriteError(error) || dataNode == null) { - return List.of(); + List errorMsgDetails = new ArrayList<>(); + if (result.allTyped()) { + for (InfluxDBPartialWriteException.LineError lineError : result.lineErrors()) { + String s = lineError.errorMessage(); + if (lineError.lineNumber() != null + && (lineError.originalLine() != null + && !lineError.originalLine().isEmpty()) + ) { + s = String.format("line %d: %s (%s)", + lineError.lineNumber(), + lineError.errorMessage(), + lineError.originalLine()); + } else if (lineError.lineNumber() != null) { + s = String.format("line %d: %s", lineError.lineNumber(), lineError.errorMessage()); + } + errorMsgDetails.add(s); } + } else { + root.get("data").forEach(lineError -> errorMsgDetails.add(lineError.toString())); + } + return errorMsgDetails; + } - if (dataNode.isArray()) { - final ErrDataArrayItem[] parsed = errReadDataArray(dataNode); - if (parsed == null) { - return List.of(); - } + @Nullable + private ParseLineErrorResult parsePartialWriteLineErrors( + @Nullable final JsonNode root + ) { + if (root == null || !root.isObject()) { + return null; + } - final List lineErrors = new ArrayList<>(); - for (ErrDataArrayItem item : parsed) { - final InfluxDBPartialWriteException.LineError lineError = errToLineError(item); - if (lineError != null) { - lineErrors.add(lineError); + var allTyped = true; + final JsonNode dataNode = root.get("data"); + final List lineErrors = new ArrayList<>(); + if (dataNode != null && dataNode.isArray()) { + for (JsonNode node : dataNode) { + if (node.isObject()) { + final String lineNumber = Optional.ofNullable(errNonEmptyField(node, "line_number")) + .orElse(""); + final String errorMessage = Optional.ofNullable(errNonEmptyField(node, "error_message")) + .orElse(""); + final String originalLine = Optional.ofNullable(errNonEmptyField(node, "original_line")) + .orElse(""); + if (errorMessage.isEmpty() || (!lineNumber.isEmpty() && !Utils.isInteger(lineNumber))) { + allTyped = false; + continue; } - } - return lineErrors; - } - - if (dataNode.isObject()) { - try { - final ErrDataArrayItem item = objectMapper.treeToValue(dataNode, ErrDataArrayItem.class); - final InfluxDBPartialWriteException.LineError lineError = errToLineError(item); - return lineError == null ? List.of() : List.of(lineError); - } catch (JsonProcessingException e) { - return List.of(); + lineErrors.add(new InfluxDBPartialWriteException.LineError( + Utils.isInteger(lineNumber) ? Integer.parseInt(lineNumber) : null, + errorMessage, + originalLine) + ); + } else { + allTyped = false; } } - - return List.of(); - } catch (JsonProcessingException e) { - LOG.debug("Can't parse line errors from response body {}", body, e); - return List.of(); } + return new ParseLineErrorResult(lineErrors, !lineErrors.isEmpty() && allTyped); } - private boolean isV3PartialWriteError(@Nullable final String errorMessage) { - if (errorMessage == null || errorMessage.isEmpty()) { + private boolean isV3PartialWriteError(@Nonnull final Integer statusCode, + @Nonnull final String path, + final boolean isAcceptPartial, + final boolean isWriteUseV2Api, + @Nullable final JsonNode bodyRoot + ) { + final String error = errNonEmptyField(bodyRoot, "error"); + if (error == null || error.isEmpty()) { return false; } - String normalized = errorMessage.toLowerCase(Locale.ROOT); - return normalized.contains("partial write of line protocol occurred") - || normalized.contains("parsing failed for write_lp endpoint") // for Core 3.9 and earlier - || normalized.contains("line protocol parsing error"); // for Core 3.10 and later + return statusCode == 400 + && path.contains("api/v3/write_lp") + && isAcceptPartial + && !isWriteUseV2Api + && bodyRoot.path("data").isArray(); } private boolean errIsJsonLikeContentType(@Nullable final String contentType) { @@ -410,77 +460,6 @@ private String errNonEmptyField(@Nullable final JsonNode object, @Nonnull final return errNonEmptyText(object.get(fieldName)); } - @Nonnull - private List errFormatDataArrayDetails(@Nonnull final JsonNode dataNode) { - final ErrDataArrayItem[] parsed = errReadDataArray(dataNode); - if (parsed != null) { - final List details = new ArrayList<>(); - for (ErrDataArrayItem item : parsed) { - final InfluxDBPartialWriteException.LineError lineError = errToLineError(item); - if (lineError == null) { - continue; - } - - if (lineError.lineNumber() != null) { - final StringBuilder detail = new StringBuilder() - .append("line ").append(lineError.lineNumber()) - .append(": ").append(lineError.errorMessage()); - if (lineError.originalLine() != null) { - detail.append(" (").append(lineError.originalLine()).append(")"); - } - details.add(detail.toString()); - } else { - details.add(lineError.errorMessage()); - } - } - return details; - } - - final List details = new ArrayList<>(); - for (JsonNode item : dataNode) { - final String raw = errNonEmptyRawJsonToken(item); - if (raw != null) { - details.add(raw); - } - } - return details; - } - - @Nullable - private String errNonEmptyRawJsonToken(@Nonnull final JsonNode node) { - if (node.isNull()) { - return null; - } - - final String value; - if (node.isNumber() || node.isBoolean()) { - value = node.asText(); - } else { - value = node.toString(); - } - return value; - } - - @Nullable - private ErrDataArrayItem[] errReadDataArray(@Nonnull final JsonNode dataNode) { - try { - return objectMapper.treeToValue(dataNode, ErrDataArrayItem[].class); - } catch (JsonProcessingException e) { - return null; - } - } - - @Nullable - private InfluxDBPartialWriteException.LineError errToLineError(@Nullable final ErrDataArrayItem item) { - if (item == null || item.errorMessage == null || item.errorMessage.isEmpty()) { - return null; - } - - final String originalLine = - (item.originalLine == null || item.originalLine.isEmpty()) ? null : item.originalLine; - return new InfluxDBPartialWriteException.LineError(item.lineNumber, item.errorMessage, originalLine); - } - private static final class ErrDataArrayItem { @JsonProperty("error_message") private String errorMessage; @@ -528,3 +507,6 @@ private X509TrustManager getX509TrustManagerFromFile(@Nonnull final String fileP public void close() { } } + +record ParseLineErrorResult(List lineErrors, boolean allTyped) { +} \ No newline at end of file diff --git a/src/main/java/com/influxdb/v3/client/internal/Utils.java b/src/main/java/com/influxdb/v3/client/internal/Utils.java new file mode 100644 index 00000000..70cc0eb1 --- /dev/null +++ b/src/main/java/com/influxdb/v3/client/internal/Utils.java @@ -0,0 +1,46 @@ +/* + * The MIT License + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.influxdb.v3.client.internal; + +public final class Utils { + + private Utils() { + } + + /** + * Determines if the provided string can be parsed into a valid integer. + * + * @param str the string to check; may be null or empty + * @return {@code true} if the string can be parsed into an integer; {@code false} otherwise + */ + public static boolean isInteger(final String str) { + if (str == null || str.isEmpty()) { + return false; + } + try { + Integer.parseInt(str); + return true; + } catch (NumberFormatException e) { + return false; + } + } +} diff --git a/src/test/java/com/influxdb/v3/client/integration/E2ETest.java b/src/test/java/com/influxdb/v3/client/integration/E2ETest.java index 77628ed5..f84dad6b 100644 --- a/src/test/java/com/influxdb/v3/client/integration/E2ETest.java +++ b/src/test/java/com/influxdb/v3/client/integration/E2ETest.java @@ -259,18 +259,11 @@ public void testWriteErrorWithoutAcceptPartial() throws Exception { .acceptPartial(false) .build(); Throwable thrown = Assertions.catchThrowable(() -> client.writeRecord(points, options)); - Assertions.assertThat(thrown).isInstanceOf(InfluxDBPartialWriteException.class); - Assertions.assertThat(thrown.getMessage()) - .contains("line protocol parsing error"); - - InfluxDBPartialWriteException partialError = (InfluxDBPartialWriteException) thrown; - Assertions.assertThat(partialError.lineErrors()).hasSize(1); - Assertions.assertThat(partialError.lineErrors().get(0).lineNumber()).isEqualTo(2); - Assertions.assertThat(partialError.lineErrors().get(0).errorMessage()) - .isEqualTo("invalid column type for column 'temp', expected iox::column_type::field::float, " - + "got iox::column_type::field::string"); - Assertions.assertThat(partialError.lineErrors().get(0).originalLine()) - .isEqualTo("home,room=Sunroom te"); + Assertions.assertThat(thrown).isInstanceOf(InfluxDBApiHttpException.class); + Assertions.assertThat(thrown.getMessage()).isEqualTo("HTTP status code: 400; Message: line " + + "protocol parsing error:\n" + + "\tline 2: invalid column type for column 'temp', expected iox::column_type::field::float, " + + "got iox::column_type::field::string (home,room=Sunroom te)"); } } diff --git a/src/test/java/com/influxdb/v3/client/internal/RestClientTest.java b/src/test/java/com/influxdb/v3/client/internal/RestClientTest.java index e9499fb6..ac5e5dde 100644 --- a/src/test/java/com/influxdb/v3/client/internal/RestClientTest.java +++ b/src/test/java/com/influxdb/v3/client/internal/RestClientTest.java @@ -31,6 +31,7 @@ import java.time.Instant; import java.time.temporal.ChronoUnit; import java.util.Base64; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; @@ -41,6 +42,7 @@ import mockwebserver3.RecordedRequest; import okhttp3.Headers; import org.assertj.core.api.Assertions; +import org.jspecify.annotations.NonNull; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; @@ -584,22 +586,13 @@ public void errorFromBodyV3WithDataObject() { // Core/Enterprise object format Throwable thrown = catchThrowable(() -> restClient.request("api/v3/write_lp", HttpMethod.POST, null, null, null)); Assertions.assertThat(thrown) - .isInstanceOf(InfluxDBPartialWriteException.class) .isInstanceOf(InfluxDBApiHttpException.class) .hasMessage("HTTP status code: 400; Message: parsing failed for write_lp endpoint:\n" + "\tinvalid field value"); - - InfluxDBPartialWriteException partialWriteException = (InfluxDBPartialWriteException) thrown; - Assertions.assertThat(partialWriteException.statusCode()).isEqualTo(400); - Assertions.assertThat(partialWriteException.lineErrors()).hasSize(1); - InfluxDBPartialWriteException.LineError lineError = partialWriteException.lineErrors().get(0); - Assertions.assertThat(lineError.lineNumber()).isNull(); - Assertions.assertThat(lineError.errorMessage()).isEqualTo("invalid field value"); - Assertions.assertThat(lineError.originalLine()).isNull(); } @Test - public void errorFromBodyV3WithDataArray() { + public void partialErrorFromBodyV3WithDataArray() { mockServer.enqueue(createResponse(400, "application/json", null, @@ -612,7 +605,8 @@ public void errorFromBodyV3WithDataArray() { .host(baseURL) .build()); - Throwable thrown = catchThrowable(() -> restClient.request("api/v3/write_lp", HttpMethod.POST, null, null, null)); + Throwable thrown = catchThrowable(() -> restClient.request("api/v3/write_lp", HttpMethod.POST, null, + null, null, true, false)); Assertions.assertThat(thrown) .isInstanceOf(InfluxDBPartialWriteException.class) .hasMessage("HTTP status code: 400; Message: partial write of line protocol occurred:\n" @@ -630,7 +624,7 @@ public void errorFromBodyV3WithDataArray() { } @Test - public void errorFromBodyV3WithDataArrayAnyInvalidItemFallsBackToHttpException() { + public void partialErrorFromBodyV3WithInvalidDataArray() { mockServer.enqueue(createResponse(400, "application/json", null, @@ -642,157 +636,233 @@ public void errorFromBodyV3WithDataArrayAnyInvalidItemFallsBackToHttpException() .host(baseURL) .build()); - Throwable thrown = catchThrowable(() -> restClient.request("api/v3/write_lp", HttpMethod.POST, null, null, null)); + Throwable thrown = catchThrowable(() -> restClient.request("api/v3/write_lp", HttpMethod.POST, null, + null, null, true, false)); Assertions.assertThat(thrown) - .isInstanceOf(InfluxDBApiHttpException.class) - .isNotInstanceOf(InfluxDBPartialWriteException.class) + .isInstanceOf(InfluxDBPartialWriteException.class) .hasMessage("HTTP status code: 400; Message: partial write of line protocol occurred:\n" + "\t{\"error_message\":\"bad line\",\"line_number\":2,\"original_line\":\"bad lp\"}\n" + "\t{\"error_message\":\"bad line 2\",\"line_number\":\"x\",\"original_line\":\"bad lp 2\"}"); } - @ParameterizedTest(name = "{0}") - @MethodSource("errorFromBodyV3WithDataArrayCases") - public void errorFromBodyV3WithDataArrayCase(final String testName, - final String body, - final String expectedMessage) { - - mockServer.enqueue(createResponse(400, - "application/json", - null, - body)); - - restClient = new RestClient(new ClientConfig.Builder() - .host(baseURL) - .build()); - - Assertions.assertThatThrownBy( - () -> restClient.request("ping", HttpMethod.GET, null, null, null) - ) - .isInstanceOf(InfluxDBApiException.class) - .hasMessage(expectedMessage); + private static final String REJECTED_LINE = "home,room=Sunroom temp=\"hi\" 1735545610"; + private static final String REJECTED_LINE_JSON = "home,room=Sunroom temp=\\\"hi\\\" 1735545610"; + private static final String LINE_ERROR = "invalid column type for column 'temp', expected " + + "iox::column_type::field::float, got iox::column_type::field::string"; + + private List testCases() { + return List.of( + new PartialWriteTestCase( + "V3 accept partial with renamed error and non-empty array", + 400, + "application/json", + "{\"error\":\"write completed with rejected rows\"," + + "\"data\":[{\"error_message\":\"" + LINE_ERROR + "\",\"line_number\":2," + + "\"original_line\":\"" + REJECTED_LINE_JSON + "\"}]}", + false, + true, + "HTTP status code: 400; Message: write completed with rejected rows:" + + "\n\tline 2: " + LINE_ERROR + " (" + REJECTED_LINE + ")", + true, + List.of(new InfluxDBPartialWriteException.LineError(2, LINE_ERROR, REJECTED_LINE)) + ), + new PartialWriteTestCase( + "V3 accept partial without content type", + 400, + null, + "{\"error\":\"write completed with rejected rows\"," + + "\"data\":[{\"error_message\":\"" + LINE_ERROR + "\",\"line_number\":2," + + "\"original_line\":\"" + REJECTED_LINE_JSON + "\"}]}", + false, + true, + "HTTP status code: 400; Message: write completed with rejected rows:" + + "\n\tline 2: " + LINE_ERROR + " (" + REJECTED_LINE + ")", + true, + List.of(new InfluxDBPartialWriteException.LineError(2, LINE_ERROR, REJECTED_LINE)) + ), + new PartialWriteTestCase( + "V3 accept partial with malformed non-empty array", + 400, + "application/json", + "{\"error\":\"write completed with rejected rows\"," + + "\"data\":[{\"line_number\":\"invalid\"," + + "\"original_line\":\"" + REJECTED_LINE_JSON + "\"}]}", + false, + true, + "HTTP status code: 400; Message: write completed with rejected rows:" + + "\n\t{\"line_number\":\"invalid\",\"original_line\":\"" + REJECTED_LINE_JSON + + "\"}", + true, + Collections.emptyList() + ), + new PartialWriteTestCase( + "V3 accept partial with mixed primitive and typed entries", + 400, + "application/json", + "{\"error\":\"write completed with rejected rows\"," + + "\"data\":[1,{\"error_message\":\"" + LINE_ERROR + "\",\"line_number\":2," + + "\"original_line\":\"" + REJECTED_LINE_JSON + "\"}]}", + false, + true, + "HTTP status code: 400; Message: write completed with rejected rows:" + + "\n\t1\n\t{\"error_message\":\"" + LINE_ERROR + + "\",\"line_number\":2,\"original_line\":\"" + REJECTED_LINE_JSON + "\"}", + true, + List.of(new InfluxDBPartialWriteException.LineError(2, LINE_ERROR, REJECTED_LINE)) + ), + new PartialWriteTestCase( + "V3 accept partial with string entries", + 400, + "application/json", + "{\"error\":\"write completed with rejected rows\"," + + "\"data\":[\"" + REJECTED_LINE_JSON + "\"]}", + false, + true, + "HTTP status code: 400; Message: write completed with rejected rows:" + + "\n\t\"" + REJECTED_LINE_JSON + "\"", + true, + Collections.emptyList() + ), + new PartialWriteTestCase( + "V3 accept partial with error message only", + 400, + "application/json", + "{\"error\":\"write completed with rejected rows\"," + + "\"data\":[{\"error_message\":\"" + LINE_ERROR + "\"}]}", + false, + true, + "HTTP status code: 400; Message: write completed with rejected rows:" + + "\n\t" + LINE_ERROR, + true, + List.of(new InfluxDBPartialWriteException.LineError(null, LINE_ERROR, null)) + ), + new PartialWriteTestCase( + "V3 accept partial with line number but no original line", + 400, + "application/json", + "{\"error\":\"write completed with rejected rows\"," + + "\"data\":[{\"error_message\":\"" + LINE_ERROR + "\",\"line_number\":2}]}", + false, + true, + "HTTP status code: 400; Message: write completed with rejected rows:\n\tline 2: " + + LINE_ERROR, + true, + List.of(new InfluxDBPartialWriteException.LineError(2, LINE_ERROR, null)) + ), + new PartialWriteTestCase( + "V3 accept partial with entry missing error message", + 400, + "application/json", + "{\"error\":\"write completed with rejected rows\"," + + "\"data\":[{\"line_number\":2,\"original_line\":\"" + REJECTED_LINE_JSON + "\"}]}", + false, + true, + "HTTP status code: 400; Message: write completed with rejected rows:" + + "\n\t{\"line_number\":2,\"original_line\":\"" + REJECTED_LINE_JSON + "\"}", + true, + Collections.emptyList() + ), + new PartialWriteTestCase("V3 accept partial with empty array", 400, "application/json", + "{\"error\":\"write failed\",\"data\":[]}", + false, + true, + "HTTP status code: 400; Message: write failed", + true + ), + new PartialWriteTestCase("V3 accept partial with object details remains generic", 400, + "application/json", + "{\"error\":\"line protocol parsing error\",\"data\":{\"error_message\":\"" + + LINE_ERROR + "\",\"line_number\":2,\"original_line\":\"" + + REJECTED_LINE_JSON + "\"}}", + false, + true, + "HTTP status code: 400; Message: line protocol parsing error:\n\tline 2: " + + LINE_ERROR + " (" + REJECTED_LINE + ")", + false + ), + new PartialWriteTestCase("V3 reject partial with object details", 400, "application/json", + "{\"error\":\"line protocol parsing error\",\"data\":{\"error_message\":\"" + + LINE_ERROR + "\",\"line_number\":2,\"original_line\":\"" + + REJECTED_LINE_JSON + "\"}}", + false, + false, + "HTTP status code: 400; Message: line protocol parsing error:\n\tline 2: " + + LINE_ERROR + " (" + REJECTED_LINE + ")", + false + ), + new PartialWriteTestCase("V2 never returns partial write error", 400, "application/json", + "{\"error\":\"partial write of line protocol occurred\"," + + "\"data\":[{\"error_message\":\"" + + LINE_ERROR + "\",\"line_number\":2,\"original_line\":\"" + + REJECTED_LINE_JSON + "\"}]}", + true, + true, + "HTTP status code: 400; Message: partial write of line protocol occurred", + false + ), + new PartialWriteTestCase("V3 non-400 never returns partial write error", 500, + "application/json", + "{\"error\":\"partial write of line protocol occurred\"," + + "\"data\":[{\"error_message\":\"" + + LINE_ERROR + "\",\"line_number\":2,\"original_line\":\"" + REJECTED_LINE_JSON + + "\"}]}", + false, + true, + "HTTP status code: 500; Message: partial write of line protocol occurred", + false + ), + new PartialWriteTestCase("V3 scalar data remains generic", 400, "application/json", + "{\"error\":\"write failed\",\"data\":\"invalid\"}", + false, + true, + "HTTP status code: 400; Message: write failed", + false + ), + new PartialWriteTestCase("V3 empty object data remains generic", 400, "application/json", + "{\"error\":\"write failed\",\"data\":{}}", + false, + true, + "HTTP status code: 400; Message: write failed", + false + ), + new PartialWriteTestCase("V3 null data remains generic", 400, "application/json", + "{\"error\":\"write failed\",\"data\":null}", + false, + true, + "HTTP status code: 400; Message: write failed", + false + ), + new PartialWriteTestCase("V3 malformed JSON preserves raw response", 400, + "application/json", + "{\"error\":\"write failed\"", + false, + true, + "HTTP status code: 400; Message: {\"error\":\"write failed\"", + false + ) + ); } - private static Stream errorFromBodyV3WithDataArrayCases() { - return Stream.of( - Arguments.of( - "message-only detail", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"error_message\":" - + "\"only error message\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n\tonly error message" - ), - Arguments.of( - "non-object item skipped", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[null,{\"error_message\":" - + "\"bad line\",\"line_number\":2,\"original_line\":\"bad lp\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\tline 2: bad line (bad lp)" - ), - Arguments.of( - "no detail fields", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"line_number\":2}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred" - ), - Arguments.of( - "empty error_message skipped", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"error_message\":\"\"}," - + "{\"error_message\":\"bad line\",\"line_number\":2,\"original_line\":\"bad lp\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\tline 2: bad line (bad lp)" - ), - Arguments.of( - "non-object primitive item skipped", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[1,{\"error_message\":" - + "\"bad line\",\"line_number\":2,\"original_line\":\"bad lp\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\t1\n" - + "\t{\"error_message\":\"bad line\",\"line_number\":2,\"original_line\":\"bad lp\"}" - ), - Arguments.of( - "null error_message skipped", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"error_message\":null}," - + "{\"error_message\":\"bad line\",\"line_number\":2,\"original_line\":\"bad lp\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\tline 2: bad line (bad lp)" - ), - Arguments.of( - "empty original_line uses message-only detail", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"error_message\":" - + "\"only error message\",\"line_number\":2,\"original_line\":\"\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\tline 2: only error message" - ), - Arguments.of( - "missing original_line uses line-prefixed detail", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"error_message\":" - + "\"only error message\",\"line_number\":2}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\tline 2: only error message" - ), - Arguments.of( - "multiple valid details append without extra colon", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"error_message\":" - + "\"bad line\",\"line_number\":2,\"original_line\":\"bad lp\"},{\"error_message\":\"second issue\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\tline 2: bad line (bad lp)\n" - + "\tsecond issue" - ), - Arguments.of( - "array of strings fallback", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[\"bad line 1\",\"bad line 2\"]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\t\"bad line 1\"\n" - + "\t\"bad line 2\"" - ), - Arguments.of( - "array fallback skips null and renders boolean", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[null,true,\"bad line\"]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\ttrue\n" - + "\t\"bad line\"" - ), - Arguments.of( - "textual numeric line_number", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"error_message\":" - + "\"bad line\",\"line_number\":\"2\",\"original_line\":\"bad lp\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\tline 2: bad line (bad lp)" - ), - Arguments.of( - "line_number integer overflow falls back to raw token details", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"error_message\":" - + "\"bad line\",\"line_number\":2147483648,\"original_line\":\"bad lp\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\t{\"error_message\":\"bad line\",\"line_number\":2147483648,\"original_line\":\"bad lp\"}" - ), - Arguments.of( - "textual non-numeric line_number", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"error_message\":" - + "\"bad line\",\"line_number\":\"x\",\"original_line\":\"bad lp\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\t{\"error_message\":\"bad line\",\"line_number\":\"x\",\"original_line\":\"bad lp\"}" - ), - Arguments.of( - "empty textual line_number with empty original_line", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"error_message\":" - + "\"only error message\",\"line_number\":\"\",\"original_line\":\"\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n\tonly error message" - ), - Arguments.of( - "non-textual line_number", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"error_message\":" - + "\"bad line\",\"line_number\":true,\"original_line\":\"bad lp\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\t{\"error_message\":\"bad line\",\"line_number\":true,\"original_line\":\"bad lp\"}" - ), - Arguments.of( - "object line_number preserved as text", - "{\"error\":\"partial write of line protocol occurred\",\"data\":[{\"error_message\":" - + "\"bad line\",\"line_number\":{\"index\":2},\"original_line\":\"bad lp\"}]}", - "HTTP status code: 400; Message: partial write of line protocol occurred:\n" - + "\t{\"error_message\":\"bad line\",\"line_number\":{\"index\":2},\"original_line\":\"bad lp\"}" - ) - ); + @Test + public void testPartialWriteException() { + for (PartialWriteTestCase testCase : testCases()) { + mockServer.enqueue(createResponse(testCase.statusCode(), + testCase.contentType(), + null, + testCase.responseBody())); + restClient = new RestClient(new ClientConfig.Builder() + .host(baseURL) + .build()); + Throwable thrown = catchThrowable(() -> restClient.request("api/v3/write_lp", HttpMethod.POST, + null, null, null, testCase.acceptPartial(), testCase.useV2Api())); + + Assertions.assertThat(testCase.expectedMsg()).as(testCase.name()).isEqualTo(thrown.getMessage()); + if (testCase.expectPartial()) { + Assertions.assertThat(thrown).as(testCase.name()).isInstanceOf(InfluxDBPartialWriteException.class); + } else { + Assertions.assertThat(thrown).as(testCase.name()).isInstanceOf(InfluxDBApiHttpException.class); + } + } } @ParameterizedTest(name = "{0}") @@ -813,7 +883,8 @@ public void errorFromBodyV3FallbackCase(final String testName, .host(baseURL) .build()); - Throwable thrown = catchThrowable(() -> restClient.request(requestPath, HttpMethod.GET, null, null, null)); + Throwable thrown = catchThrowable(() -> + restClient.request(requestPath, HttpMethod.GET, null, null, null)); Assertions.assertThat(thrown) .isInstanceOf(expectedClass) .hasMessage(expectedMessage); @@ -1053,3 +1124,39 @@ public void getServerVersionErrorNoBody() { Assertions.assertThat(version).isEqualTo(null); } } + +record PartialWriteTestCase( + String name, + int statusCode, + String contentType, + String responseBody, + boolean useV2Api, + boolean acceptPartial, + String expectedMsg, + boolean expectPartial, + List expectedLines +) { + PartialWriteTestCase(final String name, final + int statusCode, + final String contentType, + final String responseBody, + final boolean useV2Api, + final boolean acceptPartial, + final String expectedMsg, + final boolean expectPartial) { + this(name, + statusCode, + contentType, + responseBody, + useV2Api, + acceptPartial, + expectedMsg, + expectPartial, + Collections.emptyList()); + } + + @Override + public @NonNull String toString() { + return name; + } +} \ No newline at end of file diff --git a/src/test/java/com/influxdb/v3/client/internal/UtilsTest.java b/src/test/java/com/influxdb/v3/client/internal/UtilsTest.java new file mode 100644 index 00000000..f1c3d02f --- /dev/null +++ b/src/test/java/com/influxdb/v3/client/internal/UtilsTest.java @@ -0,0 +1,78 @@ +/* + * The MIT License + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.influxdb.v3.client.internal; + +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.NullAndEmptySource; +import org.junit.jupiter.params.provider.ValueSource; + +class UtilsTest { + + @ParameterizedTest + @ValueSource(strings = { + "0", + "1", + "123", + "+123", + "-1", + "-123", + "2147483647", + "-2147483648" + }) + void isIntegerValid(final String value) { + Assertions.assertThat(Utils.isInteger(value)).isTrue(); + } + + @ParameterizedTest + @NullAndEmptySource + @ValueSource(strings = { + " ", + " ", + "\t", + "\n", + "abc", + "12a", + "a12", + "1 2", + "1.0", + "-1.5", + "1e5", + "2147483648", + "-2147483649", + "99999999999999999999" + }) + void isIntegerInvalid(final String value) { + Assertions.assertThat(Utils.isInteger(value)).isFalse(); + } + + @Test + void testNull() { + Assertions.assertThat(Utils.isInteger(null)).isFalse(); + } + + @Test + void testEmpty() { + Assertions.assertThat(Utils.isInteger("")).isFalse(); + } +}