chmod a+x manage.sh
sh manage.sh up

http POST localhost:8083/connectors @connect/mysql-source.json
./bin/kafka-avro-console-consumer --topic mysql.demo.customers --from-beginning

http POST localhost:8083/connectors @connect/postgres-sink.json

docker exec -ti mongo mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/demo

PRIMARY> show collections;
PRIMARY> db.user_actions.insert({"userId": 2, "ts": new Date(), "ip": "192.168.0.8"}),
         db.user_actions.insert({"userId": 3, "ts": new Date(), "ip": "192.168.233.255"}),
         db.user_actions.insert({"userId": 4, "ts": new Date(), "ip": "192.168.0.0.1"}),
         db.user_actions.insert({"userId": 5, "ts": new Date(), "ip": "10.0.0.15"}),
         db.user_actions.insert({"userId": 6, "ts": new Date(), "ip": "10.20.33.4"}),
         db.user_actions.insert({"userId": 7, "ts": new Date(), "ip": "192.255.30.4"}),
         db.user_actions.insert({"userId": 8, "ts": new Date(), "ip": "10.0.0.182"}),
         db.user_actions.insert({"userId": 9, "ts": new Date(), "ip": "10.0.0.10"}),
         db.user_actions.insert({"userId": 10, "ts": new Date(), "ip": "168.0.0.1"}),
         db.user_actions.insert({"userId": 11, "ts": new Date(), "ip": "10.10.255.2"}),
         db.user_actions.insert({"userId": 12, "ts": new Date(), "ip": "168.255.3.4"})

http POST localhost:8083/connectors @connect/mongo-source.json 
http POST localhost:8083/connectors @connect/postgres-mongo-sink.json

http POST localhost:8083/connectors @connect/postgres-mongo-sink-no-pii.json

http POST localhost:8083/connectors @connect/elastic-sink.json

curl http://localhost:9200/mysql.demo.customers/_search?pretty=true&q=*:*

docker exec -ti ksql ksql

set 'auto.offset.reset'='earliest';
create stream customers with(kafka_topic='mysql.demo.customers', value_format='AVRO');
create stream addressLine1_changed_notification with (value_format='JSON') as 
    select before->addressLine1 rcpt, concat('Your address was changed to ', after->addressLine1) message
from customers where before->addressLine1 <> after->addressLine1;

./bin/kafka-console-consumer --topic ADDRESSLINE1_CHANGED_NOTIFICATION --from-beginning