861 renaming a table doesnt work for databases that are mapped to another name in clickhouse

README.md

@@ -1,7 +1,7 @@
 [![License](http://img.shields.io/:license-apache%202.0-brightgreen.svg)](http://www.apache.org/licenses/LICENSE-2.0.html)
 [![Sink Connector(Kafka version) tests](https://github.com/Altinity/clickhouse-sink-connector/actions/workflows/sink-connector-kafka-tests.yml/badge.svg)](https://github.com/Altinity/clickhouse-sink-connector/actions/workflows/sink-connector-kafka-tests.yml)
 [![Sink Connector(Light-weight) Tests](https://github.com/Altinity/clickhouse-sink-connector/actions/workflows/sink-connector-lightweight-tests.yml/badge.svg)](https://github.com/Altinity/clickhouse-sink-connector/actions/workflows/sink-connector-lightweight-tests.yml)
-<a href="https://join.slack.com/t/altinitydbworkspace/shared_invite/zt-w6mpotc1-fTz9oYp0VM719DNye9UvrQ">
+<a href="https://altinity.com/slack">
   <img src="https://img.shields.io/static/v1?logo=slack&logoColor=959DA5&label=Slack&labelColor=333a41&message=join%20conversation&color=3AC358" alt="AltinityDB Slack" />
 </a>
 <img alt="Docker Pulls" src="https://img.shields.io/docker/pulls/altinityinfra/clickhouse-sink-connector">
@@ -16,7 +16,7 @@ for analysis.
 
 ## Features
 
-* Initial data dump and load 
+* [Initial data dump and load(MySQL)](sink-connector/python/README.md) 
 * Change data capture of new transactions using [Debezium](https://debezium.io/)
 * Automatic loading into ClickHouse
 * Sources: Support for MySQL, PostgreSQL (other databases experimental)
@@ -103,6 +103,6 @@ to ClickHouse and analytic applications built on ClickHouse.
 - [Official website](https://altinity.com/) - Get a high level overview of Altinity and our offerings.
 - [Altinity.Cloud](https://altinity.com/cloud-database/) - Run ClickHouse in our cloud or yours.
 - [Altinity Support](https://altinity.com/support/) - Get Enterprise-class support for ClickHouse and Sink Connector.
-- [Slack](https://altinitydbworkspace.slack.com/join/shared_invite/zt-1togw9b4g-N0ZOXQyEyPCBh_7IEHUjdw#/shared-invite/email) - Talk directly with ClickHouse users and Altinity devs.
+- [Slack](https://altinity.com/slack) - Talk directly with ClickHouse users and Altinity devs.
 - [Contact us](https://hubs.la/Q020sH3Z0) - Contact Altinity with your questions or issues.
 - [Free consultation](https://hubs.la/Q020sHkv0) - Get a free consultation with a ClickHouse expert today.

doc/production_setup.md

@@ -2,7 +2,9 @@
 
 
 [Throughput & Memory Usage](#improving-throughput-and/or-memory-usage.) \
+[Low Memory environments(5GB)](#low-memory-environments5gb) \
 [Initial Load](#initial-load) \
+[MySQL Setup](#mysql-production-setup) \
 [PostgreSQL Setup](#postgresql-production-setup) \
 [ClickHouse Setup](#clickhouse-setup)
 
@@ -47,6 +49,16 @@ in terms of number of elements the queue can hold and the maximum size of the qu
     buffer.flush.time.ms: "1000"
 ```
 
+## Low Memory environments(5GB)
+The suggested configuration for a low memory environment is as follows to use a single threaded configuration.
+Single threaded configuration can be enabled in `config.yml`
+```
+single.threaded: "true"
+```
+As shown in the diagram below, the Single threaded configuration will skip the sink connector queue and threadpool
+and will insert batches directly from the debezium queue.
+![](img/single_threaded.jpg)
+
 ## Initial Load
 
 The following parameters might be useful to reduce the memory usage of the connector during the snapshotting phase.
@@ -66,6 +78,37 @@ The maximum number of rows that the connector fetches and reads into memory when
 
 **snapshot.max.threads**: Increase this number from 1 to a higher value to enable parallel snapshotting.
 
+
+## MySQL Production Setup
+# How to Reproduce
+
+1. Replicate a table only in `config.yml`:
+
+    ```yaml
+    table.include.list: "mydb.mytable"
+    ```
+
+2. Do not write to the table on the source database side.
+3. Monitor the lag:
+
+    ```sql
+    select * from altinity_sink_connector.show_replica_status\G
+    ```
+
+4. The lag increases if this table does not get written and the binary log position does not move. It should be synced periodically to show the binary log progress.
+
+# Workaround
+
+Include a heartbeat table (see [Percona Toolkit - pt-heartbeat](https://docs.percona.com/percona-toolkit/pt-heartbeat.html)):
+
+### Example
+
+```sql
+CREATE TABLE pt_heartbeat_db.heartbeat (
+  id int NOT NULL PRIMARY KEY,
+  ts datetime NOT NULL
+);
+=======
 **Single Threaded (Low Memory/Slow replication)**:
 By setting the `single.threaded: true` configuration variable in `config.yml`, the replication will skip the sink connector queue and threadpool
 and will insert batches directly from the debezium queue.

release-notes/2.3.1.md

@@ -0,0 +1,23 @@
+## What's Changed
+* Update state_storage.md by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/771
+* Update README.md to include state storage link by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/772
+* Updated offset state storage documentation. by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/774
+* Update state_storage.md to include schema storage. by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/775
+* Update state_storage.md to include postgresql offsets by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/776
+* Update quickstart.md to start sink connector service by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/785
+* Added logic to get latest release from github by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/786
+* Update quickstart.md with script to set environment variable. by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/787
+* Update quickstart_postgres.md by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/788
+* Added updates to script by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/789
+* Added single.threaded flag to Mariadb test to validate replication in… by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/781
+* Update production_setup.md by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/796
+* Changed logging level to info for STRUCT EMPTY not a valid CDC record by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/795
+* Update production_setup.md to include max_paritions_per_insert by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/804
+* Update production_setup.md , fixed broken link by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/809
+* Update config.ym by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/813
+* Update README.md to include initial data dump/load by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/821
+* 794 change logging level of struct empty not a valid cdc record + record to info by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/825
+* 801 records are not acknowledged or the offsets are not updated in singlethreaded mode by @subkanthi in https://github.com/Altinity/clickhouse-sink-connector/pull/827
+
+
+**Full Changelog**: https://github.com/Altinity/clickhouse-sink-connector/compare/2.3.0...2.3.1

sink-connector-lightweight/docker/config.yml

@@ -4,6 +4,11 @@
 # Unique name for the connector. Attempting to register again with the same name will fail.
 name: "company-1"
 
+# Primary key used for state storage. Refer to State Storage documentation.
+# If multiple connectors are writing to the same ClickHouse instance, this 
+# value needs to be unique for a connector.
+topic.prefix: "sink-connector-1"
+
 # IP address or hostname of the MySQL database server.
 database.hostname: "mysql-master"
 

sink-connector-lightweight/docker/log4j2.xml

@@ -19,8 +19,11 @@
                 additivity="false">
             <AppenderRef ref="console"/>
         </Logger>
-
-    <Root level="info" additivity="false">
+        <Logger name="io.debezium" level="ERROR"
+                additivity="false">
+            <AppenderRef ref="console"/>
+        </Logger>
+    <Root level="warn" additivity="false">
             <AppenderRef ref="console" />
         </Root>
     </Loggers>

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

@@ -125,7 +125,7 @@
                 "employees", clickHouseContainer.getUsername(), clickHouseContainer.getPassword(), null, chConn);
 
         long col2 = 0L;
         ResultSet version1Result = writer.executeQueryWithResultSet("select col2 from employees2.newtable final where col1 = 'a'");
         while(version1Result.next()) {
             col2 = version1Result.getLong("col2");
         }
@@ -148,7 +148,15 @@
         assertTrue(customersCol2 == 1);
 
 
-
+        Thread.sleep(10000);
+        // Execute the query in MySQL to rename table.
+        conn.prepareStatement("rename table products.prodtable to products.prodtable2").execute();
+        Thread.sleep(10000);
+        ResultSet customersVersionResult2 = writer.executeQueryWithResultSet("select col2 from customers.custtable2 final where col1 = 'a'");
+        while(customersVersionResult2.next()) {
+            customersCol2 = customersVersionResult2.getLong("col2");
+        }
+        assertTrue(customersCol2 == 2);
         clickHouseDebeziumEmbeddedApplication.getDebeziumEventCapture().engine.close();
 
         conn.close();

sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/AlterTableAddColumnIT.java

@@ -1,8 +1,10 @@
 package com.altinity.clickhouse.debezium.embedded.ddl.parser;
 
 import com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture;
+import com.altinity.clickhouse.debezium.embedded.config.SinkConnectorLightWeightConfig;
 import com.altinity.clickhouse.debezium.embedded.parser.SourceRecordParserService;
 import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfig;
+import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfigVariables;
 import com.altinity.clickhouse.sink.connector.db.BaseDbWriter;
 import com.clickhouse.jdbc.ClickHouseConnection;
 import org.apache.log4j.BasicConfigurator;
@@ -18,6 +20,7 @@
 import java.sql.Connection;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicReference;
@@ -50,8 +53,12 @@
         executorService.execute(() -> {
             try {
 
+                Properties properties = getDebeziumProperties();
+                // Add ddl.retry to true
+                //properties.put(SinkConnectorLightWeightConfig.DDL_RETRY, "true");
+
                 engine.set(new DebeziumChangeEventCapture());
-                engine.get().setup(getDebeziumProperties(), new SourceRecordParserService(),
+                engine.get().setup(properties, new SourceRecordParserService(),
                         new MySQLDDLParserService(new ClickHouseSinkConnectorConfig(new HashMap<>()),
                                 "employees"), false);
             } catch (Exception e) {
@@ -92,7 +99,7 @@
         Map<String, String> addTestColumns = writer.getColumnsDataTypesForTable("add_test");
 
         // Validate all ship_class columns.
         Assert.assertTrue(shipClassColumns.get("ship_spec").equalsIgnoreCase("Nullable(String)"));
         Assert.assertTrue(shipClassColumns.get("somecol").equalsIgnoreCase("Nullable(Int32)"));
         Assert.assertTrue(shipClassColumns.get("newcol").equalsIgnoreCase("Nullable(Bool)"));
         Assert.assertTrue(shipClassColumns.get("customer_address").equalsIgnoreCase("String"));