david-streamlio opened a new pull request, #77:
URL: https://github.com/apache/pulsar-connectors/pull/77
Fixes #50
### Motivation
The `dynamodb` module previously had only `DynamoDBSourceConfigTest`, which
covers config parsing. The actual source logic — `DynamoDBSource` reading from
a DynamoDB Stream via the KCL DynamoDB Streams adapter — was entirely untested.
This adds a Testcontainers/LocalStack integration test that exercises the full
path from a stream-enabled DynamoDB table into the source's record queue.
While wiring the test I hit a genuine bug in `DynamoDBSourceConfig`: all
three client builders (`buildDynamoDBStreamsClient`, `buildDynamoDBClient`,
`buildCloudwatchClient`) set **both** an `EndpointConfiguration` and a `Region`
on the AWS SDK v1 builder. The SDK rejects that with `IllegalStateException:
Only one of Region or EndpointConfiguration may be set`. Since `open()`
mandates `awsRegion`, the endpoint-override path — the only way to target
LocalStack (or any non-default endpoint) — was unusable. This bug had never
surfaced because production usage leaves the endpoint empty.
### Modifications
- `DynamoDBSourceConfig`: make endpoint configuration and region mutually
exclusive (`else if`). The `EndpointConfiguration` already carries the signing
region, so the region is applied only when no endpoint is set. The DynamoDB and
CloudWatch builders now gate on their own endpoint fields (`dynamoEndpoint` /
`cloudwatchEndpoint`) rather than `awsEndpoint`, which was a latent
inconsistency (they checked `awsEndpoint` but applied a different field).
- `dynamodb/build.gradle.kts`: add `testImplementation` for
`testcontainers-localstack` and `awaitility`.
- New `DynamoDBSourceIntegrationTest`:
- Starts `LocalStackContainer` (`localstack/localstack:4.0.3`) with
DynamoDB and CloudWatch (the KCL uses DynamoDB for its lease table and
CloudWatch for metrics).
- Creates a table with `StreamSpecification` enabled
(`NEW_AND_OLD_IMAGES`), waits for `ACTIVE`, reads its `LatestStreamArn`.
- Seeds items, opens `DynamoDBSource` with `TRIM_HORIZON` (plus a
background writer for a steady supply), and asserts the source emits records
carrying an event id and `EVENT_NAME` property.
- `read()` blocks forever on an empty queue, so records are read on a
bounded worker thread with a per-record deadline (never inside an Awaitility
block). The test is bounded by `@Test(timeOut = 600_000)`.
### Verifying this change
Run locally against Docker: `./gradlew :dynamodb:test`
```
Gradle suite > Gradle test >
org.apache.pulsar.io.dynamodb.DynamoDBSourceConfigTest >
loadFromMapCredentialFromSecretTest PASSED
Gradle suite > Gradle test >
org.apache.pulsar.io.dynamodb.DynamoDBSourceConfigTest > loadFromMapTest PASSED
Gradle suite > Gradle test >
org.apache.pulsar.io.dynamodb.DynamoDBSourceConfigTest > loadFromYamlFileTest
PASSED
Gradle suite > Gradle test >
org.apache.pulsar.io.dynamodb.DynamoDBSourceConfigTest > missingCredentialsTest
PASSED
Gradle suite > Gradle test >
org.apache.pulsar.io.dynamodb.DynamoDBSourceConfigTest > missingStartTimeTest
PASSED
Gradle suite > Gradle test >
org.apache.pulsar.io.dynamodb.DynamoDBSourceIntegrationTest >
testReadFromDynamoDBStream PASSED
BUILD SUCCESSFUL
```
**Mutation check (proving the test is not vacuous):** with the live writer
disabled and the source pointed at an empty/bogus stream ARN (and the read
deadline temporarily shortened to 90s), the test fails instead of passing or
hanging:
```
Gradle suite > Gradle test >
org.apache.pulsar.io.dynamodb.DynamoDBSourceIntegrationTest >
testReadFromDynamoDBStream FAILED
java.lang.AssertionError: Timed out after 90s waiting for a record from
the DynamoDB stream. The source produced no record; see the KCL/worker logs
above.
BUILD FAILED
```
The mutation was then reverted and `./gradlew :dynamodb:test --rerun-tasks`
was re-run to confirm green. `./gradlew :dynamodb:spotlessCheck` also passes.
--
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]
