DBZ-313:Demo: Keep *all* change events in a downstream database [DBZ-2101] + Modyfying README.md and connector configuration for .add.fields.prefix

Author: SaidiBTW

auditlog/README.md

@@ -5,37 +5,37 @@ It accompanies the blog post [Building Audit Logs with Change Data Capture and S
 
 There are two applications (based on [Quarkus](https://quarkus.io/)):
 
-* _vegetables-service_: a simple REST service for inserting and updating vegetable data into a Postgres database;
-as part of its processing, it will not only update its actual "business table" `vegetable`,
-but also insert some auditing metadata into a dedicated metadata table `transaction_context_data`:
-the user (as obtained from the passed JWT token), the client's date (as passed via the HTTP 1.1 `Date` header)
-and a use case identifier (as specified in an annotation on the REST API methods).
-* _log-enricher_: a Kafka Streams application,
-which joins the CDC topic holding the `vegetable` change events (`dbserver1.inventory.vegetable`) with the corresponding metadata in the `dbserver1.inventory.transaction_context_data` topic sourced from the `transaction_context_data` table;
-this table is keyed by transaction id, allowing for joining the vegetable `KStream` with the metadata `KTable`.
-The enriched vegetable change events are written to the `dbserver1.inventory.vegetable.enriched` topic.
+- _vegetables-service_: a simple REST service for inserting and updating vegetable data into a Postgres database;
+  as part of its processing, it will not only update its actual "business table" `vegetable`,
+  but also insert some auditing metadata into a dedicated metadata table `transaction_context_data`:
+  the user (as obtained from the passed JWT token), the client's date (as passed via the HTTP 1.1 `Date` header)
+  and a use case identifier (as specified in an annotation on the REST API methods).
+- _log-enricher_: a Kafka Streams application,
+  which joins the CDC topic holding the `vegetable` change events (`dbserver1.inventory.vegetable`) with the corresponding metadata in the `dbserver1.inventory.transaction_context_data` topic sourced from the `transaction_context_data` table;
+  this table is keyed by transaction id, allowing for joining the vegetable `KStream` with the metadata `KTable`.
+  The enriched vegetable change events are written to the `dbserver1.inventory.vegetable.enriched` topic.
 
 ## Building the Demo
 
 ```console
-$ mvn clean package
+mvn clean package
 ```
 
 ```console
-$ export DEBEZIUM_VERSION=1.8
-$ docker-compose up --build
+export DEBEZIUM_VERSION=1.8
+docker-compose up --build
 ```
 
 ## Deploy the Debezium Postgres Connector
 
 ```console
-$ http PUT http://localhost:8083/connectors/inventory-connector/config < register-postgres.json
+http PUT http://localhost:8083/connectors/inventory-connector/config < register-postgres.json
 ```
 
 ## Modifying Some Data and Observing the Audit Log
 
 ```console
-$ http POST http://localhost:8080/vegetables 'Authorization:Bearer eyJraWQiOiJqd3Qua2V5IiwidHlwIjoiSldUIiwiYWxnIjoiUlMyNTYifQ.eyJzdWIiOiJmYXJtZXJib2IiLCJ1cG4iOiJmYXJtZXJib2IiLCJhdXRoX3RpbWUiOjE1NjY0NTgxMTMsImlzcyI6ImZhcm1zaG9wIiwiZ3JvdXBzIjpbImZhcm1lcnMiLCJjdXN0b21lcnMiXSwiZXhwIjo0MTAyNDQ0Nzk5LCJpYXQiOjE1NjY0NTgxMTMsImp0aSI6IjQyIn0.CscbJN8amqKryYvnVO1184J8F67HN2iTEjVN2VOPodcnoeOd7_iQVKUjC3h-ye5apkJjvAsQKrjzlrGCHRfl-n6jC9F7IkOtjoWnJ4wQ9BBo1SAtPw_Czt1I_Ujm-Kb1p5-BWACCBCVVFgYZTWP_laz5JZS7dIvs6VqoNnw7A4VpA6iPfTVfYlNY3u86-k1FvEg_hW-N9Y9RuihMsPuTdpHK5xdjCrJiD0VJ7-0eRQ8RXpycHuHN4xfmV8MqXBYjYSYDOhbnYbdQVbf0YJoFFqfb75my5olN-97ITsi2MS62W_y-RNT0qZrbytqINA3fF3VQsSY6VcaqRAeygrKm_Q' 'Date:Thu, 22 Aug 2019 08:12:31 GMT' name=Tomatoe description=Yummy!
+http POST http://localhost:8080/vegetables 'Authorization:Bearer eyJraWQiOiJqd3Qua2V5IiwidHlwIjoiSldUIiwiYWxnIjoiUlMyNTYifQ.eyJzdWIiOiJmYXJtZXJib2IiLCJ1cG4iOiJmYXJtZXJib2IiLCJhdXRoX3RpbWUiOjE1NjY0NTgxMTMsImlzcyI6ImZhcm1zaG9wIiwiZ3JvdXBzIjpbImZhcm1lcnMiLCJjdXN0b21lcnMiXSwiZXhwIjo0MTAyNDQ0Nzk5LCJpYXQiOjE1NjY0NTgxMTMsImp0aSI6IjQyIn0.CscbJN8amqKryYvnVO1184J8F67HN2iTEjVN2VOPodcnoeOd7_iQVKUjC3h-ye5apkJjvAsQKrjzlrGCHRfl-n6jC9F7IkOtjoWnJ4wQ9BBo1SAtPw_Czt1I_Ujm-Kb1p5-BWACCBCVVFgYZTWP_laz5JZS7dIvs6VqoNnw7A4VpA6iPfTVfYlNY3u86-k1FvEg_hW-N9Y9RuihMsPuTdpHK5xdjCrJiD0VJ7-0eRQ8RXpycHuHN4xfmV8MqXBYjYSYDOhbnYbdQVbf0YJoFFqfb75my5olN-97ITsi2MS62W_y-RNT0qZrbytqINA3fF3VQsSY6VcaqRAeygrKm_Q' 'Date:Thu, 22 Aug 2019 08:12:31 GMT' name=Tomatoe description=Yummy!
 ```
 
 This uses a pre-generated JWT token (with expiration date set to 2099-12-31 and user set to "farmerbob").
@@ -44,13 +44,13 @@ To regenerate the token with different data, use the [Jwtenizr](https://github.c
 You can also update an existing vegetable record (this token is for "farmermargaret"):
 
 ```console
-$ http PUT http://localhost:8080/vegetables/10 'Authorization:Bearer eyJraWQiOiJqd3Qua2V5IiwidHlwIjoiSldUIiwiYWxnIjoiUlMyNTYifQ.eyJzdWIiOiJmYXJtZXJtYXJnYXJldCIsInVwbiI6ImZhcm1lcm1hcmdhcmV0IiwiYXV0aF90aW1lIjoxNTY5ODM1Mzk5LCJpc3MiOiJmYXJtc2hvcCIsImdyb3VwcyI6WyJmYXJtZXJzIiwiY3VzdG9tZXJzIl0sImV4cCI6NDEwMjQ0NDc5OSwiaWF0IjoxNTY5ODM1Mzk5LCJqdGkiOiI0MiJ9.DTEUA3p-xyK5nveoJIVhjfKNFdVszYIb55Qj4Xrm70DDbAXuOU2FMkffuUAUm2s7ACkp2KEmg6brRwSjvA-zhW61kDR9ZgEb9NWeDjr6Eue08xcSODKt7SGV-M7h3yhuDIhU7uaZrxRUAQTWqm1vxd2rmN_QH0frhKMUNFFsLIOGLG0zHcLosRcwZ4tAKXSSB9VE0fth6srIQCUebDkF7ucA_WSYjPRvahCBd8JvnV4VUGQxZW8zcRhTEwcaLq20ODO-dr85xgWI2Yr_1A7PDuDL4oUjCb90YyhtzaIzs2vQMjcxJ6TWmTcqJpgCfkjE-TeVwjaafcNJu0fBmcP8jA' 'Date:Thu, 22 Aug 2019 08:12:31 GMT' name=Tomatoe description=Tasty!
+http PUT http://localhost:8080/vegetables/10 'Authorization:Bearer eyJraWQiOiJqd3Qua2V5IiwidHlwIjoiSldUIiwiYWxnIjoiUlMyNTYifQ.eyJzdWIiOiJmYXJtZXJtYXJnYXJldCIsInVwbiI6ImZhcm1lcm1hcmdhcmV0IiwiYXV0aF90aW1lIjoxNTY5ODM1Mzk5LCJpc3MiOiJmYXJtc2hvcCIsImdyb3VwcyI6WyJmYXJtZXJzIiwiY3VzdG9tZXJzIl0sImV4cCI6NDEwMjQ0NDc5OSwiaWF0IjoxNTY5ODM1Mzk5LCJqdGkiOiI0MiJ9.DTEUA3p-xyK5nveoJIVhjfKNFdVszYIb55Qj4Xrm70DDbAXuOU2FMkffuUAUm2s7ACkp2KEmg6brRwSjvA-zhW61kDR9ZgEb9NWeDjr6Eue08xcSODKt7SGV-M7h3yhuDIhU7uaZrxRUAQTWqm1vxd2rmN_QH0frhKMUNFFsLIOGLG0zHcLosRcwZ4tAKXSSB9VE0fth6srIQCUebDkF7ucA_WSYjPRvahCBd8JvnV4VUGQxZW8zcRhTEwcaLq20ODO-dr85xgWI2Yr_1A7PDuDL4oUjCb90YyhtzaIzs2vQMjcxJ6TWmTcqJpgCfkjE-TeVwjaafcNJu0fBmcP8jA' 'Date:Thu, 22 Aug 2019 08:12:31 GMT' name=Tomatoe description=Tasty!
 ```
 
 Or delete a record (again using the "farmerbob" token):
 
 ```console
-$ http DELETE http://localhost:8080/vegetables/10 'Authorization:Bearer eyJraWQiOiJqd3Qua2V5IiwidHlwIjoiSldUIiwiYWxnIjoiUlMyNTYifQ.eyJzdWIiOiJmYXJtZXJib2IiLCJ1cG4iOiJmYXJtZXJib2IiLCJhdXRoX3RpbWUiOjE1NjY0NTgxMTMsImlzcyI6ImZhcm1zaG9wIiwiZ3JvdXBzIjpbImZhcm1lcnMiLCJjdXN0b21lcnMiXSwiZXhwIjo0MTAyNDQ0Nzk5LCJpYXQiOjE1NjY0NTgxMTMsImp0aSI6IjQyIn0.CscbJN8amqKryYvnVO1184J8F67HN2iTEjVN2VOPodcnoeOd7_iQVKUjC3h-ye5apkJjvAsQKrjzlrGCHRfl-n6jC9F7IkOtjoWnJ4wQ9BBo1SAtPw_Czt1I_Ujm-Kb1p5-BWACCBCVVFgYZTWP_laz5JZS7dIvs6VqoNnw7A4VpA6iPfTVfYlNY3u86-k1FvEg_hW-N9Y9RuihMsPuTdpHK5xdjCrJiD0VJ7-0eRQ8RXpycHuHN4xfmV8MqXBYjYSYDOhbnYbdQVbf0YJoFFqfb75my5olN-97ITsi2MS62W_y-RNT0qZrbytqINA3fF3VQsSY6VcaqRAeygrKm_Q' 'Date:Thu, 22 Aug 2019 08:12:31 GMT'
+http DELETE http://localhost:8080/vegetables/10 'Authorization:Bearer eyJraWQiOiJqd3Qua2V5IiwidHlwIjoiSldUIiwiYWxnIjoiUlMyNTYifQ.eyJzdWIiOiJmYXJtZXJib2IiLCJ1cG4iOiJmYXJtZXJib2IiLCJhdXRoX3RpbWUiOjE1NjY0NTgxMTMsImlzcyI6ImZhcm1zaG9wIiwiZ3JvdXBzIjpbImZhcm1lcnMiLCJjdXN0b21lcnMiXSwiZXhwIjo0MTAyNDQ0Nzk5LCJpYXQiOjE1NjY0NTgxMTMsImp0aSI6IjQyIn0.CscbJN8amqKryYvnVO1184J8F67HN2iTEjVN2VOPodcnoeOd7_iQVKUjC3h-ye5apkJjvAsQKrjzlrGCHRfl-n6jC9F7IkOtjoWnJ4wQ9BBo1SAtPw_Czt1I_Ujm-Kb1p5-BWACCBCVVFgYZTWP_laz5JZS7dIvs6VqoNnw7A4VpA6iPfTVfYlNY3u86-k1FvEg_hW-N9Y9RuihMsPuTdpHK5xdjCrJiD0VJ7-0eRQ8RXpycHuHN4xfmV8MqXBYjYSYDOhbnYbdQVbf0YJoFFqfb75my5olN-97ITsi2MS62W_y-RNT0qZrbytqINA3fF3VQsSY6VcaqRAeygrKm_Q' 'Date:Thu, 22 Aug 2019 08:12:31 GMT'
 ```
 
 Doing so, observe the contents of the `dbserver1.inventory.vegetable`, `dbserver1.inventory.transaction_context_data` and `dbserver1.inventory.vegetable.enriched` topics:
@@ -144,19 +144,102 @@ http POST http://localhost:8085/vegetables/{uuid}/auditData/{tuuid} audit:='{"us
 
 This would then fix the missing event in the transaction context data topic and trigger the enricher to provide a new log entry.
 
-## Stopping All Services
+## Fowarding Events to a Downstream Postgres Database
 
 ```console
-$ docker-compose down
+$ docker run -it --rm \
+    --network auditlog_default \
+    quay.io/debezium/tooling:1.2 \
+    /bin/bash -c "kafkacat -b kafka:9092 \
+    -C -o beginning -q -u -t dbserver1.inventory.vegetable.enriched | jq ."
+```
+
+This will show you the enriched events fowarded to the `dbserver1.inventory.vegetable.enriched` topic.
+We need to pass this events to lightweight `postgres-sink` database defined in `docker-compose.yaml` using the [Debezium JDBC connector](https://debezium.io/documentation/reference/stable/connectors/jdbc.html).
+
+Download the connector plugin `.tar.gz` from the Debzium [plugin archive](https://repo1.maven.org/maven2/io/debezium/debezium-connector-jdbc/) and save it to the `/config/plugins/debezium-connector-jdbc` directory.
+
+Restart your Kafka connect container to pick up the new plugin
+
+```console
+docker compose down -v connect
+docker compose up -d connect
+```
+
+Create your debezium sink connector by running
+
+```console
+http POST http://localhost:8083/connectors < connector-config/config/jdbc-connector-config.json
+```
+
+The connector should read from the `dbserver1.inventory.vegetable.enriched` topic and populate the `dbserver1_inventory_vegetable` table.
+
+Confirm the changes by running the `exec` command into your `postgres-sink` container
+
+```console
+docker compose exec postgres-sink psql -U postgresuser -d postgres
+```
+
+View the rows in your `dbserver1_inventory_vegetable` table:
+
+```sql
+SELECT * FROM dbserver1_inventory_vegetable;
+```
+
+An exmaple of the rows should be like this:
+
+```console
+| __connect_topic                          | __connect_partition | __connect_offset | id | description | name    | __deleted | op | lsn      | ts_ms         | tx_id | client_date      | usecase           | user_name       |
+|------------------------------------------|---------------------|------------------|----|-------------|---------|-----------|----|----------|---------------|-------|------------------|-------------------|-----------------|
+| dbserver1.inventory.vegetable.enriched   | 0                   | 0                | 10 | Yummy!      | Tomatoe | false     | c  | 36689976 | 1773921791527 | 769   | 1566461551000000 | CREATE VEGETABLE  | farmerbob       |
+| dbserver1.inventory.vegetable.enriched   | 0                   | 2                | 10 | Yummy!      | Tomatoe | false     | c  | 36689976 | 1773921791527 | 769   | 1566461551000000 | CREATE VEGETABLE  | farmerbob       |
+| dbserver1.inventory.vegetable.enriched   | 0                   | 4                | 10 | Tasty!      | Tomatoe | false     | u  | 36690432 | 1773921804910 | 770   | 1566461551000000 | UPDATE VEGETABLE  | farmermargaret  |
+| dbserver1.inventory.vegetable.enriched   | 0                   | 5                | 10 |             |         | true      | d  | 36690800 | 1773921809082 | 771   | 1566461551000000 | DELETE VEGETABLE  | farmerbob       |
+```
+
+With this our audit log events have been successfully propagated to our downstream database.
+
+The table `dbserver_inventory_vegetable` can be defined as:
+
+```sql
+-- This version is used with the Debezium JDBC Sink Connector
+CREATE TABLE dbserver1_inventory_vegetable_enriched (
+    -- Business Data
+    id INTEGER,
+    name VARCHAR(255) NULL,
+    description TEXT NULL,
+
+    -- Audit Metadata
+    tx_id VARCHAR(255),
+    user_name VARCHAR(255) NULL,
+    usecase VARCHAR(255) NULL,
+    client_date BIGINT NULL,
+
+    -- Change Metadata (for uniqueness)
+    op VARCHAR(1),
+    lsn BIGINT,
+    ts_ms BIGINT,
+
+    -- Kafka Metadata to prevent squashing of events
+    __connect_partition INTEGER,
+    __connect_offset BIGINT,
+    __connect_topic VARCHAR(255),
+
+    -- To ensure that all events are kept we will use kafka metadata to
+    -- To ensure that all events are kept we will use kafka metadata to 
+    -- identify unique rows
+PRIMARY KEY (__connect_offset, __connect_partition, __connect_topic)
+);
+
 ```
 
-## Running the Quarkus Applications Locally
+## Running the Quarkus applications locally
 
 Set `ADVERTISED_HOST_NAME` of the `kafka` service in _docker-compose.yaml_ to the IP address of your host machine.
 Start all services except the `vegetables-service` and the `log-enricher`:
 
 ```console
-$ docker-compose up --scale vegetables-service=0 --scale log-enricher=0
+docker-compose up --scale vegetables-service=0 --scale log-enricher=0
 ```
 
 Then start the three services via the Quarkus dev mode:

auditlog/connector-config/config/camel-sink-connector-config.json

@@ -15,6 +15,7 @@
     "transforms": "unwrap",
     "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
     "transforms.unwrap.add.fields": "op,lsn,source.ts_ms:ts_ms,txId:tx_id,audit.client_date:client_date,audit.usecase:usecase,audit.user_name:user_name",
+    "transforms.unwrap.add.fields.prefix": "",
     "transforms.unwrap.delete.tombstone.handling.mode": "rewrite"
   }
 }

auditlog/connector-config/config/jdbc-connector-config.json

@@ -26,31 +26,3 @@ CREATE TABLE dbserver1_inventory_vegetable_enriched (
 PRIMARY KEY (__connect_offset, __connect_partition, __connect_topic)
 );
 
-
--- This version is used with the Camel JDBC Sink Connector
-CREATE TABLE enriched_vegetables (
-    -- Business Data
-    id INTEGER,
-    name VARCHAR(255) NULL,
-    description TEXT NULL,
-    
-    -- Audit Metadata
-    tx_id VARCHAR(255),
-    user_name VARCHAR(255) NULL,
-    use_case VARCHAR(255) NULL,
-    client_date BIGINT NULL,
-    
-    -- Change Metadata (for uniqueness)
-    op VARCHAR(1),
-    lsn BIGINT,
-    ts_ms BIGINT,
-
-    -- Kafka Metadata to prevent squashing of events
-    kafka_partion INTEGER,
-    kafka_offset BIGINT,
-    kafka_topic VARCHAR(255),
-    
-    -- To ensure that all events are kept we will use kafka metadata to 
-    -- 
-PRIMARY KEY (kafka_topic, kafka_partition, kafka_offset)
-);