foo.diff

diff --git a/run.sh b/run.sh
index 90499ae..12e8f05 100755
--- a/run.sh
+++ b/run.sh
@@ -100,25 +100,26 @@ done
 # Make db tables
 info "Making db tables..."
 for d in $all_mysqlds; do
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; CREATE DATABASE etsy_shard_001"
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; USE etsy_shard_001"
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; CREATE TABLE etsy_shard_001.test (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; CREATE TABLE etsy_shard_001.test2 (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; CREATE DATABASE etsy_shard_001"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; USE etsy_shard_001"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; CREATE TABLE etsy_shard_001.test (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; CREATE TABLE etsy_shard_001.test2 (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
 
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; CREATE DATABASE etsy_aux"
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; USE etsy_aux"
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; CREATE TABLE etsy_aux.test (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; CREATE TABLE etsy_aux.test2 (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; CREATE DATABASE etsy_aux"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; USE etsy_aux"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; CREATE TABLE etsy_aux.test (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; CREATE TABLE etsy_aux.test2 (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
 
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; CREATE DATABASE etsy_shard_888"
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; USE etsy_shard_888"
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; CREATE TABLE etsy_shard_888.test (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; CREATE TABLE etsy_shard_888.test2 (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; CREATE DATABASE etsy_shard_888"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; USE etsy_shard_888"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; CREATE TABLE etsy_shard_888.test (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; CREATE TABLE etsy_shard_888.test2 (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
 
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; CREATE DATABASE etsy_index"
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; USE etsy_index"
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; CREATE TABLE etsy_index.test (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
-    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=0; CREATE TABLE etsy_index.test2 (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; CREATE DATABASE etsy_index"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; USE etsy_index"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; CREATE TABLE etsy_index.test (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
+    ./workspace/mysql/bin/mysql -uroot -proot --protocol=tcp -P${listen_port[$d]} -e "set sql_log_bin=1; CREATE TABLE etsy_index.test2 (k int primary key, v1 tinyint default '1', v2 int null, v3 int null default '3', v4 int not null default '4', v5 int not null)"
+    break
 done
 
 # Get the mysql server UUIDs
@@ -223,9 +224,20 @@ curl -f -m10 -v -i -H "Accept:application/json" -H "Content-Type:application/jso
             "database.history.kafka.topic": "test", 
             "database.whitelist": "etsy_.+",
             "database.mysql.expect_metadata_events": true, 
-            "snapshot.mode": "twitter_patch", 
+            "snapshot.mode": "never", 
             "topic.mapper": "io.debezium.relational.topic.ByLogicalTableTopicMapper", 
-            "gtid.source.includes": "'"${mysqld_uuid[a]}"'" 
+            "gtid.source.includes": "'"${mysqld_uuid[a]}"'",
+            "transforms": "sharded_router,non_sharded_router", 
+            "transforms.sharded_router.type": "io.debezium.transforms.ByLogicalTableRouter", 
+            "transforms.sharded_router.topic.regex": ".*\\.(?<logicalDb>etsy_[^.]+)_\\d+\\.(?<table>[^.]+)",
+            "transforms.sharded_router.topic.replacement": "${logicalDb}.${table}",
+            "transforms.sharded_router.key.field.regex": ".*\\.(?<db>etsy_[^.]+)\\.(?<table>[^.]+)",
+            "transforms.sharded_router.key.field.replacement": "${db}",
+            "transforms.non_sharded_router.type": "io.debezium.transforms.ByLogicalTableRouter",
+            "transforms.non_sharded_router.topic.regex": ".*\\.(?<db>etsy_[^.]+)\\.(?<table>[^.]+)",
+            "transforms.non_sharded_router.topic.replacement": "${db}.${table}",
+            "transforms.non_sharded_router.key.field.regex": ".*\\.(?<db>etsy_[^.]+)\\.(?<table>[^.]+)",
+            "transforms.non_sharded_router.key.field.replacement": "${db}"
         } 
     }'
 [ $? -eq 0 ] || die "Debezium init failed"