david-streamlio opened a new pull request, #76: URL: https://github.com/apache/pulsar-connectors/pull/76
Fixes #54 ### Motivation The `kafka/` module's tests (`KafkaAbstractSourceTest`, `KafkaAbstractSinkTest`, `KafkaBytesSourceTest`, `ByteBufferSchemaWrapperTest`) are all mock-based — nothing exercises the connectors against a real Kafka broker. That leaves the end-to-end wiring (producer/consumer config translation, serialization, ack/commit flow) unverified. This mirrors the ClickHouse JDBC sink integration test added in #32 / commit e29bb2b. ### Modifications Adds `kafka/src/test/java/org/apache/pulsar/io/kafka/KafkaConnectorIntegrationTest.java`, a Testcontainers test covering **both** directions against a real `org.testcontainers.kafka.KafkaContainer` (`apache/kafka:3.9.1`), started once per class: - **Sink** (`testSinkWritesToKafka`): opens `KafkaBytesSink`, writes keyed byte records through it, waits for the producer-callback acks, then reads the topic back with a plain `KafkaConsumer` and asserts every key/payload. - **Source** (`testSourceReadsFromKafka`): produces messages with a plain `KafkaProducer`, opens `KafkaBytesSource`, and asserts `read()` yields the same keys/payloads. Implementation notes: - `KafkaBytesSource` extends `PushSource`, whose `read()` blocks on an internal queue and never returns once drained. Every `read()` runs on a bounded worker thread with a per-record deadline (`readOne`, following the `readOne()` pattern in `DebeziumMysqlSourceTest`) so a broken test fails fast instead of hanging. Both tests are bounded with `@Test(timeOut = 300_000)`. - All async waiting uses Awaitility; no `Thread.sleep`. - The sink read-back consumer loop exposes `record.headers()`, so the Pulsar-property to Kafka-header propagation from #36 can be asserted there by extending the mock record's properties map — noted inline. This PR does not depend on or merge #36. - Build: adds `testImplementation(libs.testcontainers)` and `testImplementation(libs.testcontainers.kafka)` (version supplied by the Testcontainers BOM already on the platform). ### Verifying this change Run locally with Docker available: ``` ./gradlew :kafka:test ``` `./gradlew :kafka:test` — BUILD SUCCESSFUL, both new tests green alongside the existing suite: ``` KafkaConnectorIntegrationTest > testSinkWritesToKafka PASSED KafkaConnectorIntegrationTest > testSourceReadsFromKafka PASSED BUILD SUCCESSFUL ``` **Mutation check** (confirming the tests actually assert against the live broker, not a no-op): - Sink — pointed the verifying `KafkaConsumer` at a different topic than the sink writes to: ``` testSinkWritesToKafka FAILED org.awaitility.core.ConditionTimeoutException: ... did not yet consume all records from the topic expected [3] but found [0] within 1 minutes. ``` - Source — pointed the source at an empty topic: ``` testSourceReadsFromKafka FAILED org.awaitility.core.ConditionTimeoutException: ... did not yet read all records from the source expected [3] but found [0] within 2 minutes. ``` Both mutations were reverted and `./gradlew :kafka:test` re-run to confirm green. -- 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]
