diff --git a/CHANGELOG.md b/CHANGELOG.md index ee031daf..3c5ede50 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,10 +1,11 @@ ## 1.10.0 [unreleased] -### Breaking Changes +### Features -1. [#365](https://github.com/InfluxCommunity/influxdb3-java/pull/365): Adds partial writes support and aligns write routing with v3 defaults. - See [Partial writes](https://docs.influxdata.com/influxdb3/core/write-data/http-api/v3-write-lp/#partial-writes) for more. - For InfluxDB Clustered version, set `useV2Api=true` for writing. +1. [#388](https://github.com/InfluxCommunity/influxdb3-java/pull/388): Adds partial writes support and defaults writes to the V2 API endpoint. + - `noSync` requires `useV2Api=false` and the V3 API endpoint. + - `acceptPartial` applies only when writes are sent to the V3 API endpoint and is ignored when using the V2 API endpoint. + - See [Partial writes](https://docs.influxdata.com/influxdb3/core/write-data/http-api/v3-write-lp/#partial-writes) for more. ### Bug Fixes diff --git a/README.md b/README.md index 23832327..47ee50ef 100644 --- a/README.md +++ b/README.md @@ -134,7 +134,7 @@ try { } // -// Write via v2 compatibility endpoint (InfluxDB Clustered) +// Write via V2 API endpoint (InfluxDB Clustered and InfluxDB Cloud Dedicated/Serverless) // WriteOptions useV2 = new WriteOptions.Builder() .useV2Api(true) @@ -150,7 +150,7 @@ client.writeRecord(record); #### Accept partial writes and inspect failed lines -Partial writes are enabled by default. +Partial writes are enabled by default when writing through the V3 API endpoint (`useV2Api=false`). `acceptPartial` can be configured in three ways: client defaults via `WriteOptions`, connection string / environment variable / system property (`writeAcceptPartial` / `INFLUX_WRITE_ACCEPT_PARTIAL` / `influx.writeAcceptPartial`), or per-write `WriteOptions`. Set `acceptPartial(false)` to disable partial writes. @@ -162,13 +162,13 @@ With InfluxDB Core/Enterprise, when a write request fails due to one or more inv When partial writes are disabled, any rejected line causes all lines to be rejected. InfluxDB Clustered does not return this structured partial-write error format. -#### Compatibility with InfluxDB Clustered +#### Compatibility with InfluxDB Clustered and InfluxDB Cloud Dedicated/Serverless -For InfluxDB Clustered, enable `useV2Api` for writes. +Writes use the V2 API endpoint by default. Like other write options, this can be configured in client code, connection string / environment variable / system property (`writeUseV2Api` / `INFLUX_WRITE_USE_V2_API` / `influx.writeUseV2Api`), or per-write `WriteOptions`. -If `useV2Api` is set, `acceptPartial` is ignored because this compatibility mode does not support partial-write controls. -Any rejected line causes all lines to be rejected. +`noSync` requires the V3 API endpoint, which is available with InfluxDB 3 Core/Enterprise. `acceptPartial` applies only when writes are sent to the V3 API endpoint and is ignored when using the V2 API endpoint. To use `noSync`, set `useV2Api=false`. +Note: when writes use the V2 API endpoint, `noSync=true` returns a validation error (`invalid write options: noSync requires useV2Api=false`). ### Query diff --git a/src/main/java/com/influxdb/v3/client/InfluxDBClient.java b/src/main/java/com/influxdb/v3/client/InfluxDBClient.java index 181b914c..186dfc52 100644 --- a/src/main/java/com/influxdb/v3/client/InfluxDBClient.java +++ b/src/main/java/com/influxdb/v3/client/InfluxDBClient.java @@ -558,7 +558,7 @@ static InfluxDBClient getInstance(@Nonnull final ClientConfig config) { *
  • gzipThreshold - payload size size for gzipping data
    • *
  • writeNoSync - skip waiting for WAL persistence on write
    • *
  • writeAcceptPartial - accept partial writes
    • - *
  • writeUseV2Api - use v2 compatibility write endpoint
    • + *
  • writeUseV2Api - use V2 API endpoint
    • * * * @param connectionString connection string @@ -593,7 +593,7 @@ static InfluxDBClient getInstance(@Nonnull final String connectionString) { *
  • INFLUX_GZIP_THRESHOLD - payload size size for gzipping data
    • *
  • INFLUX_WRITE_NO_SYNC - skip waiting for WAL persistence on write
    • *
  • INFLUX_WRITE_ACCEPT_PARTIAL - accept partial writes
    • - *
  • INFLUX_WRITE_USE_V2_API - use v2 compatibility write endpoint
    • + *
  • INFLUX_WRITE_USE_V2_API - use V2 API endpoint
    • * * Supported system properties: * * * @return instance of {@link InfluxDBClient} diff --git a/src/main/java/com/influxdb/v3/client/config/ClientConfig.java b/src/main/java/com/influxdb/v3/client/config/ClientConfig.java index 8acb1ade..4581d525 100644 --- a/src/main/java/com/influxdb/v3/client/config/ClientConfig.java +++ b/src/main/java/com/influxdb/v3/client/config/ClientConfig.java @@ -61,7 +61,7 @@ *
  • gzipThreshold - threshold when gzip compression is used for writing points to InfluxDB
    • *
  • writeNoSync - skip waiting for WAL persistence on write
    • *
  • writeAcceptPartial - accept partial writes
    • - *
  • writeUseV2Api - use v2 compatibility write endpoint
    • + *
  • writeUseV2Api - use V2 API endpoint
    • *
  • timeout - deprecated in 1.4.0 timeout when connecting to InfluxDB, * please use more informative properties writeTimeout and queryTimeout
    • *
  • writeTimeout - timeout when writing data to InfluxDB
    • @@ -226,9 +226,9 @@ public Boolean getWriteAcceptPartial() { } /** - * Use v2 compatibility write endpoint? + * Use V2 API endpoint? * - * @return use v2 compatibility write endpoint + * @return use V2 API endpoint */ @Nonnull public Boolean getWriteUseV2Api() { @@ -604,9 +604,9 @@ public Builder writeAcceptPartial(@Nullable final Boolean writeAcceptPartial) { } /** - * Sets whether to use v2 compatibility write endpoint. + * Sets whether to use V2 API endpoint. * - * @param writeUseV2Api use v2 compatibility write endpoint + * @param writeUseV2Api use V2 API endpoint * @return this */ @Nonnull 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 5880549c..d6fa6466 100644 --- a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java +++ b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java @@ -386,11 +386,21 @@ private void writeData(@Nonnull final List data, @Nonnull final WriteOpti try { restClient.request(path, HttpMethod.POST, body, queryParams, headers); } catch (InfluxDBApiHttpException e) { - if (!useV2Api && e.statusCode() == HttpResponseStatus.METHOD_NOT_ALLOWED.code()) { - throw new InfluxDBApiHttpException("Server doesn't support v3 write API. " - + "Use WriteOptions.Builder.useV2Api(true) for v2 compatibility endpoint.", - e.headers(), - e.statusCode()); + if (e.statusCode() == HttpResponseStatus.METHOD_NOT_ALLOWED.code()) { + if (useV2Api && "api/v2/write".equals(path)) { + throw new InfluxDBApiHttpException( + "Server doesn't support the V2 API endpoint (/api/v2/write). " + + "Set useV2Api=false to use the V3 API endpoint.", + e.headers(), + e.statusCode()); + } + if (!useV2Api && "api/v3/write_lp".equals(path)) { + throw new InfluxDBApiHttpException( + "Server doesn't support the V3 API endpoint (/api/v3/write_lp). " + + "Set useV2Api=true to use the V2 API endpoint.", + e.headers(), + e.statusCode()); + } } throw e; } 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 de2aa116..7d92b054 100644 --- a/src/main/java/com/influxdb/v3/client/internal/RestClient.java +++ b/src/main/java/com/influxdb/v3/client/internal/RestClient.java @@ -420,11 +420,14 @@ private List errFormatDataArrayDetails(@Nonnull final JsonNode dataNode) continue; } - if (lineError.lineNumber() != null - && lineError.originalLine() != null - && !lineError.originalLine().isEmpty()) { - details.add("line " + lineError.lineNumber() + ": " - + lineError.errorMessage() + " (" + lineError.originalLine() + ")"); + 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()); } @@ -434,7 +437,7 @@ private List errFormatDataArrayDetails(@Nonnull final JsonNode dataNode) final List details = new ArrayList<>(); for (JsonNode item : dataNode) { - final String raw = errNonEmptyText(item); + final String raw = errNonEmptyRawJsonToken(item); if (raw != null) { details.add(raw); } @@ -442,6 +445,21 @@ private List errFormatDataArrayDetails(@Nonnull final JsonNode dataNode) 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 { diff --git a/src/main/java/com/influxdb/v3/client/write/WriteOptions.java b/src/main/java/com/influxdb/v3/client/write/WriteOptions.java index 3487709b..02775c07 100644 --- a/src/main/java/com/influxdb/v3/client/write/WriteOptions.java +++ b/src/main/java/com/influxdb/v3/client/write/WriteOptions.java @@ -46,8 +46,8 @@ *
  • defaultTags - specifies tags to be added by default to all write operations using points.
    • *
  • tagOrder - specifies preferred tag order for point serialization.
    • *
  • noSync - skip waiting for WAL persistence on write
    • - *
  • acceptPartial - accept partial writes
    • - *
  • useV2Api - use v2 compatibility write endpoint
    • + *
  • acceptPartial - accept partial writes on the V3 API endpoint
    • + *
  • useV2Api - route writes to the V2 API endpoint
    • *
  • headers - specifies the headers to be added to write request
    • * *

    @@ -80,7 +80,7 @@ public final class WriteOptions { /** * Default UseV2Api. */ - public static final boolean DEFAULT_USE_V2_API = false; + public static final boolean DEFAULT_USE_V2_API = true; /** * Default timeout for writes in seconds. Set to {@value} @@ -294,7 +294,7 @@ public WriteOptions(@Nullable final String database, * If it is not specified then use {@link WriteOptions#DEFAULT_NO_SYNC}. * @param acceptPartial Request partial write acceptance. * If it is not specified then use {@link WriteOptions#DEFAULT_ACCEPT_PARTIAL}. - * @param useV2Api Use v2 compatibility write endpoint. + * @param useV2Api Use V2 API endpoint. * If it is not specified then use {@link WriteOptions#DEFAULT_USE_V2_API}. * @param defaultTags Default tags to be added when writing points. * @param headers The headers to be added to write request. @@ -421,7 +421,7 @@ public boolean acceptPartialSafe(@Nonnull final ClientConfig config) { /** * @param config with default value - * @return Use v2 compatibility write endpoint. + * @return Route writes to the V2 API endpoint. */ public boolean useV2ApiSafe(@Nonnull final ClientConfig config) { Arguments.checkNotNull(config, "config"); @@ -436,7 +436,7 @@ public boolean useV2ApiSafe(@Nonnull final ClientConfig config) { public void validate(@Nonnull final ClientConfig config) { Arguments.checkNotNull(config, "config"); if (useV2ApiSafe(config) && noSyncSafe(config)) { - throw new IllegalArgumentException("invalid write options: NoSync cannot be used in V2 API"); + throw new IllegalArgumentException("invalid write options: noSync requires useV2Api=false"); } } @@ -580,9 +580,9 @@ public Builder acceptPartial(@Nonnull final Boolean acceptPartial) { } /** - * Sets whether to use v2 compatibility write endpoint. + * Sets whether to use V2 API endpoint. * - * @param useV2Api use v2 compatibility write endpoint + * @param useV2Api use V2 API endpoint * @return this */ @Nonnull diff --git a/src/test/java/com/influxdb/v3/client/InfluxDBClientWriteTest.java b/src/test/java/com/influxdb/v3/client/InfluxDBClientWriteTest.java index b635fceb..f4df4250 100644 --- a/src/test/java/com/influxdb/v3/client/InfluxDBClientWriteTest.java +++ b/src/test/java/com/influxdb/v3/client/InfluxDBClientWriteTest.java @@ -112,7 +112,7 @@ void databaseParameter() throws InterruptedException { RecordedRequest request = mockServer.takeRequest(); assertThat(request).isNotNull(); assertThat(request.getUrl()).isNotNull(); - assertThat(request.getUrl().queryParameter("db")).isEqualTo("my-database"); + assertThat(request.getUrl().queryParameter("bucket")).isEqualTo("my-database"); } @Test @@ -125,7 +125,7 @@ void databaseParameterSpecified() throws InterruptedException { RecordedRequest request = mockServer.takeRequest(); assertThat(request).isNotNull(); assertThat(request.getUrl()).isNotNull(); - assertThat(request.getUrl().queryParameter("db")).isEqualTo("my-database-2"); + assertThat(request.getUrl().queryParameter("bucket")).isEqualTo("my-database-2"); } @Test @@ -153,7 +153,7 @@ void precisionParameter() throws InterruptedException { RecordedRequest request = mockServer.takeRequest(); assertThat(request).isNotNull(); assertThat(request.getUrl()).isNotNull(); - assertThat(request.getUrl().queryParameter("precision")).isEqualTo("nanosecond"); + assertThat(request.getUrl().queryParameter("precision")).isEqualTo("ns"); } @Test @@ -166,7 +166,7 @@ void precisionParameterSpecified() throws InterruptedException { RecordedRequest request = mockServer.takeRequest(); assertThat(request).isNotNull(); assertThat(request.getUrl()).isNotNull(); - assertThat(request.getUrl().queryParameter("precision")).isEqualTo("second"); + assertThat(request.getUrl().queryParameter("precision")).isEqualTo("s"); } @Test @@ -226,13 +226,14 @@ void writeRoutingCase(final String name, private static Stream writeRoutingCases() { return Stream.of( - Arguments.of("v3 noSync=false", (Consumer) b -> b.noSync(false), + Arguments.of("v2 default", (Consumer) b -> { + }, "/api/v2/write", "bucket", "ns", null, null), + Arguments.of("v3 useV2Api=false", (Consumer) b -> b.useV2Api(false), "/api/v3/write_lp", "db", "nanosecond", null, null), - Arguments.of("v3 noSync=true", (Consumer) b -> b.noSync(true), + Arguments.of("v3 noSync=true", (Consumer) b -> b.useV2Api(false).noSync(true), "/api/v3/write_lp", "db", "nanosecond", "true", null), - Arguments.of("v3 acceptPartial=true", (Consumer) b -> b.acceptPartial(true), - "/api/v3/write_lp", "db", "nanosecond", null, null), - Arguments.of("v3 acceptPartial=false", (Consumer) b -> b.acceptPartial(false), + Arguments.of("v3 acceptPartial=false", + (Consumer) b -> b.useV2Api(false).acceptPartial(false), "/api/v3/write_lp", "db", "nanosecond", null, "false"), Arguments.of("v2 useV2Api=true", (Consumer) b -> b.useV2Api(true), "/api/v2/write", "bucket", "ns", null, null), @@ -243,11 +244,14 @@ private static Stream writeRoutingCases() { } @ParameterizedTest(name = "{0}") - @MethodSource("writeV3MethodNotAllowedCases") - void writeV3MethodNotAllowedMappedError(final String name, + @MethodSource("writeMethodNotAllowedCases") + void writeMethodNotAllowedMappedError(final String name, final Consumer configure, + final String expectedPath, + final String expectedPrecision, @Nullable final String expectedNoSync, - @Nullable final String expectedAcceptPartial) throws InterruptedException { + @Nullable final String expectedAcceptPartial, + final String expectedMessage) throws InterruptedException { mockServer.enqueue(createEmptyResponse(HttpResponseStatus.METHOD_NOT_ALLOWED.code())); WriteOptions.Builder optionsBuilder = new WriteOptions.Builder().precision(WritePrecision.MS); @@ -261,24 +265,44 @@ void writeV3MethodNotAllowedMappedError(final String name, RecordedRequest request = mockServer.takeRequest(); assertThat(request).isNotNull(); assertThat(request.getUrl()).isNotNull(); - assertThat(request.getUrl().encodedPath()).isEqualTo("/api/v3/write_lp"); + assertThat(request.getUrl().encodedPath()).isEqualTo(expectedPath); assertThat(request.getUrl().queryParameter("no_sync")).isEqualTo(expectedNoSync); assertThat(request.getUrl().queryParameter("accept_partial")).isEqualTo(expectedAcceptPartial); - assertThat(request.getUrl().queryParameter("precision")).isEqualTo("millisecond"); + assertThat(request.getUrl().queryParameter("precision")).isEqualTo(expectedPrecision); assertThat(ae.statusCode()).isEqualTo(HttpResponseStatus.METHOD_NOT_ALLOWED.code()); - assertThat(ae.getMessage()).contains("Server doesn't support v3 write API. " - + "Use WriteOptions.Builder.useV2Api(true) for v2 compatibility endpoint."); + assertThat(ae.getMessage()).isEqualTo(expectedMessage); } - private static Stream writeV3MethodNotAllowedCases() { + private static Stream writeMethodNotAllowedCases() { return Stream.of( - Arguments.of("noSync=true", (Consumer) b -> b.noSync(true), "true", null), + Arguments.of("v3 noSync=true", + (Consumer) b -> b.useV2Api(false).noSync(true), + "/api/v3/write_lp", + "millisecond", + "true", + null, + "Server doesn't support the V3 API endpoint (/api/v3/write_lp). " + + "Set useV2Api=true to use the V2 API endpoint."), + Arguments.of( + "v3 acceptPartial=true", + (Consumer) b -> b.useV2Api(false).acceptPartial(true), + "/api/v3/write_lp", + "millisecond", + null, + null, + "Server doesn't support the V3 API endpoint (/api/v3/write_lp). " + + "Set useV2Api=true to use the V2 API endpoint." + ), Arguments.of( - "acceptPartial=true", - (Consumer) b -> b.acceptPartial(true), + "v2 useV2Api=true", + (Consumer) b -> b.useV2Api(true), + "/api/v2/write", + "ms", null, - null + null, + "Server doesn't support the V2 API endpoint (/api/v2/write). " + + "Set useV2Api=false to use the V3 API endpoint." ) ); } @@ -290,7 +314,7 @@ void writeUseV2ApiNoSyncValidation() { Assertions.assertThat(thrown) .isInstanceOf(IllegalArgumentException.class) - .hasMessage("invalid write options: NoSync cannot be used in V2 API"); + .hasMessage("invalid write options: noSync requires useV2Api=false"); assertThat(mockServer.getRequestCount()).isEqualTo(0); } @@ -304,7 +328,7 @@ void writeRecordWithDefaultWriteOptionsDefaultConfig() throws Exception { client.writeRecord("mem,tag=one value=1.0"); } - checkWriteCalled("/api/v3/write_lp", "DB", "nanosecond", true, null, null, false); + checkWriteCalled("/api/v2/write", "DB", "ns", false, null, null, false); } @Test @@ -314,6 +338,7 @@ void writeRecordWithDefaultWriteOptionsCustomConfig() throws Exception { ClientConfig cfg = new ClientConfig.Builder().host(baseURL).token("TOKEN".toCharArray()).database("DB") .writePrecision(WritePrecision.S) .writeNoSync(true) + .writeUseV2Api(false) .gzipThreshold(1) .build(); try (InfluxDBClient client = InfluxDBClient.getInstance(cfg)) { @@ -334,7 +359,7 @@ void writeRecordWithDefaultWriteOptionsAcceptPartialConfig() throws Exception { client.writeRecord("mem,tag=one value=1.0"); } - checkWriteCalled("/api/v3/write_lp", "DB", "nanosecond", true, null, null, false); + checkWriteCalled("/api/v2/write", "DB", "ns", false, null, null, false); } @Test @@ -361,7 +386,7 @@ void writeRecordsWithDefaultWriteOptionsDefaultConfig() throws Exception { client.writeRecords(List.of("mem,tag=one value=1.0")); } - checkWriteCalled("/api/v3/write_lp", "DB", "nanosecond", true, null, null, false); + checkWriteCalled("/api/v2/write", "DB", "ns", false, null, null, false); } @Test @@ -371,6 +396,7 @@ void writeRecordsWithDefaultWriteOptionsCustomConfig() throws Exception { ClientConfig cfg = new ClientConfig.Builder().host(baseURL).token("TOKEN".toCharArray()).database("DB") .writePrecision(WritePrecision.S) .writeNoSync(true) + .writeUseV2Api(false) .gzipThreshold(1) .build(); try (InfluxDBClient client = InfluxDBClient.getInstance(cfg)) { @@ -393,7 +419,7 @@ void writePointWithDefaultWriteOptionsDefaultConfig() throws Exception { client.writePoint(point); } - checkWriteCalled("/api/v3/write_lp", "DB", "nanosecond", true, null, null, false); + checkWriteCalled("/api/v2/write", "DB", "ns", false, null, null, false); } @Test @@ -403,6 +429,7 @@ void writePointWithDefaultWriteOptionsCustomConfig() throws Exception { ClientConfig cfg = new ClientConfig.Builder().host(baseURL).token("TOKEN".toCharArray()).database("DB") .writePrecision(WritePrecision.S) .writeNoSync(true) + .writeUseV2Api(false) .gzipThreshold(1) .build(); try (InfluxDBClient client = InfluxDBClient.getInstance(cfg)) { @@ -431,7 +458,7 @@ void writePointsWithDefaultWriteOptionsDefaultConfig() throws Exception { client.writePoints(List.of(point)); } - checkWriteCalled("/api/v3/write_lp", "DB", "nanosecond", true, null, null, false); + checkWriteCalled("/api/v2/write", "DB", "ns", false, null, null, false); } @Test @@ -441,6 +468,7 @@ void writePointsWithDefaultWriteOptionsCustomConfig() throws Exception { ClientConfig cfg = new ClientConfig.Builder().host(baseURL).token("TOKEN".toCharArray()).database("DB") .writePrecision(WritePrecision.S) .writeNoSync(true) + .writeUseV2Api(false) .gzipThreshold(1) .build(); try (InfluxDBClient client = InfluxDBClient.getInstance(cfg)) { @@ -481,7 +509,7 @@ void pointWritesIgnoreWriteOptionsPrecision( client.writePoint(point, options); } } - checkWriteCalled("/api/v3/write_lp", "DB", "nanosecond", true, null, null, false); + checkWriteCalled("/api/v2/write", "DB", "ns", false, null, null, false); } private static Stream pointPrecisionIgnoredCases() { @@ -538,8 +566,8 @@ void allParameterSpecified() throws InterruptedException { assertThat(request.getUrl()).isNotNull(); assertThat(request.getHeaders().get("Content-Type")).isEqualTo("text/plain; charset=utf-8"); assertThat(request.getHeaders().get("Content-Encoding")).isEqualTo("gzip"); - assertThat(request.getUrl().queryParameter("precision")).isEqualTo("second"); - assertThat(request.getUrl().queryParameter("db")).isEqualTo("your-database"); + assertThat(request.getUrl().queryParameter("precision")).isEqualTo("s"); + assertThat(request.getUrl().queryParameter("bucket")).isEqualTo("your-database"); } @Test diff --git a/src/test/java/com/influxdb/v3/client/config/ClientConfigTest.java b/src/test/java/com/influxdb/v3/client/config/ClientConfigTest.java index 272c3e4e..573cc288 100644 --- a/src/test/java/com/influxdb/v3/client/config/ClientConfigTest.java +++ b/src/test/java/com/influxdb/v3/client/config/ClientConfigTest.java @@ -95,7 +95,7 @@ void toStringConfig() { Assertions.assertThat(configString.contains("gzipThreshold=1000")).isEqualTo(true); Assertions.assertThat(configString).contains("writeNoSync=false"); Assertions.assertThat(configString).contains("writeAcceptPartial=true"); - Assertions.assertThat(configString).contains("writeUseV2Api=false"); + Assertions.assertThat(configString).contains("writeUseV2Api=true"); Assertions.assertThat(configString).contains("timeout=PT30S"); Assertions.assertThat(configString).contains("writeTimeout=PT35S"); Assertions.assertThat(configString).contains("queryTimeout=PT2M"); 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 cba42021..b0bdee45 100644 --- a/src/test/java/com/influxdb/v3/client/integration/E2ETest.java +++ b/src/test/java/com/influxdb/v3/client/integration/E2ETest.java @@ -204,6 +204,7 @@ public void testAcceptPartialWriteError() throws Exception { + "home,room=Sunroom temp=88i 1735545620"; WriteOptions options = new WriteOptions.Builder() + .useV2Api(false) .acceptPartial(true) .build(); @@ -254,6 +255,7 @@ public void testWriteErrorWithoutAcceptPartial() throws Exception { + "home,room=Sunroom temp=88i 1735545620"; WriteOptions options = new WriteOptions.Builder() + .useV2Api(false) .acceptPartial(false) .build(); Throwable thrown = Assertions.catchThrowable(() -> client.writeRecord(points, options)); 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 25f18744..e9499fb6 100644 --- a/src/test/java/com/influxdb/v3/client/internal/RestClientTest.java +++ b/src/test/java/com/influxdb/v3/client/internal/RestClientTest.java @@ -407,6 +407,41 @@ public void errorFromBody() { .hasMessage("HTTP status code: 401; Message: token does not have sufficient permissions"); } + @ParameterizedTest(name = "{0}") + @MethodSource("errorFromBodyScalarMessageCases") + public void errorFromBodyScalarMessage(final String testName, + final String responseBody, + final String expectedMessage) { + + mockServer.enqueue(createResponse(401, + "application/json", + null, + responseBody)); + + 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 Stream errorFromBodyScalarMessageCases() { + return Stream.of( + Arguments.of( + "numeric message", + "{\"message\":123}", + "HTTP status code: 401; Message: 123"), + Arguments.of( + "boolean message", + "{\"message\":true}", + "HTTP status code: 401; Message: true") + ); + } + @Test public void errorFromBodyIgnoredForNonJsonContentType() { mockServer.enqueue(createResponse(400, @@ -684,7 +719,15 @@ private static Stream errorFromBodyV3WithDataArrayCases() { "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\tonly error message" + "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", @@ -698,8 +741,15 @@ private static Stream errorFromBodyV3WithDataArrayCases() { "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" - + "\tbad line 1\n" - + "\tbad line 2" + + "\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", @@ -708,6 +758,13 @@ private static Stream errorFromBodyV3WithDataArrayCases() { "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\":" diff --git a/src/test/java/com/influxdb/v3/client/issues/MemoryLeakIssueTest.java b/src/test/java/com/influxdb/v3/client/issues/MemoryLeakIssueTest.java index 995ce1a3..d2bc8395 100644 --- a/src/test/java/com/influxdb/v3/client/issues/MemoryLeakIssueTest.java +++ b/src/test/java/com/influxdb/v3/client/issues/MemoryLeakIssueTest.java @@ -64,6 +64,7 @@ void testStreamCloseWithThreadInterrupt() throws Exception { .token(token.toCharArray()) .database(database) .writeNoSync(true) + .writeUseV2Api(false) .build(); try (InfluxDBClient client = InfluxDBClient.getInstance(config)) { diff --git a/src/test/java/com/influxdb/v3/client/write/WriteOptionsTest.java b/src/test/java/com/influxdb/v3/client/write/WriteOptionsTest.java index fdf95b09..54c59849 100644 --- a/src/test/java/com/influxdb/v3/client/write/WriteOptionsTest.java +++ b/src/test/java/com/influxdb/v3/client/write/WriteOptionsTest.java @@ -284,7 +284,7 @@ void optionsValidateUseV2ApiAndNoSync() { Assertions.assertThatThrownBy(() -> options.validate(config)) .isInstanceOf(IllegalArgumentException.class) - .hasMessage("invalid write options: NoSync cannot be used in V2 API"); + .hasMessage("invalid write options: noSync requires useV2Api=false"); } @Test