Refactor event ingestion into conversion + pipeline queues with MySQL source support

conf/jvm.conf

@@ -3,6 +3,8 @@
 -Dfile.encoding=UTF-8
 -Duser.timezone=UTC
 
+-Djava.util.logging.manager=org.apache.logging.log4j.jul.LogManager
+
 -Xms8g
 -Xmx60g
 

conf/pixels-sink.aws.properties

@@ -1,5 +1,6 @@
 # engine | kafka | storage
 sink.datasource=storage
+sink.storage.mode=stream
 # -1 means no limit, Only implement in retina sink mode yet:
 sink.datasource.rate.limit=200000
 sink.monitor.report.file=/home/ubuntu/pixels-sink/result1k2_feb/rate_8192tile_3.csv
@@ -20,11 +21,11 @@ sink.retina.trans.request.batch.size=100
 sink.trans.mode=batch
 sink.monitor.report.enable=true
 # trino for freshness query
-trino.url=jdbc:trino://realtime-pixels-coordinator:8080/pixels/pixels_bench
-# trino.url=jdbc:trino://realtime-pixels-coordinator:8080/pixels/pixels_bench_sf10x
-trino.user=pixels
-trino.password=password
-trino.parallel=1
+sink.query.url=jdbc:trino://realtime-pixels-coordinator:8080/pixels/pixels_bench
+# sink.query.url=jdbc:trino://realtime-pixels-coordinator:8080/pixels/pixels_bench_sf10x
+sink.query.user=pixels
+sink.query.password=password
+sink.query.parallel=1
 # row or txn or embed
 sink.monitor.freshness.level=embed
 sink.monitor.freshness.verbose=true
@@ -41,8 +42,7 @@ bootstrap.servers=realtime-kafka-2:29092
 group.id=3078
 auto.offset.reset=earliest
 key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
-#value.deserializer=io.pixelsdb.pixels.writer.deserializer.RowChangeEventAvroDeserializer
-value.deserializer=io.pixelsdb.pixels.sink.event.deserializer.RowChangeEventJsonDeserializer
+sink.kafka.value.format=json
 # Topic & Database Config
 topic.prefix=postgresql.oltp_server
 consumer.capture_database=pixels_bench_sf1x
@@ -78,8 +78,6 @@ sink.flink.server.port=9091
 sink.registry.url=http://localhost:8080/apis/registry/v2
 # Transaction Config
 transaction.topic.suffix=transaction
-#transaction.topic.value.deserializer=io.pixelsdb.pixels.writer.deserializer.TransactionAvroMessageDeserializer
-transaction.topic.value.deserializer=io.pixelsdb.pixels.sink.event.deserializer.TransactionJsonMessageDeserializer
 sink.trans.batch.size=100
 
 # Sink Metrics
@@ -90,7 +88,6 @@ sink.monitor.freshness.interval=1000
 
 # Interact with other rpc
 sink.rpc.enable=true
-sink.rpc.mock.delay=20
 # debezium engine config
 debezium.name=testEngine
 debezium.connector.class=io.debezium.connector.postgresql.PostgresConnector

conf/pixels-sink.ch.properties

@@ -1,5 +1,6 @@
 # engine | kafka | storage
 sink.datasource=storage
+sink.storage.mode=stream
 sink.mode=retina
 #sink.datasource=engine
 #sink.mode=proto
@@ -23,10 +24,10 @@ sink.trans.mode=batch
 # sink.trans.mode=record
 sink.monitor.report.enable=true
 # trino for freshness query
-trino.url=jdbc:trino://realtime-pixels-coordinator:8080/pixels/pixels_bench
-trino.user=pixels
-trino.password=password
-trino.parallel=1
+sink.query.url=jdbc:trino://realtime-pixels-coordinator:8080/pixels/pixels_bench
+sink.query.user=pixels
+sink.query.password=password
+sink.query.parallel=1
 # row or txn or embed
 sink.monitor.freshness.level=embed
 sink.monitor.freshness.verbose=true
@@ -42,8 +43,7 @@ bootstrap.servers=realtime-kafka-2:29092
 group.id=3078
 auto.offset.reset=earliest
 key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
-#value.deserializer=io.pixelsdb.pixels.writer.deserializer.RowChangeEventAvroDeserializer
-value.deserializer=io.pixelsdb.pixels.sink.event.deserializer.RowChangeEventJsonDeserializer
+sink.kafka.value.format=json
 # Topic & Database Config
 topic.prefix=postgresql.oltp_server
 consumer.capture_database=pixels_bench_sf1x
@@ -75,8 +75,6 @@ sink.flink.server.port=9091
 sink.registry.url=http://localhost:8080/apis/registry/v2
 # Transaction Config
 transaction.topic.suffix=transaction
-#transaction.topic.value.deserializer=io.pixelsdb.pixels.writer.deserializer.TransactionAvroMessageDeserializer
-transaction.topic.value.deserializer=io.pixelsdb.pixels.sink.event.deserializer.TransactionJsonMessageDeserializer
 sink.trans.batch.size=100
 
 # Sink Metrics
@@ -87,7 +85,6 @@ sink.monitor.freshness.interval=1000
 
 # Interact with other rpc
 sink.rpc.enable=true
-sink.rpc.mock.delay=20
 # debezium engine config
 debezium.name=testEngine
 debezium.connector.class=io.debezium.connector.postgresql.PostgresConnector

conf/pixels-sink.flink.properties.chbench

@@ -1,5 +1,6 @@
 # engine | kafka | storage
 sink.datasource=storage
+sink.storage.mode=stream
 # -1 means no limit, Only implement in retina sink mode yet
 sink.datasource.rate.limit=2000
 # Sink Config: retina | csv | proto | flink | none
@@ -11,10 +12,10 @@ sink.monitor.report.enable=true
 sink.monitor.report.file=/home/ubuntu/pixels-sink/result_delta/rate_test.csv
 sink.monitor.freshness.file=/home/ubuntu/pixels-sink/result_delta/fresh_batch10k.csv
 # trino for freshness query
-trino.url=jdbc:trino://realtime-pixels-coordinator:8080/delta_lake/chbenchmark_w10000
-trino.user=pixels
-trino.password=password
-trino.parallel=1
+sink.query.url=jdbc:trino://realtime-pixels-coordinator:8080/delta_lake/chbenchmark_w10000
+sink.query.user=pixels
+sink.query.password=password
+sink.query.parallel=1
 # row or txn or embed
 sink.monitor.freshness.level=embed
 sink.monitor.freshness.embed.warmup=10

conf/pixels-sink.flink.properties.hybench

@@ -1,5 +1,6 @@
 # engine | kafka | storage
 sink.datasource=storage
+sink.storage.mode=stream
 # -1 means no limit, Only implement in retina sink mode yet
 sink.datasource.rate.limit=500
 sink.datasource.rate.limit.type=guava
@@ -12,10 +13,10 @@ sink.monitor.report.enable=true
 sink.monitor.report.file=/home/ubuntu/pixels-sink/result_delta/rate_hybench_rate500.csv
 sink.monitor.freshness.file=/home/ubuntu/pixels-sink/result_delta/fresh_hybench_rate500.csv
 # trino for freshness query
-trino.url=jdbc:trino://realtime-pixels-coordinator:8080/delta_lake/hybench_sf1000
-trino.user=pixels
-trino.password=password
-trino.parallel=1
+sink.query.url=jdbc:trino://realtime-pixels-coordinator:8080/delta_lake/hybench_sf1000
+sink.query.user=pixels
+sink.query.password=password
+sink.query.parallel=1
 # row or txn or embed
 sink.monitor.freshness.level=embed
 sink.monitor.freshness.embed.warmup=10

conf/pixels-sink.hudi.properties

@@ -1,5 +1,6 @@
 # engine | kafka | storage
 sink.datasource=storage
+sink.storage.mode=stream
 # -1 means no limit, Only implement in retina sink mode yet
 sink.datasource.rate.limit=2000
 # Sink Config: retina | csv | proto | flink | none
@@ -10,11 +11,11 @@ sink.trans.mode=batch
 sink.monitor.report.enable=true
 sink.monitor.report.file=/home/ubuntu/pixels-sink/result_delta/rate_test.csv
 sink.monitor.freshness.file=/home/ubuntu/pixels-sink/result_delta/fresh_batch10k.csv
-# trino for freshness query
-trino.url=jdbc:hive2://172.31.16.214:10000/hudi_hybench_sf1333_2
-trino.user=pixels
-trino.password=
-trino.parallel=1
+# HiveServer2 for Hudi RO incremental freshness (requires: mvn -Phudi-hive package)
+sink.query.url=jdbc:hive2://172.31.16.214:10000/hudi_hybench_sf1333_2
+sink.query.user=pixels
+sink.query.password=
+sink.query.parallel=1
 # row or txn or embed
 sink.monitor.freshness.level=embed
 sink.monitor.freshness.embed.warmup=0

conf/pixels-sink.mysql.properties

@@ -0,0 +1,100 @@
+# engine | kafka | storage
+sink.datasource=engine
+sink.datasource.engine.format=connect
+sink.storage.mode=stream
+# -1 means no limit, Only implement in retina sink mode yet
+sink.datasource.rate.limit=100000
+# Sink Config: retina | csv | proto | flink | none
+sink.mode=retina
+sink.retina.client=8
+sink.retina.log.queue=false
+## batch or single or record, batch is recommend. record is faster, but doesn't have ACID feature
+sink.trans.mode=batch
+sink.monitor.report.enable=true
+sink.monitor.report.file=/home/ubuntu/pixels-sink/resulti7i/100k_rate_mysql.csv
+sink.monitor.freshness.file=/home/ubuntu/pixels-sink/resulti7i/100k_freshness_mysql.csv
+# trino for freshness query
+sink.query.url=jdbc:trino://realtime-kafka-2:8080/pixels/pixels_bench_sf10x
+sink.query.user=pixels
+sink.query.password=password
+sink.query.parallel=8
+# row or txn or embed
+sink.monitor.freshness.level=embed
+sink.monitor.freshness.embed.warmup=10
+sink.monitor.freshness.embed.static=false
+sink.monitor.freshness.embed.snapshot=true
+sink.monitor.freshness.embed.tablelist=binlog_test
+sink.monitor.freshness.verbose=true
+sink.monitor.freshness.timestamp=true
+sink.storage.loop=true
+# Kafka Config
+bootstrap.servers=realtime-kafka-2:29092
+group.id=3078
+auto.offset.reset=earliest
+key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
+sink.kafka.value.format=json
+# Topic & Database Config
+topic.prefix=mysql-cdc
+consumer.capture_database=cdc_verify
+consumer.include_tables=
+sink.csv.path=./data
+sink.csv.enable_header=false
+## Retina Config
+sink.retina.embedded=false
+# stub or stream
+sink.retina.mode=stream
+sink.remote.host=localhost
+sink.remote.port=29422
+sink.timeout.ms=5000
+sink.flush.interval.ms=50
+sink.flush.batch.size=10
+sink.max.retries=3
+## writer commit
+# sync or async
+sink.commit.method=sync
+sink.commit.batch.size=10
+sink.commit.batch.worker=32
+sink.commit.batch.delay=3000
+## Proto Config
+sink.proto.dir=file:///home/ubuntu/disk1/hybench/
+sink.proto.data=hybench10_10
+sink.proto.maxRecords=100000
+## Flink Config
+sink.flink.server.port=9091
+## Schema Registry
+sink.registry.url=http://localhost:8080/apis/registry/v2
+# Transaction Config
+transaction.topic.suffix=transaction
+sink.trans.batch.size=100
+
+# Sink Metrics
+sink.monitor.enable=true
+sink.monitor.port=9464
+sink.monitor.report.interval=10000
+sink.monitor.freshness.interval=1000
+
+# Interact with other rpc
+sink.rpc.enable=true
+# debezium engine config
+debezium.name=mysql-cdc-engine
+debezium.connector.class=io.debezium.connector.mysql.MySqlConnector
+debezium.provide.transaction.metadata=true
+debezium.offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore
+debezium.offset.storage.file.filename=/tmp/pixels-sink-mysql-offsets.dat
+debezium.offset.flush.interval.ms=60000
+debezium.schema.history.internal=io.debezium.storage.file.history.FileSchemaHistory
+debezium.schema.history.internal.file.filename=/tmp/pixels-sink-mysql-schema-history.dat
+debezium.database.hostname=tdsql
+debezium.database.port=6060
+debezium.database.user=tdsql
+debezium.database.password=tdsql123
+debezium.database.server.id=5401
+debezium.database.include.list=cdc_verify
+debezium.snapshot.mode=never
+debezium.key.converter=org.apache.kafka.connect.json.JsonConverter
+debezium.value.converter=org.apache.kafka.connect.json.JsonConverter
+debezium.topic.prefix=mysql-cdc
+debezium.transforms=topicRouting
+debezium.transforms.topicRouting.type=org.apache.kafka.connect.transforms.RegexRouter
+debezium.transforms.topicRouting.regex=mysql-cdc\\.cdc_verify\\.(.*)
+debezium.transforms.topicRouting.replacement=mysql-cdc.cdc_verify.$1

conf/pixels-sink.pg.properties

@@ -1,5 +1,7 @@
 # engine | kafka | storage
 sink.datasource=engine
+sink.datasource.engine.format=connect
+sink.storage.mode=stream
 # -1 means no limit, Only implement in retina sink mode yet
 sink.datasource.rate.limit=100000
 # Sink Config: retina | csv | proto | flink | none
@@ -12,11 +14,11 @@ sink.monitor.report.enable=true
 sink.monitor.report.file=/home/ubuntu/pixels-sink/resulti7i/100k_rate_2.csv
 sink.monitor.freshness.file=/home/ubuntu/pixels-sink/resulti7i/100k_freshness_2.csv
 # trino for freshness query
-trino.url=jdbc:trino://realtime-kafka-2:8080/pixels/pixels_bench_sf10x
-# trino.url=jdbc:trino://realtime-pixels-coordinator:8080/pixels/pixels_bench_sf10x
-trino.user=pixels
-trino.password=password
-trino.parallel=8
+sink.query.url=jdbc:trino://realtime-kafka-2:8080/pixels/pixels_bench_sf10x
+# sink.query.url=jdbc:trino://realtime-pixels-coordinator:8080/pixels/pixels_bench_sf10x
+sink.query.user=pixels
+sink.query.password=password
+sink.query.parallel=8
 # row or txn or embed
 sink.monitor.freshness.level=embed
 sink.monitor.freshness.embed.warmup=10
@@ -31,8 +33,7 @@ bootstrap.servers=realtime-kafka-2:29092
 group.id=3078
 auto.offset.reset=earliest
 key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
-#value.deserializer=io.pixelsdb.pixels.writer.deserializer.RowChangeEventAvroDeserializer
-value.deserializer=io.pixelsdb.pixels.sink.event.deserializer.RowChangeEventJsonDeserializer
+sink.kafka.value.format=json
 # Topic & Database Config
 topic.prefix=postgresql.oltp_server
 consumer.capture_database=pixels_bench_sf1x
@@ -66,8 +67,6 @@ sink.flink.server.port=9091
 sink.registry.url=http://localhost:8080/apis/registry/v2
 # Transaction Config
 transaction.topic.suffix=transaction
-#transaction.topic.value.deserializer=io.pixelsdb.pixels.writer.deserializer.TransactionAvroMessageDeserializer
-transaction.topic.value.deserializer=io.pixelsdb.pixels.sink.event.deserializer.TransactionJsonMessageDeserializer
 sink.trans.batch.size=100
 
 # Sink Metrics
@@ -78,7 +77,6 @@ sink.monitor.freshness.interval=1000
 
 # Interact with other rpc
 sink.rpc.enable=true
-sink.rpc.mock.delay=20
 # debezium engine config
 debezium.name=testEngine
 debezium.connector.class=io.debezium.connector.postgresql.PostgresConnector