Rework metrics: Prometheus-friendly counters, flush duration, diagnostics (#48)

* Translate METRICS.md to English, retain Portuguese as METRICS-pt.md

* Wire up unrecorded metrics: schema ops, batch processing, record counts

Sensors for createTable, evolveSchema, schema-operation-*, batch-size-*,
and records-processed-* were registered but never recorded to in
production code. This connects them:

- DucklakeTableManager: instrument createTable/evolveTableSchema with
  schema operation timers
- DucklakeSinkTask: create DucklakeMetrics in start(), record
  batch/record metrics in flushBatches(), close in stop()
- DucklakeWriter: pass metrics through to DucklakeTableManager
- MetricTimer.close(): narrow to no checked exception

* Remove Portuguese METRICS-pt.md and translate remaining PT comment

* Reformat METRICS.md to use tables instead of bulleted lists

* Add per-partition flush duration metric using cumulative counters

Tracks flush write time (excluding consolidation) per partition via
flush-duration-total-ms and flush-count cumulative sums. These are
Prometheus-friendly — use rate() to derive averages, and they aggregate
correctly across partitions/tasks unlike windowed Avg/Max stats.

* Convert all metrics from windowed Avg/Max/Rate to CumulativeSum counters

Windowed stats (Avg, Max, Rate) don't aggregate correctly through the
JMX Exporter → Prometheus pipeline. Replace them with cumulative
counters that work with Prometheus rate() for correct cross-partition
and cross-task aggregation.

Renamed metrics:
- jdbc-query-time-avg/max/rate → jdbc-query-duration-total-ms
- schema-operation-time-avg/max → schema-operation-duration-total-ms
- operation-time-avg/max, operation-rate → operation-duration-total-ms
- records-processed-rate → removed (use rate(records-processed-total))
- batch-size-avg/max → batch-records-total + batch-count

* Prefix all metric names with ducklake- to avoid collisions

Kafka Connect shares a single Metrics registry across all connectors
in a worker. Generic names like jdbc-query-count could collide with
other connectors. Prefixing with ducklake- makes them unambiguous
regardless of tag filtering.

* Add diagnostic metrics: schema mismatch, flush skips, DLQ, spill, consolidation

New metrics for diagnosing perf and correctness issues:
- schema-mismatch-count: detects the many-small-files problem
- flush-skip-count: lock contention precursor to rebalance
- errant-record-count: records sent to DLQ (silent data loss signal)
- spill-batch-count/spill-bytes-total: disk spill volume
- consolidation-duration-total-ms/count: batch consolidation overhead

* Fix thread-safety: use ConcurrentHashMap for per-partition flush sensors

Flush sensors are lazily created via computeIfAbsent and accessed from
the put thread, scheduled flush thread, and partition executor threads.
HashMap.computeIfAbsent is not thread-safe and can corrupt internal
state under concurrent access.

* Fix thread leak: close Metrics registry in DucklakeMetrics.close()

new Metrics() starts an internal MetricsReporter thread and JMX
registrations. Without closing it, each task rebalance leaks a thread
and stale JMX beans, leading to OOM over hours in busy clusters.

* Only record evolveSchema metric when actual DDL executes

Previously the timer wrapped the entire method including the PRAGMA
metadata check that runs on every batch. Now it only fires when
columns are actually added, making the metric meaningful.

* Standardize all counter metric names with -total suffix

OpenMetrics/Prometheus convention: counters end with _total. Without
it, metrics like ducklake_batch_count look like gauges after JMX
Exporter converts hyphens to underscores.

* Use DucklakeMetricsInterface instead of concrete class in Writer/Factory

Allows injecting mock or no-op metrics implementations for testing.

* Add NoopDucklakeMetrics to eliminate null-check duplication

Null object pattern replaces all if-metrics-not-null branches with
unconditional calls. Reduces code duplication at every instrumentation
site and prevents forgetting a null check when adding new metrics.

* Change recordBatchProcessed to accept long instead of int

Avoids silent overflow on the long-to-int cast, even though current
defaults make it unlikely to hit.

* Remove reference to non-existent Grafana dashboard JSON

* Switch DucklakeMetrics from System.Logger to SLF4J

System.Logger uses JDK Platform Logging which may not bridge to the
same backend as SLF4J in Kafka Connect. Log output from DucklakeMetrics
(including unknown operation type warnings) could be lost or go to a
different handler.

* Add tests for new metrics: consolidation, spill, mismatch, DLQ, flush skip

Covers all metrics added in this branch plus concurrent flush sensor
creation and cleanup after close.

* Fix Metrics registry ownership: close at call site, not in DucklakeMetrics

DucklakeMetrics should not close a registry it doesn't own — if a
shared registry were ever passed in, close() would kill it for all
consumers. Move registry lifecycle to DucklakeSinkTask.stop() where
it was created. Also stores the registry reference so it's closed
even if start() fails partway through.

* Remove duplicate batch-records-total metric

records-processed-total and batch-records-total recorded identical
values. Keep records-processed-total + batch-count-total; derive
average batch size via rate(records) / rate(batch_count).

* Fix schema mismatch metric: record 1 per event, not output batch count

The metric now counts flushes that had schema mismatches, rather than
recording the number of output batches. Simpler, matches the docs,
and works correctly as a counter with rate().

* Clean up nits: dead null checks, unmodifiable maps, sensor naming, comment

- Remove dead null checks on final fields in close()
- Make operationTimingSensors/operationCountSensors unmodifiable after
  construction to prevent accidental mutation from other threads
- Normalize internal sensor names (remove inconsistent -sensor suffix)
- Add comment explaining load-bearing NoopDucklakeMetrics initializer

* Include type upgrades in evolveSchema metric timing

performTypeUpgrade() executes ALTER COLUMN DDL but was called inside
the field loop before the timer started. If only type upgrades occurred
(no new columns, _inserted_at exists), the method returned early
without recording the schema operation. Defer upgrades to execute
inside the timed block alongside column additions.

* Drop unused partition parameter from recordFlushSkip

The partition key was computed at every call site but ignored by the
implementation. Overall skip rate is sufficient for alerting; correlate
with per-partition flush duration to identify the hot partition.

* Move metrics close to finally block in stop() to prevent leak on error

connectionFactory.close() can throw, which would skip metrics cleanup
in the same try block. Moving to finally ensures the Metrics registry
thread and JMX registrations are always cleaned up.

Author: jghoman

METRICS.md

@@ -26,7 +26,7 @@ public interface DucklakeMetricsInterface extends AutoCloseable {
 
   void recordSchemaOperation(long durationNanos, String operationType);
 
-  void recordBatchProcessed(int recordCount);
+  void recordBatchProcessed(long recordCount);
 
   MetricTimer startJdbcQueryTimer();
 
@@ -36,6 +36,25 @@ public interface DucklakeMetricsInterface extends AutoCloseable {
 
   MetricTimer startSchemaOperationTimer(String operationType);
 
+  void recordFlushDuration(long durationNanos, String partition);
+
+  MetricTimer startFlushTimer(String partition);
+
+  void recordSchemaMismatch();
+
+  void recordFlushSkip();
+
+  void recordErrantRecord();
+
+  void recordSpill(long bytes);
+
+  void recordConsolidationDuration(long durationNanos);
+
+  MetricTimer startConsolidationTimer();
+
   /** Simple marker interface for timers returned by the metrics implementation. */
-  interface MetricTimer extends AutoCloseable {}
+  interface MetricTimer extends AutoCloseable {
+    @Override
+    void close();
+  }
 }

src/main/java/com/inyo/ducklake/connect/DucklakeMetrics.java

@@ -41,6 +41,7 @@
 import org.apache.arrow.memory.RootAllocator;
 import org.apache.arrow.vector.VectorSchemaRoot;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.sink.ErrantRecordReporter;
 import org.apache.kafka.connect.sink.SinkRecord;
@@ -88,6 +89,9 @@ public class DucklakeSinkTask extends SinkTask {
 
   // Errant record reporter for sending bad records to DLQ
   private ErrantRecordReporter errantRecordReporter;
+  private Metrics metricsRegistry;
+  // Initialized to no-op so stop() can call close() unconditionally even if start() fails
+  private DucklakeMetricsInterface ducklakeMetrics = NoopDucklakeMetrics.INSTANCE;
 
   // Jitter configuration to stagger flushes and avoid PG row-level contention
   // Each partition gets a random jitter offset (0 to maxJitterMs) applied to its flush timing
@@ -150,6 +154,10 @@ public String version() {
   @Override
   public void start(Map<String, String> map) {
     this.config = new DucklakeSinkConfig(DucklakeSinkConfig.CONFIG_DEF, map);
+    String connectorName = map.getOrDefault("name", "ducklake-sink");
+    String taskId = map.getOrDefault("task.id", "0");
+    this.metricsRegistry = new Metrics();
+    this.ducklakeMetrics = new DucklakeMetrics(metricsRegistry, connectorName, taskId);
     this.connectionFactory = new DucklakeConnectionFactory(config);
     this.writers = new HashMap<>();
     this.buffers = new HashMap<>();
@@ -299,6 +307,7 @@ private void checkTimeBasedFlushSpillable(long now) {
         AtomicInteger skips =
             consecutiveFlushSkips.computeIfAbsent(partition, k -> new AtomicInteger(0));
         int skipCount = skips.incrementAndGet();
+        ducklakeMetrics.recordFlushSkip();
         if (skipCount >= MAX_CONSECUTIVE_SKIPS_BEFORE_WARNING) {
           LOG.warn(
               "Flush check for partition {} skipped {} times - possible lock contention",
@@ -376,6 +385,7 @@ private void checkTimeBasedFlushInMemory(long now) {
         AtomicInteger skips =
             consecutiveFlushSkips.computeIfAbsent(partition, k -> new AtomicInteger(0));
         int skipCount = skips.incrementAndGet();
+        ducklakeMetrics.recordFlushSkip();
         if (skipCount >= MAX_CONSECUTIVE_SKIPS_BEFORE_WARNING) {
           LOG.warn(
               "Flush check for partition {} skipped {} times - possible lock contention",
@@ -427,7 +437,8 @@ public void open(Collection<TopicPartition> partitions) {
     super.open(partitions);
     try {
       this.connectionFactory.create();
-      this.writerFactory = new DucklakeWriterFactory(config, connectionFactory.getConnection());
+      this.writerFactory =
+          new DucklakeWriterFactory(config, connectionFactory.getConnection(), ducklakeMetrics);
 
       // Create one writer and buffer for each partition
       for (TopicPartition partition : partitions) {
@@ -453,7 +464,8 @@ public void open(Collection<TopicPartition> partitions) {
             }
           }
 
-          spillableBuffers.put(partition, new SpillablePartitionBuffer(partitionSpillDir));
+          spillableBuffers.put(
+              partition, new SpillablePartitionBuffer(partitionSpillDir, ducklakeMetrics));
           LOG.info("Created writer and spillable buffer for partition: {}", partition);
         } else {
           buffers.put(partition, new PartitionBuffer());
@@ -782,19 +794,30 @@ private void flushBatches(TopicPartition partition, FlushData flushData) {
 
       List<VectorSchemaRoot> consolidated;
       try {
-        consolidated = BatchConsolidator.consolidate(flushData.batches);
+        try (var timer = ducklakeMetrics.startConsolidationTimer()) {
+          consolidated = BatchConsolidator.consolidate(flushData.batches);
+        }
+        if (consolidated.size() > 1) {
+          ducklakeMetrics.recordSchemaMismatch();
+        }
       } catch (RuntimeException e) {
         // Consolidation failed mid-append — source batches may be partially consumed.
         // Close all original batches to avoid leaking Arrow memory.
         closeBatches(flushData.batches);
         throw e;
       }
       try {
-        for (VectorSchemaRoot root : consolidated) {
-          if (root.getRowCount() > 0) {
-            writer.write(root);
+        String partitionKey = partition.topic() + "-" + partition.partition();
+        try (var timer = ducklakeMetrics.startFlushTimer(partitionKey)) {
+          for (VectorSchemaRoot root : consolidated) {
+            if (root.getRowCount() > 0) {
+              writer.write(root);
+            }
           }
         }
+        if (flushData.recordCount > 0) {
+          ducklakeMetrics.recordBatchProcessed(flushData.recordCount);
+        }
       } catch (RuntimeException e) {
         LOG.error("Failed to write buffered data for partition: {}", partition, e);
         throw e;
@@ -1058,6 +1081,7 @@ private void handleSchemaConflictWithDLQ(
 
         try {
           errantRecordReporter.report(record, recordError);
+          ducklakeMetrics.recordErrantRecord();
         } catch (Exception dlqError) {
           LOG.error(
               "Failed to report record to DLQ: topic={}, partition={}, offset={}",
@@ -1208,6 +1232,20 @@ public void stop() {
       }
     } catch (Exception e) {
       throw new RuntimeException("Failed to stop DucklakeSinkTask", e);
+    } finally {
+      try {
+        ducklakeMetrics.close();
+      } catch (Exception e) {
+        LOG.warn("Failed closing metrics: {}", e.getMessage());
+      }
+      if (metricsRegistry != null) {
+        try {
+          metricsRegistry.close();
+        } catch (Exception e) {
+          LOG.warn("Failed closing metrics registry: {}", e.getMessage());
+        }
+        metricsRegistry = null;
+      }
     }
   }
 }

src/main/java/com/inyo/ducklake/connect/DucklakeMetricsInterface.java

@@ -24,14 +24,14 @@ public final class DucklakeWriterFactory {
 
   private final DucklakeSinkConfig config;
   private final DuckDBConnection conn;
-  private final DucklakeMetrics metrics;
+  private final DucklakeMetricsInterface metrics;
 
   public DucklakeWriterFactory(DucklakeSinkConfig config, DuckDBConnection conn) {
-    this(config, conn, null);
+    this(config, conn, NoopDucklakeMetrics.INSTANCE);
   }
 
   public DucklakeWriterFactory(
-      DucklakeSinkConfig config, DuckDBConnection conn, DucklakeMetrics metrics) {
+      DucklakeSinkConfig config, DuckDBConnection conn, DucklakeMetricsInterface metrics) {
     this.config = config;
     this.metrics = metrics;
     try {

src/main/java/com/inyo/ducklake/connect/DucklakeSinkTask.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2025 Inyo Contributors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.inyo.ducklake.connect;
+
+/** No-op metrics implementation that discards all recordings. */
+public final class NoopDucklakeMetrics implements DucklakeMetricsInterface {
+
+  public static final NoopDucklakeMetrics INSTANCE = new NoopDucklakeMetrics();
+
+  private static final MetricTimer NOOP_TIMER = () -> {};
+
+  private NoopDucklakeMetrics() {}
+
+  @Override
+  public void recordJdbcQuery(long durationNanos) {}
+
+  @Override
+  public void recordJdbcQuery(long durationNanos, String operationType) {}
+
+  @Override
+  public void recordSchemaOperation(long durationNanos) {}
+
+  @Override
+  public void recordSchemaOperation(long durationNanos, String operationType) {}
+
+  @Override
+  public void recordBatchProcessed(long recordCount) {}
+
+  @Override
+  public MetricTimer startJdbcQueryTimer() {
+    return NOOP_TIMER;
+  }
+
+  @Override
+  public MetricTimer startJdbcQueryTimer(String operationType) {
+    return NOOP_TIMER;
+  }
+
+  @Override
+  public MetricTimer startSchemaOperationTimer() {
+    return NOOP_TIMER;
+  }
+
+  @Override
+  public MetricTimer startSchemaOperationTimer(String operationType) {
+    return NOOP_TIMER;
+  }
+
+  @Override
+  public void recordFlushDuration(long durationNanos, String partition) {}
+
+  @Override
+  public MetricTimer startFlushTimer(String partition) {
+    return NOOP_TIMER;
+  }
+
+  @Override
+  public void recordSchemaMismatch() {}
+
+  @Override
+  public void recordFlushSkip() {}
+
+  @Override
+  public void recordErrantRecord() {}
+
+  @Override
+  public void recordSpill(long bytes) {}
+
+  @Override
+  public void recordConsolidationDuration(long durationNanos) {}
+
+  @Override
+  public MetricTimer startConsolidationTimer() {
+    return NOOP_TIMER;
+  }
+
+  @Override
+  public void close() {}
+}

src/main/java/com/inyo/ducklake/connect/DucklakeWriterFactory.java

@@ -44,6 +44,7 @@ public class SpillablePartitionBuffer {
   private static final Logger LOG = LoggerFactory.getLogger(SpillablePartitionBuffer.class);
 
   private final Path spillDirectory;
+  private final DucklakeMetricsInterface metrics;
   private final List<SpilledBatch> spilledBatches = new ArrayList<>();
   private long recordCount = 0;
   private long estimatedBytes = 0;
@@ -73,7 +74,15 @@ private static class SpilledBatch {
       value = "CT_CONSTRUCTOR_THROW",
       justification = "Failing fast on invalid spill directory is intentional")
   public SpillablePartitionBuffer(Path spillDirectory) {
+    this(spillDirectory, NoopDucklakeMetrics.INSTANCE);
+  }
+
+  @SuppressFBWarnings(
+      value = "CT_CONSTRUCTOR_THROW",
+      justification = "Failing fast on invalid spill directory is intentional")
+  public SpillablePartitionBuffer(Path spillDirectory, DucklakeMetricsInterface metrics) {
     this.spillDirectory = spillDirectory;
+    this.metrics = metrics;
     try {
       Files.createDirectories(spillDirectory);
     } catch (IOException e) {
@@ -106,6 +115,7 @@ public void add(VectorSchemaRoot root) {
       recordCount += root.getRowCount();
       estimatedBytes += byteSize;
 
+      metrics.recordSpill(byteSize);
       LOG.debug(
           "Spilled batch to {}: {} rows, {} bytes",
           spillFile.getFileName(),