diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java
index 28131a8cbfe4..fed2187c8e19 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java
@@ -99,7 +99,6 @@ public long extractWatermark(IcebergSourceSplit split) {
                       && scanTask.file().lowerBounds().get(eventTimeFieldId) != null,
                   "Missing statistics for column name = %s in file = %s",
                   eventTimeFieldName,
-                  eventTimeFieldId,
                   scanTask.file());
               return timeUnit.toMillis(
                   Conversions.fromByteBuffer(
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractor.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractor.java
index a8872d8c5540..6c84363c9167 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractor.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractor.java
@@ -180,11 +180,14 @@ public void testEmptyStatistics() throws IOException {
     assumeThat(columnName).isEqualTo("timestamp_column");
 
     // Create an extractor for a column we do not have statistics
+    IcebergSourceSplit split = split(0);
     ColumnStatsWatermarkExtractor extractor =
         new ColumnStatsWatermarkExtractor(10, "missing_field");
-    assertThatThrownBy(() -> extractor.extractWatermark(split(0)))
+    assertThatThrownBy(() -> extractor.extractWatermark(split))
         .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("Missing statistics for column");
+        .hasMessage(
+            "Missing statistics for column name = missing_field in file = "
+                + split.task().files().iterator().next().file());
   }
 
   private IcebergSourceSplit split(int id) throws IOException {
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java
index 28131a8cbfe4..fed2187c8e19 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java
@@ -99,7 +99,6 @@ public long extractWatermark(IcebergSourceSplit split) {
                       && scanTask.file().lowerBounds().get(eventTimeFieldId) != null,
                   "Missing statistics for column name = %s in file = %s",
                   eventTimeFieldName,
-                  eventTimeFieldId,
                   scanTask.file());
               return timeUnit.toMillis(
                   Conversions.fromByteBuffer(
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractor.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractor.java
index a8872d8c5540..6c84363c9167 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractor.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractor.java
@@ -180,11 +180,14 @@ public void testEmptyStatistics() throws IOException {
     assumeThat(columnName).isEqualTo("timestamp_column");
 
     // Create an extractor for a column we do not have statistics
+    IcebergSourceSplit split = split(0);
     ColumnStatsWatermarkExtractor extractor =
         new ColumnStatsWatermarkExtractor(10, "missing_field");
-    assertThatThrownBy(() -> extractor.extractWatermark(split(0)))
+    assertThatThrownBy(() -> extractor.extractWatermark(split))
         .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("Missing statistics for column");
+        .hasMessage(
+            "Missing statistics for column name = missing_field in file = "
+                + split.task().files().iterator().next().file());
   }
 
   private IcebergSourceSplit split(int id) throws IOException {
diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java
index 28131a8cbfe4..fed2187c8e19 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java
+++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/reader/ColumnStatsWatermarkExtractor.java
@@ -99,7 +99,6 @@ public long extractWatermark(IcebergSourceSplit split) {
                       && scanTask.file().lowerBounds().get(eventTimeFieldId) != null,
                   "Missing statistics for column name = %s in file = %s",
                   eventTimeFieldName,
-                  eventTimeFieldId,
                   scanTask.file());
               return timeUnit.toMillis(
                   Conversions.fromByteBuffer(
diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractor.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractor.java
index a8872d8c5540..6c84363c9167 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractor.java
+++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractor.java
@@ -180,11 +180,14 @@ public void testEmptyStatistics() throws IOException {
     assumeThat(columnName).isEqualTo("timestamp_column");
 
     // Create an extractor for a column we do not have statistics
+    IcebergSourceSplit split = split(0);
     ColumnStatsWatermarkExtractor extractor =
         new ColumnStatsWatermarkExtractor(10, "missing_field");
-    assertThatThrownBy(() -> extractor.extractWatermark(split(0)))
+    assertThatThrownBy(() -> extractor.extractWatermark(split))
         .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("Missing statistics for column");
+        .hasMessage(
+            "Missing statistics for column name = missing_field in file = "
+                + split.task().files().iterator().next().file());
   }
 
   private IcebergSourceSplit split(int id) throws IOException {