Added config with keepermap storage and removed FINAL in CREATE VIEW …

sink-connector-lightweight/clickhouse/config.xml

@@ -12,4 +12,5 @@
         <replica>clickhouse</replica>
         <shard>02</shard>
     </macros>
+    <keeper_map_path_prefix>/keeper_map_tables</keeper_map_path_prefix>
 </clickhouse>

sink-connector-lightweight/docker/config_keepermap_storage.yml

@@ -0,0 +1,52 @@
+name: "company-1"
+database.hostname: "mysql-master"
+database.port: "3306"
+database.user: "root"
+database.password: "root"
+database.server.id: "12345"
+database.server.name: "ER54"
+database.include.list: test
+table.include.list: ""
+clickhouse.server.url: "clickhouse"
+clickhouse.server.user: "root"
+clickhouse.server.password: "root"
+clickhouse.server.port: "8123"
+database.allowPublicKeyRetrieval: "true"
+snapshot.mode: "initial"
+offset.flush.interval.ms: 5000
+connector.class: "io.debezium.connector.mysql.MySqlConnector"
+offset.storage: "io.debezium.storage.jdbc.offset.JdbcOffsetBackingStore"
+offset.storage.jdbc.offset.table.name: "altinity_sink_connector.replica_source_info"
+offset.storage.jdbc.url: "jdbc:clickhouse://clickhouse:8123/altinity_sink_connector"
+offset.storage.jdbc.user: "root"
+offset.storage.jdbc.password: "root"
+offset.storage.jdbc.offset.table.ddl:  "CREATE TABLE if not exists %s on cluster '{cluster}'
+(
+    id String,
+    offset_key String,
+    offset_val String,
+    record_insert_ts DateTime,
+    record_insert_seq UInt64,
+) ENGINE =  KeeperMap('/asc_offsets201',10)
+PRIMARY KEY offset_key"
+offset.storage.jdbc.offset.table.delete: "select 1"
+schema.history.internal: "io.debezium.storage.jdbc.history.JdbcSchemaHistory"
+schema.history.internal.jdbc.url: "jdbc:clickhouse://clickhouse:8123/altinity_sink_connector"
+schema.history.internal.jdbc.user: "root"
+schema.history.internal.jdbc.password: "root"
+schema.history.internal.jdbc.schema.history.table.ddl: "CREATE TABLE if not exists %s on cluster '{cluster}'
+(
+   id FixedString(36),
+   history_data String,
+   history_data_seq UInt32,
+   record_insert_ts DateTime,
+   record_insert_seq UInt32
+) ENGINE=ReplicatedReplacingMergeTree(record_insert_seq)
+order by id"
+schema.history.internal.jdbc.schema.history.table.name: "altinity_sink_connector.replicate_schema_history"
+enable.snapshot.ddl: "true"
+persist.raw.bytes: "false"
+auto.create.tables: "true"
+database.connectionTimeZone: "UTC"
+restart.event.loop: "true"
+restart.event.loop.timeout.period.secs: "3000"

sink-connector-lightweight/src/main/resources/config.properties

@@ -21,4 +21,4 @@ auto.create.tables= false
 replacingmergetree.delete.column=_sign
 metrics.port= 8083
 snapshot.mode= "initial"
-replica.status.view="CREATE VIEW IF NOT EXISTS %s.show_replica_status AS SELECT now() - fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')) AS seconds_behind_source,  toDateTime(fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')), 'UTC') AS utc_time, fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')) AS local_time FROM %s FINAL"
+replica.status.view="CREATE VIEW IF NOT EXISTS %s.show_replica_status AS SELECT now() - fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')) AS seconds_behind_source,  toDateTime(fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')), 'UTC') AS utc_time, fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')) AS local_time FROM %s settings final=1"

sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ITCommon.java

@@ -90,7 +90,7 @@ static public Properties getDebeziumPropertiesForSchemaOnly(MySQLContainer mySql
         props.replace("snapshot.mode", "schema_only");
         props.replace("disable.drop.truncate", "true");
         props.setProperty("disable.ddl", "true");
-
+        props.setProperty("replica.status.view", "CREATE VIEW IF NOT EXISTS %s.show_replica_status AS SELECT now() - fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')) AS seconds_behind_source,  toDateTime(fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')), 'UTC') AS utc_time, fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')) AS local_time FROM %s settings final=1");
         return props;
     }
 }