feat: default to v3 write endpoint and add UseV2Api compatibility mode

Author: alespour

README.md

@@ -115,11 +115,9 @@ client.writePoint(
 );
 
 //
-// Write with partial acceptance
+// Write with partial acceptance (default behavior)
 //
-WriteOptions partialWrite = new WriteOptions.Builder()
-        .acceptPartial(true)
-        .build();
+WriteOptions partialWrite = new WriteOptions.Builder().build();
 try {
     client.writeRecords(List.of(
             "temperature,region=west value=20.0",
@@ -131,6 +129,14 @@ try {
             System.out.printf("line=%s msg=%s lp=%s%n", line.lineNumber(), line.errorMessage(), line.originalLine()));
 }
 
+//
+// Write via v2 compatibility endpoint (InfluxDB Clustered)
+//
+WriteOptions useV2 = new WriteOptions.Builder()
+        .useV2Api(true)
+        .build();
+client.writeRecord("temperature,location=north value=60.0", useV2);
+
 //
 // Write by LineProtocol
 //

src/main/java/com/influxdb/v3/client/InfluxDBClient.java

@@ -558,6 +558,7 @@ static InfluxDBClient getInstance(@Nonnull final ClientConfig config) {
      *   <li>gzipThreshold - payload size size for gzipping data</li>
      *   <li>writeNoSync - skip waiting for WAL persistence on write</li>
      *   <li>writeAcceptPartial - accept partial writes</li>
+     *   <li>writeUseV2Api - use v2 compatibility write endpoint</li>
      * </ul>
      *
      * @param connectionString connection string
@@ -592,6 +593,7 @@ static InfluxDBClient getInstance(@Nonnull final String connectionString) {
      *   <li>INFLUX_GZIP_THRESHOLD - payload size size for gzipping data</li>
      *   <li>INFLUX_WRITE_NO_SYNC - skip waiting for WAL persistence on write</li>
      *   <li>INFLUX_WRITE_ACCEPT_PARTIAL - accept partial writes</li>
+     *   <li>INFLUX_WRITE_USE_V2_API - use v2 compatibility write endpoint</li>
      * </ul>
      * Supported system properties:
      * <ul>
@@ -604,6 +606,7 @@ static InfluxDBClient getInstance(@Nonnull final String connectionString) {
      *   <li>influx.gzipThreshold - payload size size for gzipping data</li>
      *   <li>influx.writeNoSync - skip waiting for WAL persistence on write</li>
      *   <li>influx.writeAcceptPartial - accept partial writes</li>
+     *   <li>influx.writeUseV2Api - use v2 compatibility write endpoint</li>
      * </ul>
      *
      * @return instance of {@link InfluxDBClient}

src/main/java/com/influxdb/v3/client/config/ClientConfig.java

@@ -58,6 +58,7 @@
  *     <li><code>gzipThreshold</code> - threshold when gzip compression is used for writing points to InfluxDB</li>
  *     <li><code>writeNoSync</code> - skip waiting for WAL persistence on write</li>
  *     <li><code>writeAcceptPartial</code> - accept partial writes</li>
+ *     <li><code>writeUseV2Api</code> - use v2 compatibility write endpoint</li>
  *     <li><code>timeout</code> - <i>deprecated in 1.4.0</i> timeout when connecting to InfluxDB,
  *     please use more informative properties <code>writeTimeout</code> and <code>queryTimeout</code></li>
  *     <li><code>writeTimeout</code> - timeout when writing data to InfluxDB</li>
@@ -109,6 +110,7 @@ public final class ClientConfig {
     private final Integer gzipThreshold;
     private final Boolean writeNoSync;
     private final Boolean writeAcceptPartial;
+    private final Boolean writeUseV2Api;
     private final Map<String, String> defaultTags;
     @Deprecated
     private final Duration timeout;
@@ -220,6 +222,16 @@ public Boolean getWriteAcceptPartial() {
         return writeAcceptPartial;
     }
 
+    /**
+     * Use v2 compatibility write endpoint?
+     *
+     * @return use v2 compatibility write endpoint
+     */
+    @Nonnull
+    public Boolean getWriteUseV2Api() {
+        return writeUseV2Api;
+    }
+
     /**
      * Gets default tags used when writing points.
      * @return default tags
@@ -383,6 +395,7 @@ public boolean equals(final Object o) {
                 && Objects.equals(gzipThreshold, that.gzipThreshold)
                 && Objects.equals(writeNoSync, that.writeNoSync)
                 && Objects.equals(writeAcceptPartial, that.writeAcceptPartial)
+                && Objects.equals(writeUseV2Api, that.writeUseV2Api)
                 && Objects.equals(defaultTags, that.defaultTags)
                 && Objects.equals(timeout, that.timeout)
                 && Objects.equals(writeTimeout, that.writeTimeout)
@@ -401,7 +414,7 @@ public boolean equals(final Object o) {
     @Override
     public int hashCode() {
         return Objects.hash(host, Arrays.hashCode(token), authScheme, organization,
-                database, writePrecision, gzipThreshold, writeNoSync, writeAcceptPartial,
+                database, writePrecision, gzipThreshold, writeNoSync, writeAcceptPartial, writeUseV2Api,
                 timeout, writeTimeout, queryTimeout, allowHttpRedirects, disableServerCertificateValidation,
                 proxy, proxyUrl, authenticator, headers,
                 defaultTags, sslRootsFilePath, disableGRPCCompression, interceptors);
@@ -417,6 +430,7 @@ public String toString() {
                 .add("gzipThreshold=" + gzipThreshold)
                 .add("writeNoSync=" + writeNoSync)
                 .add("writeAcceptPartial=" + writeAcceptPartial)
+                .add("writeUseV2Api=" + writeUseV2Api)
                 .add("timeout=" + timeout)
                 .add("writeTimeout=" + writeTimeout)
                 .add("queryTimeout=" + queryTimeout)
@@ -447,6 +461,7 @@ public static final class Builder {
         private Integer gzipThreshold;
         private Boolean writeNoSync;
         private Boolean writeAcceptPartial;
+        private Boolean writeUseV2Api;
         private Map<String, String> defaultTags;
         @Deprecated
         private Duration timeout;
@@ -582,6 +597,19 @@ public Builder writeAcceptPartial(@Nullable final Boolean writeAcceptPartial) {
             return this;
         }
 
+        /**
+         * Sets whether to use v2 compatibility write endpoint.
+         *
+         * @param writeUseV2Api use v2 compatibility write endpoint
+         * @return this
+         */
+        @Nonnull
+        public Builder writeUseV2Api(@Nullable final Boolean writeUseV2Api) {
+
+            this.writeUseV2Api = writeUseV2Api;
+            return this;
+        }
+
         /**
          * Sets default tags to be written with points.
          *
@@ -831,6 +859,9 @@ public ClientConfig build(@Nonnull final String connectionString) throws Malform
             if (parameters.containsKey("writeAcceptPartial")) {
                 this.writeAcceptPartial(Boolean.parseBoolean(parameters.get("writeAcceptPartial")));
             }
+            if (parameters.containsKey("writeUseV2Api")) {
+                this.writeUseV2Api(Boolean.parseBoolean(parameters.get("writeUseV2Api")));
+            }
             if (parameters.containsKey("disableGRPCCompression")) {
                 this.disableGRPCCompression(Boolean.parseBoolean(parameters.get("disableGRPCCompression")));
             }
@@ -890,6 +921,10 @@ public ClientConfig build(@Nonnull final Map<String, String> env, final Properti
             if (writeAcceptPartial != null) {
                 this.writeAcceptPartial(Boolean.parseBoolean(writeAcceptPartial));
             }
+            final String writeUseV2Api = get.apply("INFLUX_WRITE_USE_V2_API", "influx.writeUseV2Api");
+            if (writeUseV2Api != null) {
+                this.writeUseV2Api(Boolean.parseBoolean(writeUseV2Api));
+            }
             final String writeTimeout = get.apply("INFLUX_WRITE_TIMEOUT", "influx.writeTimeout");
             if (writeTimeout != null) {
                 long to = Long.parseLong(writeTimeout);
@@ -949,6 +984,9 @@ private ClientConfig(@Nonnull final Builder builder) {
         writeAcceptPartial = builder.writeAcceptPartial != null
                 ? builder.writeAcceptPartial
                 : WriteOptions.DEFAULT_ACCEPT_PARTIAL;
+        writeUseV2Api = builder.writeUseV2Api != null
+                ? builder.writeUseV2Api
+                : WriteOptions.DEFAULT_USE_V2_API;
         defaultTags = builder.defaultTags;
         timeout = builder.timeout != null ? builder.timeout : Duration.ofSeconds(WriteOptions.DEFAULT_WRITE_TIMEOUT);
         writeTimeout = builder.writeTimeout != null

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

@@ -305,14 +305,21 @@ private <T> void writeData(@Nonnull final List<T> data, @Nonnull final WriteOpti
         }
 
         WritePrecision precision = options.precisionSafe(config);
+        options.validate(config);
 
         String path;
         Map<String, String> queryParams;
         boolean noSync = options.noSyncSafe(config);
         boolean acceptPartial = options.acceptPartialSafe(config);
-        boolean useV3Write = noSync || acceptPartial;
-        if (useV3Write) {
-            // no_sync=true and accept_partial=true are supported only in the v3 API.
+        boolean useV2Api = options.useV2ApiSafe(config);
+        if (useV2Api) {
+            path = "api/v2/write";
+            queryParams = new HashMap<>() {{
+                put("org", config.getOrganization());
+                put("bucket", database);
+                put("precision", WritePrecisionConverter.toV2ApiString(precision));
+            }};
+        } else {
             path = "api/v3/write_lp";
             queryParams = new HashMap<>() {{
                 put("org", config.getOrganization());
@@ -322,17 +329,9 @@ private <T> void writeData(@Nonnull final List<T> data, @Nonnull final WriteOpti
             if (noSync) {
                 queryParams.put("no_sync", "true");
             }
-            if (acceptPartial) {
-                queryParams.put("accept_partial", "true");
+            if (!acceptPartial) {
+                queryParams.put("accept_partial", "false");
             }
-        } else {
-            // By default, use the v2 API.
-            path = "api/v2/write";
-            queryParams = new HashMap<>() {{
-                put("org", config.getOrganization());
-                put("bucket", database);
-                put("precision", WritePrecisionConverter.toV2ApiString(precision));
-            }};
         }
 
         Map<String, String> defaultTags = options.defaultTagsSafe(config);
@@ -380,10 +379,9 @@ private <T> void writeData(@Nonnull final List<T> data, @Nonnull final WriteOpti
         try {
             restClient.request(path, HttpMethod.POST, body, queryParams, headers);
         } catch (InfluxDBApiHttpException e) {
-            if (useV3Write && e.statusCode() == HttpResponseStatus.METHOD_NOT_ALLOWED.code()) {
-                // Server does not support the v3 write API, can't use v3-only write options.
-                throw new InfluxDBApiHttpException("Server doesn't support write with NoSync=true "
-                        + "or AcceptPartial=true (supported by InfluxDB 3 Core/Enterprise servers only).",
+            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());
             }

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

@@ -36,6 +36,7 @@
 import java.security.cert.X509Certificate;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Stream;
@@ -245,7 +246,7 @@ HttpResponse<String> request(@Nonnull final String path,
 
             String message = String.format("HTTP status code: %d; Message: %s", statusCode, reason);
             List<InfluxDBPartialWriteException.LineError> lineErrors =
-                    parsePartialWriteLineErrors(path, body, contentType);
+                    parsePartialWriteLineErrors(body, contentType);
             if (!lineErrors.isEmpty()) {
                 throw new InfluxDBPartialWriteException(message, response.headers(), response.statusCode(), lineErrors);
             }
@@ -312,13 +313,8 @@ private String formatErrorMessage(@Nonnull final String body, @Nullable final St
 
     @Nonnull
     private List<InfluxDBPartialWriteException.LineError> parsePartialWriteLineErrors(
-            @Nonnull final String path,
             @Nonnull final String body,
             @Nullable final String contentType) {
-        if (!isWriteEndpoint(path)) {
-            return List.of();
-        }
-
         if (body.isEmpty()) {
             return List.of();
         }
@@ -337,7 +333,7 @@ private List<InfluxDBPartialWriteException.LineError> parsePartialWriteLineError
 
             final String error = errNonEmptyField(root, "error");
             final JsonNode dataNode = root.get("data");
-            if (error == null || dataNode == null) {
+            if (!isV3PartialWriteError(error) || dataNode == null) {
                 return List.of();
             }
 
@@ -374,8 +370,13 @@ private List<InfluxDBPartialWriteException.LineError> parsePartialWriteLineError
         }
     }
 
-    private boolean isWriteEndpoint(@Nonnull final String path) {
-        return "api/v2/write".equals(path) || "api/v3/write_lp".equals(path);
+    private boolean isV3PartialWriteError(@Nullable final String errorMessage) {
+        if (errorMessage == null || errorMessage.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");
     }
 
     @Nullable