Add logs to debug double consumption during checkpoint recovery from …

…native kafka
pinterest · Jan 13, 2025 · 7d20d6c · 7d20d6c
1 parent 3b00c3d
commit 7d20d6c
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/...a/com/pinterest/flink/connector/psc/source/split/PscTopicUriPartitionSplitSerializer.java b/...a/com/pinterest/flink/connector/psc/source/split/PscTopicUriPartitionSplitSerializer.java
@@ -63,13 +63,15 @@ public byte[] serialize(PscTopicUriPartitionSplit split) throws IOException {
             out.writeLong(split.getStartingOffset());
             out.writeLong(split.getStoppingOffset().orElse(PscTopicUriPartitionSplit.NO_STOPPING_OFFSET));
             out.flush();
+            LOG.info("Serializing split with version: " + getVersion() + " for topicUri-partition: " + split.getTopicUri() + "-" + split.getPartition());
             return baos.toByteArray();
         }
     }
 
     @Override
     public PscTopicUriPartitionSplit deserialize(int version, byte[] serialized) throws IOException {
         LOG.info("Deserializing split with version: " + version);
+        Thread.dumpStack();
         try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
                 DataInputStream in = new DataInputStream(bais)) {
             String topicUri = in.readUTF();
@@ -90,6 +92,7 @@ public PscTopicUriPartitionSplit deserialize(int version, byte[] serialized) thr
             int partition = in.readInt();
             long offset = in.readLong();
             long stoppingOffset = in.readLong();
+            LOG.info("Deserialized split for topicUri-partition: " + topicUri + "-" + partition + " for offset=" + offset + " and stopping offset=" + stoppingOffset);
             return new PscTopicUriPartitionSplit(
                     new TopicUriPartition(topicUri, partition), offset, stoppingOffset);
         }