Skip to content

Commit fc6e19c

Browse files
jansimonbjansimonb
committed
fix: use WriteOptions from ClientConfig on write
1 parent f111cf4 commit fc6e19c

File tree

2 files changed

+173
-9
lines changed

2 files changed

+173
-9
lines changed

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

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,7 @@ public final class InfluxDBClientImpl implements InfluxDBClient {
8181

8282
private final RestClient restClient;
8383
private final FlightSqlClient flightSqlClient;
84+
private final WriteOptions emptyWriteOptions;
8485

8586
/**
8687
* Creates an instance using the specified config.
@@ -110,11 +111,12 @@ public InfluxDBClientImpl(@Nonnull final ClientConfig config) {
110111
this.config = config;
111112
this.restClient = restClient != null ? restClient : new RestClient(config);
112113
this.flightSqlClient = flightSqlClient != null ? flightSqlClient : new FlightSqlClient(config);
114+
this.emptyWriteOptions = new WriteOptions(null);
113115
}
114116

115117
@Override
116118
public void writeRecord(@Nullable final String record) {
117-
writeRecord(record, WriteOptions.DEFAULTS);
119+
writeRecord(record, emptyWriteOptions);
118120
}
119121

120122
@Override
@@ -128,7 +130,7 @@ public void writeRecord(@Nullable final String record, @Nonnull final WriteOptio
128130

129131
@Override
130132
public void writeRecords(@Nonnull final List<String> records) {
131-
writeRecords(records, WriteOptions.DEFAULTS);
133+
writeRecords(records, emptyWriteOptions);
132134
}
133135

134136
@Override
@@ -138,7 +140,7 @@ public void writeRecords(@Nonnull final List<String> records, @Nonnull final Wri
138140

139141
@Override
140142
public void writePoint(@Nullable final Point point) {
141-
writePoint(point, WriteOptions.DEFAULTS);
143+
writePoint(point, emptyWriteOptions);
142144
}
143145

144146
@Override
@@ -152,7 +154,7 @@ public void writePoint(@Nullable final Point point, @Nonnull final WriteOptions
152154

153155
@Override
154156
public void writePoints(@Nonnull final List<Point> points) {
155-
writePoints(points, WriteOptions.DEFAULTS);
157+
writePoints(points, emptyWriteOptions);
156158
}
157159

158160
@Override

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

Lines changed: 167 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,15 +23,19 @@
2323

2424
import java.util.Arrays;
2525
import java.util.Collections;
26+
import java.util.List;
2627
import java.util.Map;
2728

2829
import io.netty.handler.codec.http.HttpResponseStatus;
30+
import okhttp3.HttpUrl;
2931
import okhttp3.mockwebserver.RecordedRequest;
3032
import org.assertj.core.api.Assertions;
33+
import org.jetbrains.annotations.NotNull;
3134
import org.junit.jupiter.api.AfterEach;
3235
import org.junit.jupiter.api.BeforeEach;
3336
import org.junit.jupiter.api.Test;
3437

38+
import com.influxdb.v3.client.config.ClientConfig;
3539
import com.influxdb.v3.client.write.WriteOptions;
3640
import com.influxdb.v3.client.write.WritePrecision;
3741

@@ -223,10 +227,8 @@ void writeNoSyncTrueOnV2ServerThrowsException() throws InterruptedException {
223227
mockServer.enqueue(createEmptyResponse(HttpResponseStatus.METHOD_NOT_ALLOWED.code()));
224228

225229
InfluxDBApiHttpException ae = org.junit.jupiter.api.Assertions.assertThrows(InfluxDBApiHttpException.class,
226-
() -> {
227-
client.writeRecord("mem,tag=one value=1.0",
228-
new WriteOptions.Builder().precision(WritePrecision.MS).noSync(true).build());
229-
}
230+
() -> client.writeRecord("mem,tag=one value=1.0",
231+
new WriteOptions.Builder().precision(WritePrecision.MS).noSync(true).build())
230232
);
231233

232234
assertThat(mockServer.getRequestCount()).isEqualTo(1);
@@ -242,6 +244,166 @@ void writeNoSyncTrueOnV2ServerThrowsException() throws InterruptedException {
242244
+ " (supported by InfluxDB 3 Core/Enterprise servers only).");
243245
}
244246

247+
@Test
248+
void writeRecordWithDefaultWriteOptionsDefaultConfig() throws Exception {
249+
mockServer.enqueue(createResponse(200));
250+
251+
ClientConfig cfg = new ClientConfig.Builder().host(baseURL).token("TOKEN".toCharArray()).database("DB")
252+
.build();
253+
try (InfluxDBClient client = InfluxDBClient.getInstance(cfg)) {
254+
client.writeRecord("mem,tag=one value=1.0");
255+
}
256+
257+
checkWriteCalled("/api/v2/write", "DB", "ns", false, false);
258+
}
259+
260+
@Test
261+
void writeRecordWithDefaultWriteOptionsCustomConfig() throws Exception {
262+
mockServer.enqueue(createResponse(200));
263+
264+
ClientConfig cfg = new ClientConfig.Builder().host(baseURL).token("TOKEN".toCharArray()).database("DB")
265+
.writePrecision(WritePrecision.S)
266+
.writeNoSync(true)
267+
.gzipThreshold(1)
268+
.build();
269+
try (InfluxDBClient client = InfluxDBClient.getInstance(cfg)) {
270+
client.writeRecord("mem,tag=one value=1.0");
271+
}
272+
273+
checkWriteCalled("/api/v3/write_lp", "DB", "second", true, true);
274+
}
275+
276+
@Test
277+
void writeRecordsWithDefaultWriteOptionsDefaultConfig() throws Exception {
278+
mockServer.enqueue(createResponse(200));
279+
280+
ClientConfig cfg = new ClientConfig.Builder().host(baseURL).token("TOKEN".toCharArray()).database("DB")
281+
.build();
282+
try (InfluxDBClient client = InfluxDBClient.getInstance(cfg)) {
283+
client.writeRecords(List.of("mem,tag=one value=1.0"));
284+
}
285+
286+
checkWriteCalled("/api/v2/write", "DB", "ns", false, false);
287+
}
288+
289+
@Test
290+
void writeRecordsWithDefaultWriteOptionsCustomConfig() throws Exception {
291+
mockServer.enqueue(createResponse(200));
292+
293+
ClientConfig cfg = new ClientConfig.Builder().host(baseURL).token("TOKEN".toCharArray()).database("DB")
294+
.writePrecision(WritePrecision.S)
295+
.writeNoSync(true)
296+
.gzipThreshold(1)
297+
.build();
298+
try (InfluxDBClient client = InfluxDBClient.getInstance(cfg)) {
299+
client.writeRecords(List.of("mem,tag=one value=1.0"));
300+
}
301+
302+
checkWriteCalled("/api/v3/write_lp", "DB", "second", true, true);
303+
}
304+
305+
@Test
306+
void writePointWithDefaultWriteOptionsDefaultConfig() throws Exception {
307+
mockServer.enqueue(createResponse(200));
308+
309+
ClientConfig cfg = new ClientConfig.Builder().host(baseURL).token("TOKEN".toCharArray()).database("DB")
310+
.build();
311+
try (InfluxDBClient client = InfluxDBClient.getInstance(cfg)) {
312+
Point point = new Point("mem");
313+
point.setTag("tag", "one");
314+
point.setField("value", 1.0);
315+
client.writePoint(point);
316+
}
317+
318+
checkWriteCalled("/api/v2/write", "DB", "ns", false, false);
319+
}
320+
321+
@Test
322+
void writePointWithDefaultWriteOptionsCustomConfig() throws Exception {
323+
mockServer.enqueue(createResponse(200));
324+
325+
ClientConfig cfg = new ClientConfig.Builder().host(baseURL).token("TOKEN".toCharArray()).database("DB")
326+
.writePrecision(WritePrecision.S)
327+
.writeNoSync(true)
328+
.gzipThreshold(1)
329+
.build();
330+
try (InfluxDBClient client = InfluxDBClient.getInstance(cfg)) {
331+
Point point = new Point("mem");
332+
point.setTag("tag", "one");
333+
point.setField("value", 1.0);
334+
client.writePoint(point);
335+
}
336+
337+
checkWriteCalled("/api/v3/write_lp", "DB", "second", true, true);
338+
}
339+
340+
@Test
341+
void writePointsWithDefaultWriteOptionsDefaultConfig() throws Exception {
342+
mockServer.enqueue(createResponse(200));
343+
344+
ClientConfig cfg = new ClientConfig.Builder().host(baseURL).token("TOKEN".toCharArray()).database("DB")
345+
.build();
346+
try (InfluxDBClient client = InfluxDBClient.getInstance(cfg)) {
347+
Point point = new Point("mem");
348+
point.setTag("tag", "one");
349+
point.setField("value", 1.0);
350+
client.writePoints(List.of(point));
351+
}
352+
353+
checkWriteCalled("/api/v2/write", "DB", "ns", false, false);
354+
}
355+
356+
@Test
357+
void writePointsWithDefaultWriteOptionsCustomConfig() throws Exception {
358+
mockServer.enqueue(createResponse(200));
359+
360+
ClientConfig cfg = new ClientConfig.Builder().host(baseURL).token("TOKEN".toCharArray()).database("DB")
361+
.writePrecision(WritePrecision.S)
362+
.writeNoSync(true)
363+
.gzipThreshold(1)
364+
.build();
365+
try (InfluxDBClient client = InfluxDBClient.getInstance(cfg)) {
366+
Point point = new Point("mem");
367+
point.setTag("tag", "one");
368+
point.setField("value", 1.0);
369+
client.writePoints(List.of(point));
370+
}
371+
372+
checkWriteCalled("/api/v3/write_lp", "DB", "second", true, true);
373+
}
374+
375+
private void checkWriteCalled(String expectedPath, String expectedDB, String expectedPrecision,
376+
boolean expectedNoSync, boolean expectedGzip) throws InterruptedException {
377+
RecordedRequest request = assertThatServerRequested();
378+
HttpUrl requestUrl = request.getRequestUrl();
379+
assertThat(requestUrl).isNotNull();
380+
assertThat(requestUrl.encodedPath()).isEqualTo(expectedPath);
381+
if (expectedNoSync) {
382+
assertThat(requestUrl.queryParameter("db")).isEqualTo(expectedDB);
383+
} else {
384+
assertThat(requestUrl.queryParameter("bucket")).isEqualTo(expectedDB);
385+
}
386+
assertThat(requestUrl.queryParameter("precision")).isEqualTo(expectedPrecision);
387+
if (expectedNoSync) {
388+
assertThat(requestUrl.queryParameter("no_sync")).isEqualTo("true");
389+
} else {
390+
assertThat(requestUrl.queryParameter("no_sync")).isNull();
391+
}
392+
if (expectedGzip) {
393+
assertThat(request.getHeader("Content-Encoding")).isEqualTo("gzip");
394+
} else {
395+
assertThat(request.getHeader("Content-Encoding")).isNull();
396+
}
397+
}
398+
399+
@NotNull
400+
private RecordedRequest assertThatServerRequested() throws InterruptedException {
401+
assertThat(mockServer.getRequestCount()).isEqualTo(1);
402+
RecordedRequest request = mockServer.takeRequest();
403+
assertThat(request).isNotNull();
404+
return request;
405+
}
406+
245407
@Test
246408
void allParameterSpecified() throws InterruptedException {
247409
mockServer.enqueue(createResponse(200));
@@ -344,7 +506,7 @@ void defaultTags() throws InterruptedException {
344506
}
345507

346508
@Test
347-
public void retryHandled429Test() throws InterruptedException {
509+
public void retryHandled429Test() {
348510
mockServer.enqueue(createResponse(429)
349511
.setBody("{ \"message\" : \"Too Many Requests\" }")
350512
.setHeader("retry-after", "42")

0 commit comments

Comments
 (0)