danzhewuju · Oct 29, 2026
diff --git a/docs/content.zh/docs/connectors/table/formats/json.md b/docs/content.zh/docs/connectors/table/formats/json.md
@@ -247,6 +247,13 @@ Format 参数
 特性
 --------
 
+### 被忽略的解析错误指标
+
+当 `json.ignore-parse-errors` 为 `true` 时，JSON 格式会在连接器提供的反序列化指标组下注册计数器
+`json.numMessagesWithParseErrors`。该指标统计至少有一个解析或字段转换错误被忽略的输入消息数，
+包括仍能输出数据、但部分字段被置为 null 的消息。最外层 JSON Array 按一条消息计数。
+`debezium-json`、`canal-json` 等 CDC 格式不上报该指标。
+
 ### 允许 json array 直接展开成多行数据
 
 通常，我们假设 JSON 的最外层数据是一个 JSON Object。所以一条 JSON 会转换成一行结果。

diff --git a/docs/content/docs/connectors/table/formats/json.md b/docs/content/docs/connectors/table/formats/json.md
@@ -260,6 +260,14 @@ The following table lists the type mapping from Flink type to JSON type.
 Features
 --------
 
+### Ignored parse error metric
+
+When `json.ignore-parse-errors` is `true`, the JSON format registers the counter
+`json.numMessagesWithParseErrors` under the connector's deserialization metric group.
+It counts input messages in which at least one parse or field conversion error was ignored,
+including messages that still produce rows with null fields. A top-level JSON array counts
+as one message. CDC formats such as `debezium-json` and `canal-json` do not report this metric.
+
 ### Allow top-level JSON Arrays
 
 Usually, we assume the top-level of JSON string is a stringified JSON object. Then this stringified JSON object can be converted into one SQL row.

diff --git a/...k-json/src/main/java/org/apache/flink/formats/json/AbstractJsonDeserializationSchema.java b/...k-json/src/main/java/org/apache/flink/formats/json/AbstractJsonDeserializationSchema.java
@@ -22,6 +22,7 @@
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.metrics.Counter;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.logical.DecimalType;
 import org.apache.flink.table.types.logical.RowType;
@@ -80,6 +81,12 @@ public abstract class AbstractJsonDeserializationSchema implements Deserializati
 
     private transient List<RowData> reusableCollectList;
 
+    private boolean parseErrorMetricsEnabled;
+
+    private transient Counter numMessagesWithParseErrors;
+
+    private transient boolean messageHasParseError;
+
     public AbstractJsonDeserializationSchema(
             RowType rowType,
             TypeInformation<RowData> resultTypeInfo,
@@ -109,6 +116,31 @@ public void open(InitializationContext context) throws Exception {
         }
         reusableCollectList = new ArrayList<>();
         collector = new ListCollector<>(reusableCollectList);
+        if (parseErrorMetricsEnabled && ignoreParseErrors) {
+            numMessagesWithParseErrors =
+                    context.getMetricGroup().addGroup("json").counter("numMessagesWithParseErrors");
+            LOG.info("Registered metric json.numMessagesWithParseErrors for ignored parse errors.");
+        }
+    }
+
+    // CDC formats own their message boundaries and opt in separately.
+    void enableParseErrorMetrics() {
+        parseErrorMetricsEnabled = true;
+    }
+
+    protected final Runnable getParseErrorListener() {
+        return this::markParseError;
+    }
+
+    protected final void markParseError() {
+        messageHasParseError = true;
+    }
+
+    protected final void finishMessage() {
+        if (messageHasParseError && numMessagesWithParseErrors != null) {
+            numMessagesWithParseErrors.inc();
+        }
+        messageHasParseError = false;
     }
 
     /**
@@ -153,13 +185,19 @@ public boolean equals(Object o) {
         AbstractJsonDeserializationSchema that = (AbstractJsonDeserializationSchema) o;
         return failOnMissingField == that.failOnMissingField
                 && ignoreParseErrors == that.ignoreParseErrors
+                && parseErrorMetricsEnabled == that.parseErrorMetricsEnabled
                 && resultTypeInfo.equals(that.resultTypeInfo)
                 && timestampFormat.equals(that.timestampFormat);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(failOnMissingField, ignoreParseErrors, resultTypeInfo, timestampFormat);
+        return Objects.hash(
+                failOnMissingField,
+                ignoreParseErrors,
+                parseErrorMetricsEnabled,
+                resultTypeInfo,
+                timestampFormat);
     }
 
     /**

diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java
@@ -84,23 +84,29 @@ public DeserializationSchema<RowData> createRuntimeDecoder(
                 final RowType rowType = (RowType) producedDataType.getLogicalType();
                 final TypeInformation<RowData> rowDataTypeInfo =
                         context.createTypeInformation(producedDataType);
+                final AbstractJsonDeserializationSchema schema;
                 if (jsonParserEnabled) {
-                    return new JsonParserRowDataDeserializationSchema(
-                            rowType,
-                            rowDataTypeInfo,
-                            failOnMissingField,
-                            ignoreParseErrors,
-                            timestampOption,
-                            toProjectedNames(
-                                    (RowType) physicalDataType.getLogicalType(), projections));
+                    schema =
+                            new JsonParserRowDataDeserializationSchema(
+                                    rowType,
+                                    rowDataTypeInfo,
+                                    failOnMissingField,
+                                    ignoreParseErrors,
+                                    timestampOption,
+                                    toProjectedNames(
+                                            (RowType) physicalDataType.getLogicalType(),
+                                            projections));
                 } else {
-                    return new JsonRowDataDeserializationSchema(
-                            rowType,
-                            rowDataTypeInfo,
-                            failOnMissingField,
-                            ignoreParseErrors,
-                            timestampOption);
+                    schema =
+                            new JsonRowDataDeserializationSchema(
+                                    rowType,
+                                    rowDataTypeInfo,
+                                    failOnMissingField,
+                                    ignoreParseErrors,
+                                    timestampOption);
                 }
+                schema.enableParseErrorMetrics();
+                return schema;
             }
 
             @Override

diff --git a/...n/src/main/java/org/apache/flink/formats/json/JsonParserRowDataDeserializationSchema.java b/...n/src/main/java/org/apache/flink/formats/json/JsonParserRowDataDeserializationSchema.java
@@ -43,6 +43,8 @@
 @Internal
 public class JsonParserRowDataDeserializationSchema extends AbstractJsonDeserializationSchema {
 
+    private final JsonParserToRowDataConverters converters;
+
     /**
      * Runtime converter that converts {@link JsonParser}s into objects of Flink SQL internal data
      * structures.
@@ -66,10 +68,16 @@ public JsonParserRowDataDeserializationSchema(
             TimestampFormat timestampFormat,
             @Nullable String[][] projectedFields) {
         super(rowType, resultTypeInfo, failOnMissingField, ignoreParseErrors, timestampFormat);
-        this.runtimeConverter =
+        this.converters =
                 new JsonParserToRowDataConverters(
-                                failOnMissingField, ignoreParseErrors, timestampFormat)
-                        .createConverter(projectedFields, checkNotNull(rowType));
+                        failOnMissingField, ignoreParseErrors, timestampFormat);
+        this.runtimeConverter = converters.createConverter(projectedFields, checkNotNull(rowType));
+    }
+
+    @Override
+    public void open(InitializationContext context) throws Exception {
+        super.open(context);
+        converters.setParseErrorListener(getParseErrorListener());
     }
 
     @Override
@@ -100,7 +108,10 @@ public void deserialize(byte[] message, Collector<RowData> out) throws IOExcepti
                 throw new IOException(
                         format("Failed to deserialize JSON '%s'.", new String(message)), t);
             }
+            markParseError();
             logParseErrorIfDebugEnabled(message, t);
+        } finally {
+            finishMessage();
         }
     }
 

diff --git a/...flink-json/src/main/java/org/apache/flink/formats/json/JsonParserToRowDataConverters.java b/...flink-json/src/main/java/org/apache/flink/formats/json/JsonParserToRowDataConverters.java
@@ -67,6 +67,7 @@
 import static org.apache.flink.formats.common.TimeFormats.SQL_TIMESTAMP_FORMAT;
 import static org.apache.flink.formats.common.TimeFormats.SQL_TIMESTAMP_WITH_LOCAL_TIMEZONE_FORMAT;
 import static org.apache.flink.formats.common.TimeFormats.SQL_TIME_FORMAT;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /** Tool class used to convert fields from {@link JsonParser} to {@link RowData}. */
 @Internal
@@ -83,6 +84,9 @@ public class JsonParserToRowDataConverters implements Serializable {
     /** Timestamp format specification which is used to parse timestamp. */
     private final TimestampFormat timestampFormat;
 
+    /** Notified when a field conversion error is ignored and the field is set to null. */
+    private transient Runnable parseErrorListener = () -> {};
+
     public JsonParserToRowDataConverters(
             boolean failOnMissingField,
             boolean ignoreParseErrors,
@@ -92,6 +96,10 @@ public JsonParserToRowDataConverters(
         this.timestampFormat = timestampFormat;
     }
 
+    void setParseErrorListener(Runnable parseErrorListener) {
+        this.parseErrorListener = checkNotNull(parseErrorListener);
+    }
+
     /**
      * Runtime converter that converts {@link JsonParser}s into objects of Flink Table & SQL
      * internal data structures. Unlike {@link JsonToRowDataConverters.JsonToRowDataConverter}, this
@@ -473,6 +481,7 @@ private JsonParserToRowDataConverter wrapIntoNullableConverter(
                 if (!ignoreParseErrors) {
                     throw t;
                 }
+                parseErrorListener.run();
                 return null;
             }
         };

diff --git a/...nk-json/src/main/java/org/apache/flink/formats/json/JsonRowDataDeserializationSchema.java b/...nk-json/src/main/java/org/apache/flink/formats/json/JsonRowDataDeserializationSchema.java
@@ -46,6 +46,8 @@
 public class JsonRowDataDeserializationSchema extends AbstractJsonDeserializationSchema {
     private static final long serialVersionUID = 1L;
 
+    private final JsonToRowDataConverters converters;
+
     /**
      * Runtime converter that converts {@link JsonNode}s into objects of Flink SQL internal data
      * structures.
@@ -59,9 +61,15 @@ public JsonRowDataDeserializationSchema(
             boolean ignoreParseErrors,
             TimestampFormat timestampFormat) {
         super(rowType, resultTypeInfo, failOnMissingField, ignoreParseErrors, timestampFormat);
-        this.runtimeConverter =
-                new JsonToRowDataConverters(failOnMissingField, ignoreParseErrors, timestampFormat)
-                        .createConverter(checkNotNull(rowType));
+        this.converters =
+                new JsonToRowDataConverters(failOnMissingField, ignoreParseErrors, timestampFormat);
+        this.runtimeConverter = converters.createConverter(checkNotNull(rowType));
+    }
+
+    @Override
+    public void open(InitializationContext context) throws Exception {
+        super.open(context);
+        converters.setParseErrorListener(getParseErrorListener());
     }
 
     @Override
@@ -84,6 +92,7 @@ public void deserialize(@Nullable byte[] message, Collector<RowData> out) throws
                             // will be caught by outer try-catch
                             throw t;
                         }
+                        markParseError();
                         logParseErrorIfDebugEnabled(message, t);
                     }
                 }
@@ -98,7 +107,10 @@ public void deserialize(@Nullable byte[] message, Collector<RowData> out) throws
                 throw new IOException(
                         format("Failed to deserialize JSON '%s'.", new String(message)), t);
             }
+            markParseError();
             logParseErrorIfDebugEnabled(message, t);
+        } finally {
+            finishMessage();
         }
     }
 

diff --git a/...rmats/flink-json/src/main/java/org/apache/flink/formats/json/JsonToRowDataConverters.java b/...rmats/flink-json/src/main/java/org/apache/flink/formats/json/JsonToRowDataConverters.java
@@ -65,6 +65,7 @@
 import static org.apache.flink.formats.common.TimeFormats.SQL_TIMESTAMP_FORMAT;
 import static org.apache.flink.formats.common.TimeFormats.SQL_TIMESTAMP_WITH_LOCAL_TIMEZONE_FORMAT;
 import static org.apache.flink.formats.common.TimeFormats.SQL_TIME_FORMAT;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /** Tool class used to convert from {@link JsonNode} to {@link RowData}. * */
 @Internal
@@ -81,6 +82,9 @@ public class JsonToRowDataConverters implements Serializable {
     /** Timestamp format specification which is used to parse timestamp. */
     private final TimestampFormat timestampFormat;
 
+    /** Notified when a field conversion error is ignored and the field is set to null. */
+    private transient Runnable parseErrorListener = () -> {};
+
     public JsonToRowDataConverters(
             boolean failOnMissingField,
             boolean ignoreParseErrors,
@@ -90,6 +94,10 @@ public JsonToRowDataConverters(
         this.timestampFormat = timestampFormat;
     }
 
+    void setParseErrorListener(Runnable parseErrorListener) {
+        this.parseErrorListener = checkNotNull(parseErrorListener);
+    }
+
     /**
      * Runtime converter that converts {@link JsonNode}s into objects of Flink Table & SQL internal
      * data structures.
@@ -394,6 +402,7 @@ private JsonToRowDataConverter wrapIntoNullableConverter(JsonToRowDataConverter
                 if (!ignoreParseErrors) {
                     throw t;
                 }
+                parseErrorListener.run();
                 return null;
             }
         };

diff --git a/...formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonFormatFactoryTest.java b/...formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonFormatFactoryTest.java
@@ -165,6 +165,7 @@ private void testSchemaDeserializationSchema(Map<String, String> options) {
                         false,
                         true,
                         TimestampFormat.ISO_8601);
+        expectedDeser.enableParseErrorMetrics();
 
         DeserializationSchema<RowData> actualDeser =
                 createTableSource(options)
@@ -252,7 +253,7 @@ private void testJsonParserConfiguration(boolean enabled, Class<?> expectedClass
                                 ScanRuntimeProviderContext.INSTANCE,
                                 SCHEMA.toPhysicalRowDataType());
 
-        DeserializationSchema<RowData> expectedDeser =
+        AbstractJsonDeserializationSchema expectedDeser =
                 enabled
                         ? new JsonParserRowDataDeserializationSchema(
                                 PHYSICAL_TYPE,
@@ -266,6 +267,7 @@ private void testJsonParserConfiguration(boolean enabled, Class<?> expectedClass
                                 false,
                                 true,
                                 TimestampFormat.ISO_8601);
+        expectedDeser.enableParseErrorMetrics();
 
         assertThat(actualDeser).isInstanceOf(expectedClass);
         assertThat(actualDeser).isEqualTo(expectedDeser);