# Dynamodb Stream Source Connector

## Description

The `DynamoDbSourceConnector` ingests events from a [DynmamoDB Stream](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html)
and publishes the events to a kafka_topic.
Ref: https://camel.apache.org/components/4.4.x/aws2-ddbstream-component.html

## Build steps

To create a local kafka, connect network.

```bash
cp .env.shadow .env
# update .env with desired values

awsume dev
aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin 086679231553.dkr.ecr.us-east-1.amazonaws.com
docker-compose up --build -d
```

To create a kafka topic
```bash
make create_topic
```

To list kafka topics
```bash
make list_topics
```

### References

- [AWS DYNAMODB STREAMS](https://camel.apache.org/components/4.4.x/aws2-ddbstream-component.html)
- [Connector Plugin doc](https://camel.apache.org/components/4.4.x/aws2-ddbstream-component.html)
- [Connector JAR files](https://repo1.maven.org/maven2/org/apache/camel/kafkaconnector/camel-aws-ddb-streams-source-kafka-connector/4.4.3/)
- [AWS STS JAR](https://repo1.maven.org/maven2/software/amazon/awssdk/sts/2.20.76/)
- [AWS https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials-chain.html] (https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials-chain.html)

Example config

```json
{
  "connector.class": "org.apache.camel.kafkaconnector.awsddbstreamssource.CamelAwsddbstreamssourceSourceConnector",
  "camel.kamelet.aws-ddb-streams-source.region": "us-east-1",
  "camel.kamelet.aws-ddb-streams-source.streamIteratorType": "FROM_START",
  "camel.kamelet.aws-ddb-streams-source.table": "qa_pp_identity",
  "camel.kamelet.aws-ddb-streams-source.useDefaultCredentialsProvider": "true",
  "kafka.bootstrap.servers": "b-3.dev-managed-kafka-cdc.rk4es0.c11.kafka.us-east-1.amazonaws.com:9094,b-2.dev-managed-kafka-cdc.rk4es0.c11.kafka.us-east-1.amazonaws.com:9094,b-1.dev-managed-kafka-cdc.rk4es0.c11.kafka.us-east-1.amazonaws.com:9094",
  "topics": "dynamodb-qa_pp_identity",
  "transforms": "CamelTypeConverterTransformer",
  "transforms.CamelTypeConverterTransformer.type": "org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value",
  "transforms.CamelTypeConverterTransformer.target.type": "java.lang.String"
}
```