From a69967c0b490a38724080007657cfadd2d6dd0f8 Mon Sep 17 00:00:00 2001
From: Ismail Simsek <ismail.simsek@ppro.com>
Date: Fri, 17 Sep 2021 11:51:30 +0200
Subject: [PATCH] Minor improve metrics logging

---
 .../java/io/debezium/server/batch/DebeziumMetrics.java | 10 ++++++++++
 .../server/batch/batchsizewait/MaxBatchSizeWait.java   |  6 ++++--
 2 files changed, 14 insertions(+), 2 deletions(-)

diff --git a/debezium-server-batch-sinks/src/main/java/io/debezium/server/batch/DebeziumMetrics.java b/debezium-server-batch-sinks/src/main/java/io/debezium/server/batch/DebeziumMetrics.java
index 0e824689..b01f3159 100644
--- a/debezium-server-batch-sinks/src/main/java/io/debezium/server/batch/DebeziumMetrics.java
+++ b/debezium-server-batch-sinks/src/main/java/io/debezium/server/batch/DebeziumMetrics.java
@@ -89,4 +89,14 @@ public long streamingMilliSecondsBehindSource() {
     }
   }
 
+  public void logMetrics() {
+    LOGGER.info("Debezium Metrics:" +
+        " snapshotCompleted=" + this.snapshotCompleted() +
+        " snapshotRunning=" + this.snapshotRunning() +
+        " streamingQueueCurrentSize=" + this.streamingQueueCurrentSize() +
+        " streamingQueueRemainingCapacity=" + this.streamingQueueRemainingCapacity() +
+        " maxQueueSize=" + this.maxQueueSize +
+        " streamingMilliSecondsBehindSource=" + this.streamingMilliSecondsBehindSource()
+    );
+  }
 }
diff --git a/debezium-server-batch-sinks/src/main/java/io/debezium/server/batch/batchsizewait/MaxBatchSizeWait.java b/debezium-server-batch-sinks/src/main/java/io/debezium/server/batch/batchsizewait/MaxBatchSizeWait.java
index f8c87aef..704b4985 100644
--- a/debezium-server-batch-sinks/src/main/java/io/debezium/server/batch/batchsizewait/MaxBatchSizeWait.java
+++ b/debezium-server-batch-sinks/src/main/java/io/debezium/server/batch/batchsizewait/MaxBatchSizeWait.java
@@ -67,8 +67,10 @@ public void waitMs(Integer numRecordsProcessed, Integer processingTimeMs) throws
     final int streamingSecondsBehindSource = (int) (debeziumMetrics.streamingMilliSecondsBehindSource() / 1000);
     final boolean snapshotCompleted = debeziumMetrics.snapshotCompleted();
 
-    LOGGER.debug("Processed {}, QueueCurrentSize:{}, QueueTotalCapacity:{}, SecondsBehindSource:{}, SnapshotCompleted:{}",
-        numRecordsProcessed, streamingQueueCurrentSize, maxQueueSize, streamingSecondsBehindSource, snapshotCompleted
+    LOGGER.debug("Processed {}, QueueCurrentSize:{}, QueueTotalCapacity:{}, SecondsBehindSource:{}, " +
+            "SnapshotCompleted:{}, snapshotRunning:{}",
+        numRecordsProcessed, streamingQueueCurrentSize, maxQueueSize, streamingSecondsBehindSource, snapshotCompleted,
+        debeziumMetrics.snapshotRunning()
     );
 
     int totalWaitMs = 0;