Add log line for committing/retrieving watermarks in streaming (apach…

…e#3578)
phet · Dec 3, 2022 · 3102ff1 · 3102ff1
1 parent 07a7303
commit 3102ff1
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 0 deletions.
diff --git a/...n-metastore/src/main/java/org/apache/gobblin/runtime/StateStoreBasedWatermarkStorage.java b/...n-metastore/src/main/java/org/apache/gobblin/runtime/StateStoreBasedWatermarkStorage.java
@@ -115,6 +115,7 @@ public void commitWatermarks(Iterable<CheckpointableWatermark> watermarks)
     for (CheckpointableWatermark watermark: watermarks) {
       String tableName = watermark.getSource();
       _stateStore.put(_storeName, tableName, new CheckpointableWatermarkState(watermark, GSON));
+      log.info("Committed watermark {} for table {}", watermark.getWatermark().toString(), tableName);
     }
   }
 

diff --git a/.../main/java/org/apache/gobblin/source/extractor/extract/kafka/KafkaStreamingExtractor.java b/.../main/java/org/apache/gobblin/source/extractor/extract/kafka/KafkaStreamingExtractor.java
@@ -288,6 +288,9 @@ private Map<KafkaPartition, LongWatermark> getTopicPartitionWatermarks(List<Kafk
         longWatermarkMap.put(topicPartition, new LongWatermark(0L));
       }
     }
+    for (Map.Entry<KafkaPartition, LongWatermark> entry : longWatermarkMap.entrySet()) {
+      log.info("Retrieved watermark {} for partition {}", entry.getValue().toString(), entry.getKey().toString());
+    }
     return longWatermarkMap;
   }