[monitoring] Impl metrics

build.gradle.kts

@@ -54,6 +54,7 @@ extra.apply {
     set("kafkaVersion", "2.7.0")
     set("gson", "2.13.1")
     set("jackson", "2.19.1")
+    set("dropwizardMetrics", "5.0.0")
 
     // Testing dependencies
     set("junitJupiterVersion", "5.9.2")
@@ -86,9 +87,10 @@ dependencies {
     implementation("com.clickhouse:clickhouse-data:${project.extra["clickHouseDriverVersion"]}")
     implementation("com.clickhouse:client-v2:${project.extra["clickHouseDriverVersion"]}")
     implementation("com.google.code.gson:gson:${project.extra["gson"]}")
-
     implementation("org.apache.httpcomponents.client5:httpclient5:5.5")
 
+    implementation("io.dropwizard:dropwizard-core:${project.extra["dropwizardMetrics"]}")
+
     // Avoid telescoping constructors problem with the builder pattern using Lombok
     compileOnly("org.projectlombok:lombok:1.18.38")
     annotationProcessor("org.projectlombok:lombok:1.18.38")
@@ -109,6 +111,7 @@ dependencies {
     clickhouseDependencies("com.fasterxml.jackson.core:jackson-core:${project.extra["jackson"]}")
     clickhouseDependencies("com.fasterxml.jackson.core:jackson-databind:${project.extra["jackson"]}")
     clickhouseDependencies("com.fasterxml.jackson.core:jackson-annotations:${project.extra["jackson"]}")
+    clickhouseDependencies("io.dropwizard:dropwizard-core:${project.extra["dropwizardMetrics"]}")
 
     // Unit Tests
     testImplementation(platform("org.junit:junit-bom:${project.extra["junitJupiterVersion"]}"))

src/main/java/com/clickhouse/kafka/connect/sink/ClickHouseSinkTask.java

@@ -103,4 +103,7 @@ static ErrorReporter devNullErrorReporter() {
         };
     }
 
+    public int taskId() {
+        return this.proxySinkTask == null ? Integer.MAX_VALUE : this.proxySinkTask.getId();
+    }
 }

src/main/java/com/clickhouse/kafka/connect/sink/ProxySinkTask.java

@@ -1,6 +1,5 @@
 package com.clickhouse.kafka.connect.sink;
 
-import com.clickhouse.client.config.ClickHouseClientOption;
 import com.clickhouse.kafka.connect.sink.data.Record;
 import com.clickhouse.kafka.connect.sink.db.ClickHouseWriter;
 import com.clickhouse.kafka.connect.sink.db.DBWriter;
@@ -11,7 +10,6 @@
 import com.clickhouse.kafka.connect.sink.state.provider.InMemoryState;
 import com.clickhouse.kafka.connect.sink.state.provider.KeeperStateProvider;
 import com.clickhouse.kafka.connect.util.jmx.ExecutionTimer;
-import com.clickhouse.kafka.connect.util.jmx.MBeanServerUtils;
 import com.clickhouse.kafka.connect.util.jmx.SinkTaskStatistics;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.slf4j.Logger;
@@ -30,14 +28,14 @@ public class ProxySinkTask {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(ProxySinkTask.class);
     private static final AtomicInteger NEXT_ID = new AtomicInteger();
-    private Processing processing = null;
-    private StateProvider stateProvider = null;
-    private DBWriter dbWriter = null;
-    private ClickHouseSinkConfig clickHouseSinkConfig = null;
+    private final Processing processing;
+    private final StateProvider stateProvider;
+    private final DBWriter dbWriter;
+    private final ClickHouseSinkConfig clickHouseSinkConfig;
     private Timer tableRefreshTimer;
 
     private final SinkTaskStatistics statistics;
-    private int id = NEXT_ID.getAndAdd(1);
+    private final int id = NEXT_ID.getAndAdd(1);
 
     public ProxySinkTask(final ClickHouseSinkConfig clickHouseSinkConfig, final ErrorReporter errorReporter) {
         this.clickHouseSinkConfig = clickHouseSinkConfig;
@@ -47,8 +45,10 @@ public ProxySinkTask(final ClickHouseSinkConfig clickHouseSinkConfig, final Erro
         } else {
             this.stateProvider = new InMemoryState();
         }
+        this.statistics = new SinkTaskStatistics(id);
+        this.statistics.registerMBean();
 
-        ClickHouseWriter chWriter = new ClickHouseWriter();
+        ClickHouseWriter chWriter = new ClickHouseWriter(this.statistics);
         this.dbWriter = chWriter;
 
         // Add table mapping refresher
@@ -62,26 +62,12 @@ public ProxySinkTask(final ClickHouseSinkConfig clickHouseSinkConfig, final Erro
         boolean isStarted = dbWriter.start(clickHouseSinkConfig);
         if (!isStarted)
             throw new RuntimeException("Connection to ClickHouse is not active.");
-        processing = new Processing(stateProvider, dbWriter, errorReporter, clickHouseSinkConfig);
+        processing = new Processing(stateProvider, dbWriter, errorReporter, clickHouseSinkConfig, statistics);
 
-        this.statistics = MBeanServerUtils.registerMBean(new SinkTaskStatistics(), getMBeanNAme());
-    }
-
-    private String getMBeanNAme() {
-        return String.format("com.clickhouse:type=ClickHouseKafkaConnector,name=SinkTask%d,version=%s", id, com.clickhouse.kafka.connect.sink.Version.ARTIFACT_VERSION);
     }
 
     public void stop() {
-        if (tableRefreshTimer != null) {
-            try {
-                tableRefreshTimer.cancel();
-            } catch (Exception e) {
-                LOGGER.error("Error canceling table refresh timer on com.clickhouse.kafka.connect.sink.ProxySinkTask.stop", e);
-            }
-        }
-
-        LOGGER.info("Stopping MBean server {}", getMBeanNAme());
-        MBeanServerUtils.unregisterMBean(getMBeanNAme());
+        statistics.unregisterMBean();
     }
 
     public void put(final Collection<SinkRecord> records) throws IOException, ExecutionException, InterruptedException {
@@ -90,8 +76,8 @@ public void put(final Collection<SinkRecord> records) throws IOException, Execut
             return;
         }
         // Group by topic & partition
+        statistics.receivedRecords(records);
         ExecutionTimer taskTime = ExecutionTimer.start();
-        statistics.receivedRecords(records.size());
         LOGGER.trace(String.format("Got %d records from put API.", records.size()));
         ExecutionTimer processingTime = ExecutionTimer.start();
 
@@ -112,4 +98,8 @@ public void put(final Collection<SinkRecord> records) throws IOException, Execut
         }
         statistics.taskProcessingTime(taskTime);
     }
+
+    public int getId() {
+        return id;
+    }
 }

src/main/java/com/clickhouse/kafka/connect/sink/db/ClickHouseWriter.java

@@ -28,6 +28,7 @@
 import com.clickhouse.kafka.connect.sink.dlq.ErrorReporter;
 import com.clickhouse.kafka.connect.util.QueryIdentifier;
 import com.clickhouse.kafka.connect.util.Utils;
+import com.clickhouse.kafka.connect.util.jmx.SinkTaskStatistics;
 import com.google.gson.ExclusionStrategy;
 import com.google.gson.FieldAttributes;
 import com.google.gson.Gson;
@@ -75,9 +76,10 @@ public class ClickHouseWriter implements DBWriter {
 
     private Map<String, Table> mapping = null;
     private AtomicBoolean isUpdateMappingRunning = new AtomicBoolean(false);
-
-    public ClickHouseWriter() {
+    private final SinkTaskStatistics statistics;
+    public ClickHouseWriter(SinkTaskStatistics statistics) {
         this.mapping = new HashMap<String, Table>();
+        this.statistics = statistics;
     }
 
     protected void setClient(ClickHouseHelperClient chc) {
@@ -199,7 +201,6 @@ public void doInsert(List<Record> records, QueryIdentifier queryId, ErrorReporte
         Table table = getTable(database, topic);
         if (table == null) { return; }//We checked the error flag in getTable, so we don't need to check it again here
         LOGGER.debug("Trying to insert [{}] records to table name [{}] (QueryId: [{}])", records.size(), table.getName(), queryId.getQueryId());
-
         switch (first.getSchemaType()) {
             case SCHEMA:
                 if (csc.isBypassRowBinary()) {
@@ -837,6 +838,8 @@ protected void doInsertRawBinaryV2(List<Record> records, Table table, QueryIdent
         }
         long s3 = System.currentTimeMillis();
         LOGGER.info("topic: {} partition: {} batchSize: {} push stream ms: {} data ms: {} send ms: {} (QueryId: [{}])", topic, partition, records.size(), pushStreamTime,s2 - s1, s3 - s2, queryId.getQueryId());
+        statistics.insertTime(s2 - s1, topic);
+
     }
     protected void doInsertRawBinaryV1(List<Record> records, Table table, QueryIdentifier queryId, boolean supportDefaults) throws IOException, ExecutionException, InterruptedException {
         long s1 = System.currentTimeMillis();
@@ -892,6 +895,7 @@ protected void doInsertRawBinaryV1(List<Record> records, Table table, QueryIdent
 
         long s3 = System.currentTimeMillis();
         LOGGER.info("topic :{} partition: {} batchSize: {} push stream ms: {} data ms: {} send ms: {} (QueryId: [{}])", topic, partition, records.size(), pushStreamTime,s2 - s1, s3 - s2, queryId.getQueryId());
+        statistics.insertTime(s2 - s1, topic);
     }
 
     protected void doInsertJson(List<Record> records, Table table, QueryIdentifier queryId) throws IOException, ExecutionException, InterruptedException {
@@ -951,12 +955,15 @@ protected void doInsertJsonV1(List<Record> records, Table table, QueryIdentifier
                         long beforeSerialize = System.currentTimeMillis();
                         String gsonString = gson.toJson(cleanupExtraFields(data, table), gsonType);
                         dataSerializeTime += System.currentTimeMillis() - beforeSerialize;
+
                         LOGGER.trace("topic {} partition {} offset {} payload {}",
                                 record.getTopic(),
                                 record.getRecordOffsetContainer().getPartition(),
                                 record.getRecordOffsetContainer().getOffset(),
                                 gsonString);
-                        BinaryStreamUtils.writeBytes(stream, gsonString.getBytes(StandardCharsets.UTF_8));
+                        byte[] bytes = gsonString.getBytes(StandardCharsets.UTF_8);
+                        statistics.bytesInserted(bytes.length);
+                        BinaryStreamUtils.writeBytes(stream, bytes);
                     } else {
                         LOGGER.warn(String.format("Getting empty record skip the insert topic[%s] offset[%d]", record.getTopic(), record.getSinkRecord().kafkaOffset()));
                     }
@@ -972,6 +979,7 @@ protected void doInsertJsonV1(List<Record> records, Table table, QueryIdentifier
         }
         long s3 = System.currentTimeMillis();
         LOGGER.info("topic: {} partition: {} batchSize: {} serialization ms: {} data ms: {} send ms: {} (QueryId: [{}])", topic, partition, records.size(), dataSerializeTime, s2 - s1, s3 - s2, queryId.getQueryId());
+        statistics.insertTime(s2 - s1, topic);
     }
 
     protected void doInsertJsonV2(List<Record> records, Table table, QueryIdentifier queryId) throws IOException, ExecutionException, InterruptedException {
@@ -1031,7 +1039,9 @@ protected void doInsertJsonV2(List<Record> records, Table table, QueryIdentifier
                         record.getRecordOffsetContainer().getPartition(),
                         record.getRecordOffsetContainer().getOffset(),
                         gsonString);
-                BinaryStreamUtils.writeBytes(stream, gsonString.getBytes(StandardCharsets.UTF_8));
+                byte[] bytes = gsonString.getBytes(StandardCharsets.UTF_8);
+                statistics.bytesInserted(bytes.length);
+                BinaryStreamUtils.writeBytes(stream, bytes);
             } else {
                 LOGGER.warn(String.format("Getting empty record skip the insert topic[%s] offset[%d]", record.getTopic(), record.getSinkRecord().kafkaOffset()));
             }
@@ -1044,6 +1054,7 @@ protected void doInsertJsonV2(List<Record> records, Table table, QueryIdentifier
         }
         long s3 = System.currentTimeMillis();
         LOGGER.info("topic: {} partition: {} batchSize: {} serialization ms: {} data ms: {} send ms: {} (QueryId: [{}])", topic, partition, records.size(), dataSerializeTime, s2 - s1, s3 - s2, queryId.getQueryId());
+        statistics.insertTime(s2 - s1, topic);
     }
 
     protected Map<String, Object> cleanupExtraFields(Map<String, Object> m, Table t) {
@@ -1108,6 +1119,7 @@ protected void doInsertStringV1(List<Record> records, Table table, QueryIdentifi
                         String data = (String)record.getSinkRecord().value();
                         LOGGER.debug(String.format("data: %s", data));
                         byte[] bytes = data.getBytes(StandardCharsets.UTF_8);
+                        statistics.bytesInserted(bytes.length);
                         long beforePushStream = System.currentTimeMillis();
                         BinaryStreamUtils.writeBytes(stream, bytes);
                         pushStreamTime += System.currentTimeMillis() - beforePushStream;
@@ -1134,6 +1146,7 @@ protected void doInsertStringV1(List<Record> records, Table table, QueryIdentifi
         }
         long s3 = System.currentTimeMillis();
         LOGGER.info("topic: {} partition: {} batchSize: {} push stream ms: {} data ms: {} send ms: {} (QueryId: [{}])", topic, partition, records.size(), pushStreamTime, s2 - s1, s3 - s2, queryId.getQueryId());
+        statistics.insertTime(s2 - s1, topic);
     }
     protected void doInsertStringV2(List<Record> records, Table table, QueryIdentifier queryId) throws IOException, ExecutionException, InterruptedException {
         byte[] endingLine = new byte[]{'\n'};
@@ -1182,6 +1195,7 @@ protected void doInsertStringV2(List<Record> records, Table table, QueryIdentifi
                 String data = (String)record.getSinkRecord().value();
                 LOGGER.debug(String.format("data: %s", data));
                 byte[] bytes = data.getBytes(StandardCharsets.UTF_8);
+                statistics.bytesInserted(bytes.length);
                 long beforePushStream = System.currentTimeMillis();
                 BinaryStreamUtils.writeBytes(stream, bytes);
                 pushStreamTime += System.currentTimeMillis() - beforePushStream;
@@ -1204,6 +1218,7 @@ protected void doInsertStringV2(List<Record> records, Table table, QueryIdentifi
         }
         long s3 = System.currentTimeMillis();
         LOGGER.info("topic: {} partition: {} batchSize: {} push stream ms: {} data ms: {} send ms: {} (QueryId: [{}])", topic, partition, records.size(), pushStreamTime, s2 - s1, s3 - s2, queryId.getQueryId());
+        statistics.insertTime(s2 - s1, topic);
     }
 
 

src/main/java/com/clickhouse/kafka/connect/sink/processing/Processing.java

@@ -7,11 +7,11 @@
 import com.clickhouse.kafka.connect.sink.dlq.ErrorReporter;
 import com.clickhouse.kafka.connect.sink.kafka.RangeContainer;
 import com.clickhouse.kafka.connect.sink.kafka.RangeState;
-import com.clickhouse.kafka.connect.sink.state.State;
 import com.clickhouse.kafka.connect.sink.state.StateProvider;
 import com.clickhouse.kafka.connect.sink.state.StateRecord;
 import com.clickhouse.kafka.connect.util.QueryIdentifier;
 import com.clickhouse.kafka.connect.util.Utils;
+import com.clickhouse.kafka.connect.util.jmx.SinkTaskStatistics;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -28,15 +28,17 @@ public class Processing {
     private StateProvider stateProvider = null;
     private DBWriter dbWriter = null;
     private ClickHouseSinkConfig clickHouseSinkConfig;
-
+    private SinkTaskStatistics statistics;
 
     private ErrorReporter errorReporter = null;
 
-    public Processing(StateProvider stateProvider, DBWriter dbWriter, ErrorReporter errorReporter, ClickHouseSinkConfig clickHouseSinkConfig) {
+    public Processing(StateProvider stateProvider, DBWriter dbWriter, ErrorReporter errorReporter,
+                      ClickHouseSinkConfig clickHouseSinkConfig, SinkTaskStatistics statistics) {
         this.stateProvider = stateProvider;
         this.dbWriter = dbWriter;
         this.errorReporter = errorReporter;
         this.clickHouseSinkConfig = clickHouseSinkConfig;
+        this.statistics = statistics;
     }
     /**
      * the logic is only for topic partition scoop
@@ -49,14 +51,23 @@ private void doInsert(List<Record> records, RangeContainer rangeContainer) {
             LOGGER.trace("doInsert - No records to insert.");
             return;
         }
-        QueryIdentifier queryId = new QueryIdentifier(records.get(0).getRecordOffsetContainer().getTopic(), records.get(0).getRecordOffsetContainer().getPartition(),
+
+        final Record firstRecord = records.get(0);
+        long eventReceiveLag = 0;
+        if (firstRecord.getSinkRecord().timestamp() != null) {
+            eventReceiveLag = System.currentTimeMillis() - firstRecord.getSinkRecord().timestamp();
+        }
+        final String topic = firstRecord.getRecordOffsetContainer().getTopic();
+        QueryIdentifier queryId = new QueryIdentifier(topic, records.get(0).getRecordOffsetContainer().getPartition(),
                 rangeContainer.getMinOffset(), rangeContainer.getMaxOffset(),
                 UUID.randomUUID().toString());
 
         try {
             LOGGER.debug("doInsert - Records: [{}] - {}", records.size(), queryId);
             dbWriter.doInsert(records, queryId, errorReporter);
+            statistics.recordTopicStats(records.size(), queryId.getTopic(), eventReceiveLag);
         } catch (Exception e) {
+            statistics.recordTopicStatsOnFailure(records.size(), queryId.getTopic(), eventReceiveLag);
             throw new RuntimeException(queryId.toString(), e);//This way the queryId will propagate
         }
     }
@@ -66,12 +77,21 @@ private void doInsert(List<Record> records) {
             LOGGER.trace("doInsert - No records to insert.");
             return;
         }
-        QueryIdentifier queryId = new QueryIdentifier(records.get(0).getRecordOffsetContainer().getTopic(), UUID.randomUUID().toString());
+
+        final Record firstRecord = records.get(0);
+        long eventReceiveLag = 0;
+        if (firstRecord.getSinkRecord().timestamp() != null) {
+            eventReceiveLag = System.currentTimeMillis() - firstRecord.getSinkRecord().timestamp();
+        }
+        final String topic = firstRecord.getRecordOffsetContainer().getTopic();
+        QueryIdentifier queryId = new QueryIdentifier(topic, UUID.randomUUID().toString());
 
         try {
             LOGGER.info("doInsert - Records: [{}] - {}", records.size(), queryId);
             dbWriter.doInsert(records, queryId, errorReporter);
+            statistics.recordTopicStats(records.size(), topic, eventReceiveLag);
         } catch (Exception e) {
+            statistics.recordTopicStatsOnFailure(records.size(), topic, eventReceiveLag);
             throw new RuntimeException(queryId.toString(), e);//This way the queryId will propagate
         }
     }