== Camel Debezium with Azure Event Hubs and Azure Storage Blob example

=== Introduction

An example which shows how to integrate Camel with Debezium, Azure Event Hubs and sink everything into Azure Storage Blob.

This project consists of the following examples:

  1. Send events using Debezium component to Azure Event Hubs.
  2. Example how data can be sinked into Azure Storage Blob.

=== Prerequisites

==== MySQL
In order to stream changes from MySQL, you will need to have https://debezium.io/documentation/reference/0.9/connectors/mysql.html#enabling-the-binlog[_row-level_] binary binlog enabled. However,
for the sake of this example, we will use the following docker image which is setup with row enabled binary logs and some sample data:

[source,sh]
----
$ docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw debezium/example-mysql:0.9
----
The above docker image will start a MySQL server exposed to port `3306` with root password set.

==== Azure Event Hubs
Since we will use Azure Event Hubs to stream changes from Debezium as an example, you will need to create an event hub with any name you like. As well, you will need to generate the connection string for your newly created event hub, please refer to this https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-get-connection-string[documentation] to learn more on how to generate the connection string, update the following properties in `src/main/resources/application.properties`:
```
eventhubs.connectionString = {{generated_connection_string}}
```

*Note:* Normally, the connection string looks like this:
```
Endpoint=sb://{{eventhub_namespace}}.servicebus.windows.net/;SharedAccessKeyName={{shared_access_key_name}};SharedAccessKey={{shared_access_key_value}}=;EntityPath={{eventhub_name}}
```

*Note:* To consume the data from Azure Event Hub, we will need to use CheckpointStore. However, for the sake of this example, we will use the default BlobCheckpointStore, therefore we will need to supply the access details for our blob storage account.

==== Azure Storage Blob
We will use the storage blob for two purposes in this example:

1. As CheckpointStore to store the offsets as we mentioned in the earlier section.
2. As sink data store for our data from Azure Event Hubs.

First, we will need to create the following:

1. Azure Storage Account.
2. A storage container under the newly created storage account.
3. Generate the access key for the storage account in the Azure storage online portal or Azure command line.

Once you are done from this, update the following properties in `src/main/resources/application.properties`:
```
blob.containerName = {{created_container_name}}
blob.accountName = {{storage_account_name}}
blob.accessKey = {{generated_access_key}}
blob.blobName = {{blob_name}} //You can create it beforehand or the component can create it if does not exist
```

=== Build

Due to licensing issues, you will need to add the dependency for `mysql-conenctor-java`, just add the following to your POM file:

[source,xml]
------------------------------------------------------------
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.15</version>
</dependency>
------------------------------------------------------------

You will need to compile this example first:

[source,sh]
----
$ mvn clean compile
----

=== Run

Run the Azure Event Hubs to Azure Storage Blob producer first

[source,sh]
----
$ mvn compile exec:java -Peventhubs-producer
----

Run the Debezium consumer in the seperate shell

[source,sh]
----
$ mvn compile exec:java -Pdebezium-consumer
----

Initially, you will Debezium will perform a snapshot of the whitelisted tables per `application.properties`, hence you should expect
the data to be replicated into Cassandra. Once the snapshot mode is done, you can try to insert a new row, update fields, delete .. etc on  MySQL whitelisted table(s), you should see
the changes reflecting on Cassandra as well, you can verify that by running the following query on cqlsh:
```
select * from dbzSink.products;
```

=== Configuration

You can configure the details in the file:
  `src/main/resources/application.properties`

You can enable verbose logging by adjusting the `src/main/resources/log4j2.properties`
  file as documented in the file.

=== Help and contributions

If you hit any problem using Camel or have some feedback, 
then please https://camel.apache.org/community/support/[let us know].

We also love contributors, 
so https://camel.apache.org/community/contributing/[get involved] :-)

The Camel riders!