david-streamlio opened a new pull request, #73:
URL: https://github.com/apache/pulsar-connectors/pull/73
Fixes #53
### Motivation
The `influxdb` module has no integration coverage. `InfluxDBSinkTest`
verifies only that the sink calls `WriteApiBlocking.writePoints()` on a
**mock**, which leaves the parts most likely to break untested: whether
InfluxDB actually accepts the points the sink builds, and whether the
measurement, tags, fields, and timestamp survive the round trip.
Separately, that mock test is **flaky**, and it is not a hypothetical:
`testOpenWriteCloseJson` failed on unrelated PR #39 today with `Wanted but not
invoked: writeApiBlocking.writePoints(<any List>)`.
### Root cause of the flake
`BatchSink.write()` queues the record and, once `batchSize` records are
pending, dispatches the flush to a **background** `ScheduledExecutorService`:
```java
if (currentSize >= batchSize) {
flushExecutor.execute(this::flush);
}
```
The test waited a fixed `Thread.sleep(100)` and then asserted `writePoints`
had been called. Under CI load the executor had not yet run the flush, so the
verification failed. Nothing was wrong with the sink.
### Modifications
**New `InfluxDBSinkIntegrationTest`** — runs the v2 sink against a real
`influxdb:2.7` container (Testcontainers `GenericContainer`, initialized with
`DOCKER_INFLUXDB_INIT_MODE=setup`), writes two records through the sink's real
client, then queries them back with Flux and asserts the measurement, tags,
field value, and persisted timestamp. `batchTimeMs` is set high so the
assertion exercises the batch-size flush path rather than the timer. Bounded
with a `timeOut`.
**Deflake `InfluxDBSinkTest`** — replace the fixed `Thread.sleep(100)` with
an Awaitility poll, so the test waits for the asynchronous flush instead of
racing it. The assertion is unchanged.
### Verifying this change
Full module suite passes locally:
```
InfluxDBSinkTest > testOpenWriteCloseJson PASSED
InfluxDBSinkIntegrationTest > testWritesReachInfluxDb PASSED
BUILD SUCCESSFUL
```
The integration test was also checked for vacuous success: pointing the sink
at a nonexistent bucket makes it fail with `ConditionTimeoutException ... for
input of <[]>` rather than passing silently, confirming it genuinely observes
the data reaching InfluxDB.
### Notes
- Covers the **v2** sink. The v1 sink (`influxdb/v1`) still has mock-only
coverage; a follow-up could add an `influxdb:1.8` container test using the same
pattern.
- Uses `GenericContainer` rather than Testcontainers' `InfluxDBContainer`,
which targets the 1.x line and does not perform the v2 `setup` initialization.
--
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]
