david-streamlio opened a new pull request, #98:
URL: https://github.com/apache/pulsar-connectors/pull/98
Fixes #48
### Motivation
The `canal` source connector (`CanalStringSource`) had **zero tests**. Canal
is Alibaba's MySQL-binlog CDC: the source connects to a canal-server, which
reads the MySQL binlog. Nothing verified that a MySQL change actually flows
through canal-server and out of the source as a CDC event.
### Modifications
Add `CanalStringSourceIntegrationTest`, which stands up the full pipeline
with Testcontainers:
```
MySQL (mysql:8.0, ROW binlog) <--dump-- canal-server
(canal/canal-server:v1.1.7) <--tcp-- CanalStringSource
```
- Two `GenericContainer`s share a `Network`. canal reaches MySQL through the
`mysql` network alias; the source (in the test JVM) reaches canal through the
mapped `11111` port.
- canal-server `v1.1.7` matches the module's `canal.client`/`canal.protocol`
`1.1.7`. The image does no env-var substitution, so the instance config is
mounted from `src/test/resources/canal/instance.properties`
(`canal.instance.master.address=mysql:3306`, root/binlog credentials,
`connectionCharset=UTF-8`, `filter.regex=testdb\..*`).
- Both containers use `.withStartupTimeout(Duration.ofMinutes(3))` so a
stalled pull can't hang the job.
- The tracked table is created **before** canal starts (so its DDL isn't in
the tracked stream); an `INSERT` is made **after** the source subscribes, and
the resulting CDC event is read back on a **bounded worker thread with a
deadline** — a missing event fails fast instead of blocking `read()` forever.
`@Test(timeOut = 600_000)`.
- `canal/build.gradle.kts`: add `testImplementation(libs.testcontainers)`
and `testImplementation(libs.mysql.connector.j)`.
**Required dependency fix (protobuf).** While writing the test, the canal
client turned out to be **non-functional against the platform-enforced
`protobuf-java` 4.31.1**. `canal.protocol` 1.1.7's generated `CanalPacket`
calls `GeneratedMessageV3.makeExtensionsImmutable()`, which was removed in
protobuf 4.x, so `SimpleCanalConnector.connect()` throws:
```
java.lang.NoSuchMethodError: 'void
com.alibaba.otter.canal.protocol.CanalPacket$Packet.makeExtensionsImmutable()'
```
This affects the connector at runtime, not just the test — the client cannot
complete the canal-server handshake at all. The fix is to pin `protobuf-java`
back to the `3.6.1` version canal declares (`resolutionStrategy.force` in
`canal/build.gradle.kts`), which also flows into the module's NAR. Reviewers
should confirm this protobuf choice is acceptable for the shipped NAR.
### Verifying this change
Run locally with Docker:
```
./gradlew :canal:test --tests
"org.apache.pulsar.io.canal.CanalStringSourceIntegrationTest"
```
**Result — PASS.** The source read back the real INSERT event:
```
Gradle test > ...CanalStringSourceIntegrationTest > testCanalCdcEvents PASSED
Received CDC record: key=1, message=[{"data":[{...
"columnName":"name","columnValue":"canal-widget",...}],
"database":"testdb", ..., "isDdl":false, "table":"products",
"type":"INSERT"}]
BUILD SUCCESSFUL
```
**Mutation check (negative control).** With the `INSERT` removed (no DB
change), the bounded read must *fail*, not hang:
```
...CanalStringSourceIntegrationTest > testCanalCdcEvents FAILED
java.lang.AssertionError: Timed out after 180s waiting for a CDC record
from canal.
The pipeline produced no record; see the canal-server logs above.
1 test completed, 1 failed
BUILD FAILED in 3m 25s (total 206s — bounded, did not hang; well under the
600s test timeout)
```
Restoring the `INSERT` returns the suite to green (re-run confirmed).
`./gradlew :canal:spotlessCheck :canal:nar` also pass.
### Does this pull request potentially affect one of the following parts
- Dependencies (add or upgrade a dependency): **yes** — adds two
`testImplementation` deps and pins `protobuf-java` to 3.6.1 for the `canal`
module (see above).
- Anything that affects deployment: no.
--
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]
