david-streamlio opened a new pull request, #96:
URL: https://github.com/apache/pulsar-connectors/pull/96
Fixes #47
### Motivation
The `aerospike` module had **zero tests**. This adds the first one: an
end-to-end Testcontainers integration test that runs a real Aerospike Community
Edition server, writes records through `AerospikeStringSink`, and reads them
back with an independent `AerospikeClient` to prove the data round-trips.
Writing the test surfaced a latent defect in the sink itself.
`AerospikeAbstractSink.write()` stored values with `Value.getAsBlob(...)` (Java
object serialization). In the version pinned by the project
(`com.aerospike:aerospike-client-bc:4.5.0`), `Value$BlobValue.estimateSize()`
**unconditionally** throws `AerospikeException(-10, "Object serializer has been
disabled")` — object serialization was removed from that client build. As a
result every write failed and the sink could not persist anything.
### Modifications
- `AerospikeAbstractSink`: replaced `Value.getAsBlob(keyValue.getValue())`
with `Value.get(keyValue.getValue())`. For the string sink this stores a
native, first-class string bin (interoperable and directly queryable) instead
of a dead Java-serialized blob. This is the minimal change needed for the sink
to actually write with the pinned client.
- `aerospike/build.gradle.kts`:
- `testImplementation(libs.testcontainers)` for the `GenericContainer`
harness.
- `testRuntimeOnly(libs.jakarta.xml.bind.api)` — the 4.5.0 client calls
`javax.xml.bind.DatatypeConverter` (removed from the JDK since Java 11) while
parsing partition maps. Pulsar's runtime supplies JAXB in production; the test
classpath must provide it explicitly.
- New `AerospikeStringSinkIntegrationTest`:
- Starts `aerospike/aerospike-server:6.4.0.2` (CE), exposing port 3000.
- Container readiness uses `Wait.forLogMessage(".*service ready.*", 1)`
with an explicit `withStartupTimeout(Duration.ofMinutes(2))` so a stalled image
pull can never hang CI.
- Awaitility-gates the client connection and probes the `test` namespace
with a real write+read before opening the sink (the node logs "service ready"
slightly before partitions finish rebalancing).
- Writes 5 records through the sink (keyed, string values), waits on the
sink's own ack callbacks rather than sleeping, then reads each key back via
`AerospikeClient.get(...)` and asserts the bin value. Bounded with
`@Test(timeOut = 300_000)`.
### Verifying this change
`./gradlew :aerospike:test` (actual execution, `--rerun-tasks`):
```
Gradle suite > Gradle test >
org.apache.pulsar.io.aerospike.AerospikeStringSinkIntegrationTest >
testWriteAndReadBack PASSED
BUILD SUCCESSFUL
```
Ran 3x back-to-back with the sink's default config — green every time (no
flakiness).
`./gradlew :aerospike:check` (spotless/style): `BUILD SUCCESSFUL`.
**Mutation check** (proving the assertion is real, not vacuous): temporarily
changed the expected bin value to a string never written
(`MUTATION-never-written-<i>`) and re-ran — the test failed exactly as it
should:
```
testWriteAndReadBack FAILED
java.lang.AssertionError: bin value for key-0 should match what was written
expected [MUTATION-never-written-0] but found [value-0]
```
Reverting the mutation returns the test to green. This confirms the
assertion actually reads sink-written data back out of Aerospike.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01KsD6v3Cm2STC9B4brKE39x
--
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]
