From 8e9c67d415a668886455df3fc7e62d4c4c590454 Mon Sep 17 00:00:00 2001
From: Mingliang Liu <liuml07@apache.org>
Date: Tue, 20 Jun 2023 15:16:04 -0700
Subject: [PATCH] [Connectors] Add table information to IcebergCommitter

---
 .../sink/committer/IcebergCommitterStage.java        | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

diff --git a/mantis-connectors/mantis-connector-iceberg/src/main/java/io/mantisrx/connector/iceberg/sink/committer/IcebergCommitterStage.java b/mantis-connectors/mantis-connector-iceberg/src/main/java/io/mantisrx/connector/iceberg/sink/committer/IcebergCommitterStage.java
index ccde0172f..fb2dbb84b 100644
--- a/mantis-connectors/mantis-connector-iceberg/src/main/java/io/mantisrx/connector/iceberg/sink/committer/IcebergCommitterStage.java
+++ b/mantis-connectors/mantis-connector-iceberg/src/main/java/io/mantisrx/connector/iceberg/sink/committer/IcebergCommitterStage.java
@@ -178,14 +178,22 @@ public Observable<Map<String, Object>> call(Observable<MantisDataFile> source) {
                             return summary;
                         } catch (RuntimeException e) {
                             metrics.increment(CommitterMetrics.COMMIT_FAILURE_COUNT);
-                            logger.error("error committing to Iceberg", e);
+                            logger.error("error committing to Iceberg table {}.{}.{}",
+                                config.getCatalog(),
+                                config.getDatabase(),
+                                config.getTable(),
+                                e);
                             return new HashMap<String, Object>();
                         }
                     })
                     .filter(summary -> !summary.isEmpty())
                     .doOnNext(summary -> {
                         metrics.increment(CommitterMetrics.COMMIT_SUCCESS_COUNT);
-                        logger.info("committed {}", summary);
+                        logger.info("committed to table {}.{}.{} with summary: {}",
+                            config.getCatalog(),
+                            config.getDatabase(),
+                            config.getTable(),
+                            summary);
                     });
         }
     }