Replaced slf4j calls with log4j2 api calls

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/api/DebeziumEmbeddedRestApi.java

@@ -8,10 +8,10 @@
 import com.google.inject.Injector;
 import io.javalin.Javalin;
 import io.javalin.http.HttpStatus;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.Properties;
 import java.util.concurrent.CompletableFuture;
@@ -21,7 +21,7 @@
 
 public class DebeziumEmbeddedRestApi {
 
-    private static final Logger log = LoggerFactory.getLogger(DebeziumEmbeddedRestApi.class);
+    private static final Logger log = LogManager.getLogger(DebeziumEmbeddedRestApi.class);
 
     public static void startRestApi(Properties props, Injector injector,
                              DebeziumChangeEventCapture debeziumChangeEventCapture,

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumChangeEventCapture.java

@@ -25,11 +25,11 @@
 import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.source.SourceRecord;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.json.simple.JSONArray;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.ParseException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.sql.ResultSet;
@@ -48,7 +48,7 @@
  */
 public class DebeziumChangeEventCapture {
 
-    private static final Logger log = LoggerFactory.getLogger(DebeziumChangeEventCapture.class);
+    private static final Logger log = LogManager.getLogger(DebeziumChangeEventCapture.class);
 
     private ClickHouseBatchExecutor executor;
 

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumConnectorCallback.java

@@ -2,11 +2,11 @@
 
 import io.debezium.engine.DebeziumEngine;
 import io.debezium.engine.DebeziumEngine.ConnectorCallback;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 public class DebeziumConnectorCallback implements ConnectorCallback {
-    private static final Logger log = LoggerFactory.getLogger(DebeziumConnectorCallback.class);
+    private static final Logger log = LogManager.getLogger(DebeziumConnectorCallback.class);
 
     public DebeziumConnectorCallback() {
     }

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/ErrorListenerImpl.java

@@ -6,14 +6,14 @@
 import org.antlr.v4.runtime.Recognizer;
 import org.antlr.v4.runtime.atn.ATNConfigSet;
 import org.antlr.v4.runtime.dfa.DFA;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.util.BitSet;
 
 public class ErrorListenerImpl implements ANTLRErrorListener {
 
-    private static final Logger log = LoggerFactory.getLogger(ErrorListenerImpl.class);
+    private static final Logger log = LogManager.getLogger(ErrorListenerImpl.class);
 
     @Override
     public void syntaxError(Recognizer<?, ?> recognizer, Object o, int i, int i1, String s, RecognitionException e) {

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySQLDDLParserService.java

@@ -11,8 +11,6 @@
 import org.antlr.v4.runtime.CommonTokenStream;
 import org.antlr.v4.runtime.Token;
 import org.antlr.v4.runtime.tree.ParseTreeWalker;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImpl.java

@@ -13,21 +13,21 @@
 import org.antlr.v4.runtime.tree.ParseTree;
 import org.antlr.v4.runtime.tree.TerminalNodeImpl;
 import org.antlr.v4.runtime.ParserRuleContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.time.ZoneId;
 import java.util.HashSet;
 import java.util.List;
 import java.util.ListIterator;
 import java.util.Set;
-import java.util.concurrent.ThreadLocalRandom;
+
 
 /**
  * This class contains the only overridden functions from the generated parser.
  */
 public class MySqlDDLParserListenerImpl extends MySQLDDLParserBaseListener {
-    private static final Logger log = LoggerFactory.getLogger(MySqlDDLParserListenerImpl.class);
+    private static final Logger log = LogManager.getLogger(MySqlDDLParserListenerImpl.class);
     StringBuffer query;
     String tableName;
     ClickHouseSinkConnectorConfig config;

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/parser/SourceRecordParserService.java

@@ -8,11 +8,11 @@
 import io.debezium.engine.DebeziumEngine;
 import org.apache.kafka.connect.data.*;
 import org.apache.kafka.connect.source.SourceRecord;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
 import org.json.simple.parser.ParseException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -24,7 +24,7 @@
 @Singleton
 public class SourceRecordParserService implements DebeziumRecordParserService {
 
-    private static final Logger log = LoggerFactory.getLogger(SourceRecordParserService.class);
+    private static final Logger log = LogManager.getLogger(SourceRecordParserService.class);
 
     @Override
     public ClickHouseStruct parse(ChangeEvent<SourceRecord, SourceRecord> record, DebeziumEngine.RecordCommitter<ChangeEvent<SourceRecord, SourceRecord>> committer,

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

@@ -2,22 +2,22 @@
 
 import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfig;
 import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfigVariables;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.junit.Assert;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 
 public class MySqlDDLParserListenerImplTest {
 
-    private static final Logger log = LoggerFactory.getLogger(MySqlDDLParserListenerImplTest.class);
+    private static final Logger log = LogManager.getLogger(MySqlDDLParserListenerImplTest.class);
 
     private static MySQLDDLParserService mySQLDDLParserService;
     @BeforeAll
@@ -43,7 +43,7 @@
         StringBuffer clickHouseQuery = new StringBuffer();
 
         mySQLDDLParserService.parseSql(createQuery, "Persons",  clickHouseQuery);
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE employees.employees_predated(emp_no Int32 NOT NULL ,birth_date Date32 NOT NULL ,first_name String NOT NULL ,last_name String NOT NULL ,gender String NOT NULL ,hire_date Date32 NOT NULL ,`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY (emp_no)"));
         log.info("Create table " + clickHouseQuery);
     }
     @Test
@@ -52,7 +52,7 @@
                 "PARTITION p2 VALUES LESS THAN (15,30,'sss'), PARTITION p3 VALUES LESS THAN (MAXVALUE,MAXVALUE,MAXVALUE));";
         StringBuffer clickHouseQuery = new StringBuffer();
         mySQLDDLParserService.parseSql(createQuery, "Persons",  clickHouseQuery);
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE employees.rcx(a Nullable(Int32),b Nullable(Int32),c Nullable(String),d Nullable(Int32),`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) PARTITION BY  (a,d,c) ORDER BY tuple()"));
         log.info("Create table " + clickHouseQuery);
     }
 
@@ -73,7 +73,7 @@
         StringBuffer clickHouseQuery = new StringBuffer();
         mySQLDDLParserService.parseSql(createQuery, "Persons", clickHouseQuery);
 
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE employees.t(id Nullable(Int32),dt Date32 NOT NULL ,`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) PARTITION BY  (dt) ORDER BY id"));
         log.info("Create table " + clickHouseQuery);
 
         String createQueryWithoutPrimaryKey =  "create table t(\n" +
@@ -106,7 +106,7 @@
                 "PARTITIONS 6;";
         StringBuffer clickHouseQuery = new StringBuffer();
         mySQLDDLParserService.parseSql(createQuery, "Persons", clickHouseQuery);
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE employees.members(firstname String NOT NULL ,lastname String NOT NULL ,username String NOT NULL ,email Nullable(String),joined Date32 NOT NULL ,`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) PARTITION BY  joined ORDER BY tuple()"));
         log.info("Create table " + clickHouseQuery);
     }
 
@@ -123,7 +123,7 @@
                 ") ENGINE=InnoDB DEFAULT CHARSET=latin1;";
         StringBuffer clickHouseQuery = new StringBuffer();
         mySQLDDLParserService.parseSql(createQuery6, "Persons", clickHouseQuery);
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE employees.`temporal_types_DATETIME4`(`Type` String NOT NULL ,`Minimum_Value` DateTime64(6, 0) NOT NULL ,`Mid_Value` DateTime64(6, 0) NOT NULL ,`Maximum_Value` DateTime64(6, 0) NOT NULL ,`Null_Value` Nullable(DateTime64(6, 0)),`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY (`Type`)"));
 
         String createQuery1 = "CREATE TABLE `temporal_types_DATETIME4` (\n" +
                 "  `Type` varchar(50) NOT NULL,\n" +
@@ -160,7 +160,7 @@
         String createQuery3 = "CREATE TABLE table_1 (id INT NOT NULL PRIMARY KEY, data DATETIME(3))";
         StringBuffer clickHouseQuery3 = new StringBuffer();
         mySQLDDLParserService.parseSql(createQuery3, "Persons", clickHouseQuery3);
         Assert.assertTrue(clickHouseQuery3.toString().equalsIgnoreCase("CREATE TABLE employees.table_1(id Int32 NOT NULL ,data Nullable(DateTime64(3, 0)),`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY id"));
 
         // DateTime(4) with Primary Key
         String createQuery4 = "CREATE TABLE table_1 (id INT NOT NULL PRIMARY KEY, data DATETIME(4))";
@@ -186,7 +186,7 @@
 
         MySQLDDLParserService mySQLDDLParserService1 = new MySQLDDLParserService(new ClickHouseSinkConnectorConfig(props), "datatypes");
         mySQLDDLParserService1.parseSql(createQuery6, "Persons", clickHouseQuery);
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE datatypes.`temporal_types_DATETIME4`(`Type` String NOT NULL ,`Minimum_Value` DateTime64(6,'UTC') NOT NULL ,`Mid_Value` DateTime64(6,'UTC') NOT NULL ,`Maximum_Value` DateTime64(6,'UTC') NOT NULL ,`Null_Value` Nullable(DateTime64(6,'UTC')),`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY (`Type`)"));
         log.info("Create table " + clickHouseQuery);
     }
 
@@ -207,7 +207,7 @@
 
         MySQLDDLParserService mySQLDDLParserService1 = new MySQLDDLParserService(new ClickHouseSinkConnectorConfig(props), "datatypes");
         mySQLDDLParserService1.parseSql(createQuery6, "Persons", clickHouseQuery);
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE datatypes.`temporal_types_DATETIME4`(`Type` String NOT NULL ,`Minimum_Value` DateTime64(1,'UTC') NOT NULL ,`Mid_Value` DateTime64(2,'UTC') NOT NULL ,`Maximum_Value` DateTime64(3,'UTC') NOT NULL ,`Null_Value` Nullable(DateTime64(4,'UTC')),`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY (`Type`)"));
         log.info("Create table " + clickHouseQuery);
     }
 
@@ -218,7 +218,7 @@
         mySQLDDLParserService.parseSql(createDB, "Persons", clickHouseQuery);
         log.info("Create table " + clickHouseQuery);
 
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE if not exists employees.730b595f_d475_11ed_b64a_398b553542b2(id Nullable(Int32),x Nullable(Int32),`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY (id)"));
     }
 
     @Test
@@ -234,7 +234,7 @@
         String createDB = "create table if not exists ship_class(id int, class_name varchar(100), tonange decimal(10,2), max_length decimal(10,2), start_build year, end_build year(4), max_guns_size int)";
         mySQLDDLParserService.parseSql(createDB, "Persons", clickHouseQuery);
 
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE if not exists employees.ship_class(id Nullable(Int32),class_name Nullable(String),tonange Nullable(Decimal(10,2)),max_length Nullable(Decimal(10,2)),start_build Nullable(Int32),end_build Nullable(Int32),max_guns_size Nullable(Int32),`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY tuple()"));
         log.info("Create table " + clickHouseQuery);
 
     }
@@ -244,7 +244,7 @@
         String createDB = "create table ship_class(id int, class_name varchar(100), tonange decimal(10,2) not null, max_length decimal(65,2), start_build year, end_build year(4), max_guns_size int)";
         mySQLDDLParserService.parseSql(createDB, "Persons", clickHouseQuery);
 
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE employees.ship_class(id Nullable(Int32),class_name Nullable(String),tonange Decimal(10,2) NOT NULL ,max_length Nullable(Decimal(65,2)),start_build Nullable(Int32),end_build Nullable(Int32),max_guns_size Nullable(Int32),`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY tuple()"));
         log.info("Create table " + clickHouseQuery);
 
     }
@@ -255,7 +255,7 @@
         StringBuffer clickHouseQuery = new StringBuffer();
         mySQLDDLParserService.parseSql(createDBQuery, "Persons", clickHouseQuery);
 
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE if not exists employees.730b595f_d475_11ed_b64a_398b553542b2(id Nullable(Int32),x Nullable(Int32),`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY (id)"));
         log.info("Create table " + clickHouseQuery);
 
     }
@@ -273,7 +273,7 @@
                 ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci";
         mySQLDDLParserService.parseSql(createDB, "Persons", clickHouseQuery);
 
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE employees.`salaries`(`emp_no` Int32 NOT NULL ,`salary` Int32 NOT NULL ,`from_date` Date32 NOT NULL ,`to_date` Date32 NOT NULL ,`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY (`emp_no`,`from_date`)"));
         log.info("Create table query" + clickHouseQuery.toString());
     }
 
@@ -516,7 +516,7 @@
         String sql = "truncate table add_test";
         mySQLDDLParserService.parseSql(sql, "table1", clickHouseQuery);
 
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(sql));
     }
 
     @Test
@@ -644,7 +644,7 @@
         String sql = "CREATE TABLE employees.contacts (fullname varchar(101) GENERATED ALWAYS AS (CONCAT(first_name,' ',last_name)), email VARCHAR(100) NOT NULL);";
         mySQLDDLParserService.parseSql(sql, "", clickHouseQuery);
 
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE employees.contacts(fullname Nullable(String) MATERIALIZED CONCAT(first_name,' ',last_name),email String NOT NULL ,`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY tuple()"));
     }
 
     @Test
@@ -654,7 +654,7 @@
         String sql = "create table new_table(col1 varchar(255), col2 int, is_deleted int, _sign int);";
         mySQLDDLParserService.parseSql(sql, "", clickHouseQuery);
 
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE employees.new_table(col1 Nullable(String),col2 Nullable(Int32),is_deleted Nullable(Int32),_sign Nullable(Int32),`_version` UInt64,`__is_deleted` UInt8) Engine=ReplacingMergeTree(_version,__is_deleted) ORDER BY tuple()"));
     }
 
     @ParameterizedTest
@@ -690,7 +690,7 @@
         mySQLDDLParserService.parseSql(sql, "temporal_types_DATETIME4", clickHouseQuery, isDropOrTruncate);
 
         String expectedResult = "CREATE TABLE datatypes.temporal_types_TIMESTAMP1 ON CLUSTER `{cluster}`(`Mid_Value` DateTime64(1, 0) NOT NULL ,`_version` UInt64,`is_deleted` UInt8)Engine=ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/temporal_types_TIMESTAMP1', '{replica}', _version, is_deleted) ORDER BY tuple()";
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(expectedResult));
 
 
     }
@@ -710,7 +710,7 @@
 
         AtomicBoolean isDropOrTruncate = new AtomicBoolean();
         mySQLDDLParserService.parseSql(sql, "", clickHouseQuery, isDropOrTruncate);
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(expectedResult));
 
     }
 
@@ -729,7 +729,7 @@
 
         AtomicBoolean isDropOrTruncate = new AtomicBoolean();
         mySQLDDLParserService.parseSql(sql, "", clickHouseQuery, isDropOrTruncate);
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(expectedResult));
 
     }
     @Test

sink-connector/pom.xml

@@ -388,24 +388,18 @@
             <scope>test</scope>
         </dependency>
 
-        <!--Mockito for unit test-->
-        <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-api</artifactId>
-            <version>2.0.11</version>
-        </dependency>
         <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-log4j12</artifactId>
-            <version>2.0.11</version>
-            <scope>test</scope>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-core</artifactId>
+            <version>2.23.1</version>
         </dependency>
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>
-            <artifactId>log4j-core</artifactId>
-            <version>2.17.1</version>
-            <scope>test</scope>
+            <artifactId>log4j-api</artifactId>
+            <version>2.23.1</version>
         </dependency>
+
+        <!--Mockito for unit test-->
         <dependency>
             <groupId>com.github.stefanbirkner</groupId>
             <artifactId>system-rules</artifactId>

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnector.java

@@ -7,8 +7,8 @@
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.connect.connector.Task;
 import org.apache.kafka.connect.sink.SinkConnector;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -20,7 +20,7 @@ public class ClickHouseSinkConnector extends SinkConnector {
 
     // String configuration as properties
     private Map<String, String> config;
-    private static final Logger log = LoggerFactory.getLogger(ClickHouseSinkConnector.class);
+    private static final Logger log = LogManager.getLogger(ClickHouseSinkConnector.class);
     private boolean ready;
 
     /**

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfig.java

@@ -9,8 +9,9 @@
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.config.ConfigDef.Type;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
 
 import java.util.Map;
 
@@ -23,7 +24,7 @@ public class ClickHouseSinkConnectorConfig extends AbstractConfig {
 
     public static long BUFFER_COUNT_DEFAULT = 100;
 
-    private static final Logger log = LoggerFactory.getLogger(ClickHouseSinkConnectorConfig.class.getName());
+    private static final Logger log = LogManager.getLogger(ClickHouseSinkConnectorConfig.class);
 
     // Configuration groups
 

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkTask.java

@@ -13,8 +13,8 @@
 import org.apache.kafka.connect.errors.RetriableException;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.sink.SinkTask;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.util.*;
 import java.util.concurrent.*;
@@ -27,7 +27,7 @@
 public class ClickHouseSinkTask extends SinkTask {
 
     private String id = "-1";
-    private static final Logger log = LoggerFactory.getLogger(ClickHouseSinkTask.class);
+    private static final Logger log = LogManager.getLogger(ClickHouseSinkTask.class);
 
     public ClickHouseSinkTask() {
 

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/common/Logging.java

@@ -15,8 +15,10 @@
 package com.altinity.clickhouse.sink.connector.common;
 
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.util.Objects;
 
@@ -25,7 +27,7 @@
  */
 public class Logging {
     // todo: change to interface when upgrading to Java 9 or later
-    private final Logger log = LoggerFactory.getLogger(getClass().getName());
+    private final Logger log = LogManager.getLogger(getClass().getName());
 
     // only message
     protected void logInfo(String msg) {

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/common/Metrics.java

@@ -29,7 +29,8 @@
 import io.prometheus.client.CollectorRegistry;
 import io.prometheus.client.Gauge;
 import org.apache.commons.lang3.tuple.MutablePair;
-import org.slf4j.Logger;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.io.IOException;
 import java.io.OutputStream;
@@ -43,7 +44,7 @@
  */
 public class Metrics {
 
-    private static final Logger log = org.slf4j.LoggerFactory.getLogger(Metrics.class);
+    private static final Logger log = LogManager.getLogger(Metrics.class);
     private static MetricRegistry registry = null;
     private static CollectorRegistry collectorRegistry;
     private static PrometheusMeterRegistry meterRegistry;

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/common/Utils.java

@@ -16,14 +16,14 @@
 package com.altinity.clickhouse.sink.connector.common;
 
 import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfigVariables;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.util.HashMap;
 import java.util.Map;
 
 public class Utils {
-    private static final Logger LOGGER = LoggerFactory.getLogger(Utils.class);
+    private static final Logger LOGGER = LogManager.getLogger(Utils.class);
 
     public static final String TASK_ID = "task_id";
     // Connector version, change every release

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseConverter.java

@@ -8,15 +8,15 @@
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.source.SourceRecord;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 public class ClickHouseConverter implements AbstractConverter {
-    private static final Logger log = LoggerFactory.getLogger(ClickHouseConverter.class);
+    private static final Logger log = LogManager.getLogger(ClickHouseConverter.class);
 
     /**
      * Enum to store the OP Types.

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/DebeziumConverter.java

@@ -1,22 +1,15 @@
 package com.altinity.clickhouse.sink.connector.converters;
 
 import com.altinity.clickhouse.sink.connector.metadata.DataTypeRange;
-import com.clickhouse.data.ClickHouseChecker;
 import com.clickhouse.data.ClickHouseDataType;
-import com.clickhouse.data.ClickHouseValues;
 import com.clickhouse.data.format.BinaryStreamUtils;
 import org.apache.commons.lang3.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
-import java.sql.Date;
-import java.sql.Timestamp;
-import java.time.*;
+import java.sql.Date;import java.time.*;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
-import java.util.Calendar;
-import java.util.TimeZone;
-import java.util.concurrent.TimeUnit;
 
 import static java.time.Instant.ofEpochMilli;
 
@@ -25,7 +18,7 @@ public class DebeziumConverter {
     private static final int MICROS_IN_SEC = 1000000;
     private static final int MICROS_IN_MILLI = 1000;
 
-    private static final Logger log = LoggerFactory.getLogger(DebeziumConverter.class);
+    private static final Logger log = LogManager.getLogger(DebeziumConverter.class);
 
 
     public static class MicroTimeConverter {

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/BaseDbWriter.java

@@ -4,8 +4,6 @@
 import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfigVariables;
 import com.clickhouse.jdbc.ClickHouseConnection;
 import com.clickhouse.jdbc.ClickHouseDataSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -14,6 +12,8 @@
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Properties;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 public class BaseDbWriter {
 
@@ -29,7 +29,7 @@ public class BaseDbWriter {
 
     private ClickHouseSinkConnectorConfig config;
 
-    private static final Logger log = LoggerFactory.getLogger(BaseDbWriter.class);
+    private static final Logger log = LogManager.getLogger(BaseDbWriter.class);
 
     public BaseDbWriter(
             String hostName,