zhangshenghang opened a new pull request, #12741:
URL: https://github.com/apache/seatunnel/pull/12741
### Purpose of this pull request
Fixes three confirmed bugs in the InfluxDB sink, all verifiable directly
from the code:
**1. The `rp` option was silently ignored and a client was leaked** —
`InfluxDBClient#getWriteClient` applied the retention policy to a *second,
throwaway* client instead of the one it returns:
```java
InfluxDB influxdb = getInfluxDB(sinkConfig);
influxdb.setDatabase(sinkConfig.getDatabase());
setWriteProperty(getInfluxDB(sinkConfig), sinkConfig); // rp set on a new
client that is never returned nor closed
return influxdb;
```
So every write went to the database's default retention policy no matter
what `rp` was configured, and each writer additionally opened (and never
closed) a full extra InfluxDB/OkHttp client. The fix applies `setWriteProperty`
to the returned client.
**2. Null timestamp crashed the writer instead of falling back to processing
time** — `DefaultSerializer#createTimestampExtractor` sets a current-time
fallback when the configured `key_time` column is null but is missing the
`return`, so execution falls into the switch and dereferences the null
(`Long.parseLong(null)`, `(LocalDateTime) null`, unboxing `(Long) null`):
```java
if (time == null) {
builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
// missing return -> switch dereferences null below
```
**3. Every flush executed `maxRetries + 1` writes with growing duplicate
payloads** — `InfluxDBSinkWriter#flush` had no `break` after a successful
write, and called `batchPoints.points(batchList)` inside the loop.
`BatchPoints.Builder#points(Collection)` is `addAll` (append), so attempt *k*
sent *k×N* duplicate points: with `max_retries = 3`, one successful flush
issued 4 write requests carrying 1N + 2N + 3N + 4N points. The fix populates
the builder once before the loop and breaks on the first success.
Also extracts a `protected createClient()` hook (same pattern the IoTDB sink
test uses) so the writer can be unit tested without a live InfluxDB.
### Does this PR introduce _any_ user-facing change?
No API/config change. Behaviour changes:
- The documented `rp` sink option now takes effect (writes go to the
configured retention policy instead of the default one), and the leaked extra
client is gone.
- A null value in the configured `key_time` column now falls back to the
current time instead of throwing `NumberFormatException`/`NullPointerException`
and failing the job.
- A successful flush now performs exactly one write request; retries only
happen on failure and each attempt carries the same batch (no duplicate points,
no growing payloads).
### How was this patch tested?
Added unit tests (no live InfluxDB needed):
- `DefaultSerializerTest`: null `key_time` for STRING/BIGINT/TIMESTAMP
columns now serializes with the processing-time fallback (previously threw),
plus non-null pass-through, tag and no-`key_time` cases.
- `InfluxDBSinkWriterTest` (mocked `InfluxDB` injected via the new
`createClient()` hook): a successful flush writes exactly once with exactly N
points; a transient failure retries without accumulating duplicate points;
exhausted retries throw after `maxRetries + 1` attempts.
Full module suite passes: `connector-influxdb` — 49 tests, 0 failures, 0
errors. `spotless:check` passes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
