david-streamlio opened a new pull request, #95:
URL: https://github.com/apache/pulsar-connectors/pull/95
Fixes #56
### Motivation
The `nsq` module only had `NSQConfigTest`, which exercises config parsing.
The actual source logic in `NSQSource` — subscribing to an NSQ topic/channel
via nsqlookupd and pushing records onto the Pulsar source queue — was untested.
This adds an end-to-end integration test that stands up a real NSQ cluster in
Docker and verifies messages flow from NSQ through the source.
### Modifications
- `nsq/build.gradle.kts`: add `testImplementation(libs.testcontainers)` (the
repo already pins Testcontainers centrally at 1.21.4; this module does not
depend on debezium-bom, so `GenericContainer` works directly).
-
`nsq/src/test/java/org/apache/pulsar/io/nsq/NSQSourceIntegrationTest.java`: new
test.
- Starts two `GenericContainer`s on a shared `Network` using the
`nsqio/nsq:v1.3.0` image: `nsqlookupd` and `nsqd` (with
`--lookupd-tcp-address=nsqlookupd:4160`).
- The nsq-j `Subscriber` discovers nsqd by querying nsqlookupd's HTTP
`/lookup` endpoint, which returns each nsqd's `broadcast_address` and
`tcp_port`; the client then opens a direct TCP connection to that address. For
a host-side client to reach nsqd, that advertised address must be
host-reachable, so nsqd runs with `--broadcast-address=127.0.0.1` and its TCP
port `4150` is pinned to the same host port. nsqd's and nsqlookupd's HTTP ports
are ordinary mapped ports because the test addresses them directly.
- Pre-creates the topic and channel over nsqd's HTTP API so published
messages are retained in the channel queue, seeds 5 unique messages, waits
(Awaitility) for nsqlookupd to advertise a producer, then opens `NSQSource` and
asserts it yields all published payloads.
- **Critical for source connectors:** `PushSource.read()` blocks forever
on an empty queue, so it is read on a bounded single-thread executor with a
per-read deadline (`readOne()`), never inside an Awaitility `untilAsserted`
block. A missing record surfaces as a prompt, diagnosable failure instead of
hanging CI to its timeout. The test is bounded with `@Test(timeOut = 300_000)`.
### Verifying this change
Run locally with Docker available:
`./gradlew :nsq:test`
Literal output (full module suite, all green):
```
Gradle suite > Gradle test > org.apache.pulsar.io.nsq.NSQConfigTest >
defaultValuesTest PASSED
Gradle suite > Gradle test > org.apache.pulsar.io.nsq.NSQConfigTest >
getlookupdsTest PASSED
Gradle suite > Gradle test > org.apache.pulsar.io.nsq.NSQConfigTest >
loadFromMapTest PASSED
Gradle suite > Gradle test > org.apache.pulsar.io.nsq.NSQConfigTest >
loadFromYamlFileTest PASSED
Gradle suite > Gradle test > org.apache.pulsar.io.nsq.NSQConfigTest >
missingConsumerKeyValidateTest PASSED
Gradle suite > Gradle test > org.apache.pulsar.io.nsq.NSQConfigTest >
validValidateTest PASSED
Gradle suite > Gradle test >
org.apache.pulsar.io.nsq.NSQSourceIntegrationTest > testReadFromNsq PASSED
BUILD SUCCESSFUL
```
**Mutation check (proves the test can actually fail):** temporarily pointing
the source at a registered but empty topic (nothing published) makes the
bounded read time out and fail, rather than passing or hanging:
```
org.apache.pulsar.io.nsq.NSQSourceIntegrationTest > testReadFromNsq FAILED
java.lang.AssertionError: Timed out after 60s waiting for a record from
NSQ.
The source produced no record; see the nsqd/consumer logs above.
BUILD FAILED
```
The mutation was then reverted and the suite re-run green.
### Does this pull request potentially affect one of the following parts:
- Dependencies (add or upgrade a dependency): no (test-only dependency
already pinned centrally)
- The public API: no
- The schema: no
- The default values of configurations: no
- The threading model: no
- The binary protocol: no
- The REST endpoints: no
- The admin CLI options: no
- Anything that affects deployment: no
### Documentation
- `doc-not-needed` (test-only change)
--
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]
