From a44f0b646ad2e5b93b2a1cf99ec6feafc3419572 Mon Sep 17 00:00:00 2001
From: Jeff Xiang <jxiang@pinterest.com>
Date: Thu, 31 Oct 2024 18:24:17 -0400
Subject: [PATCH] Add more debug logs to see why producerPool doesn't get added

---
 .../com/pinterest/flink/connector/psc/sink/PscWriter.java    | 5 ++++-
 .../flink/connector/psc/sink/TransactionAborter.java         | 1 +
 .../main/java/com/pinterest/psc/consumer/PscConsumer.java    | 2 +-
 3 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/psc-flink/src/main/java/com/pinterest/flink/connector/psc/sink/PscWriter.java b/psc-flink/src/main/java/com/pinterest/flink/connector/psc/sink/PscWriter.java
index 0cff978..08d1648 100644
--- a/psc-flink/src/main/java/com/pinterest/flink/connector/psc/sink/PscWriter.java
+++ b/psc-flink/src/main/java/com/pinterest/flink/connector/psc/sink/PscWriter.java
@@ -182,6 +182,7 @@ class PscWriter<IN>
                     checkNotNull(recoveredStates, "recoveredStates"), lastCheckpointId + 1);
             this.currentProducer = getTransactionalProducer(lastCheckpointId + 1);
             this.currentProducer.beginTransaction();
+            LOG.info("producerPool.size(): " + producerPool.size());
         } else if (deliveryGuarantee == DeliveryGuarantee.AT_LEAST_ONCE
                 || deliveryGuarantee == DeliveryGuarantee.NONE) {
             this.currentProducer = new FlinkPscInternalProducer<>(this.pscProducerConfig, null);
@@ -252,7 +253,7 @@ public List<PscWriterState> snapshotState(long checkpointId) throws IOException
     @Override
     public void close() throws Exception {
         closed = true;
-        LOG.debug("Closing writer with {}", currentProducer);
+        LOG.info("Closing writer with {}", currentProducer);
         closeAll(
                 this::abortCurrentProducer,
                 closer,
@@ -344,6 +345,8 @@ private FlinkPscInternalProducer<byte[], byte[]> getTransactionalProducer(long c
 
     private FlinkPscInternalProducer<byte[], byte[]> getOrCreateTransactionalProducer(
             String transactionalId) {
+        LOG.info("producerPool.size() in getOrCreateTransactionalProducer: " + producerPool.size());
+        LOG.info("producerPool: " + producerPool);
         FlinkPscInternalProducer<byte[], byte[]> producer = producerPool.poll();
         try {
             if (producer == null) {
diff --git a/psc-flink/src/main/java/com/pinterest/flink/connector/psc/sink/TransactionAborter.java b/psc-flink/src/main/java/com/pinterest/flink/connector/psc/sink/TransactionAborter.java
index 266ec5f..aa506a7 100644
--- a/psc-flink/src/main/java/com/pinterest/flink/connector/psc/sink/TransactionAborter.java
+++ b/psc-flink/src/main/java/com/pinterest/flink/connector/psc/sink/TransactionAborter.java
@@ -132,6 +132,7 @@ private int abortTransactionOfSubtask(String prefix, long startCheckpointId, int
 
     public void close() {
         LOG.info("Closing transaction aborter");
+        Thread.dumpStack();
         if (producer != null) {
             LOG.info("CloseAction: {}", closeAction);
             closeAction.accept(producer);
diff --git a/psc/src/main/java/com/pinterest/psc/consumer/PscConsumer.java b/psc/src/main/java/com/pinterest/psc/consumer/PscConsumer.java
index 0b0b584..d781d92 100644
--- a/psc/src/main/java/com/pinterest/psc/consumer/PscConsumer.java
+++ b/psc/src/main/java/com/pinterest/psc/consumer/PscConsumer.java
@@ -1844,7 +1844,7 @@ public Map<TopicUriPartition, MessageId> getMessageIdByTimestamp(Map<TopicUriPar
      */
     public Map<MetricName, Metric> metrics() throws ClientException {
         ensureOpen();
-        Map<MetricName, Metric> metrics = new HashMap<>();
+        Map<MetricName, Metric> metrics = new ConcurrentHashMap<>();
         for (PscBackendConsumer<K, V> backendConsumer : backendConsumers) {
             metrics.putAll(backendConsumer.metrics());
         }