Flink批处理作业FlatMap算子出现Could not forward element to next operator错误求助

大家好，我现在在开发一个Flink批处理作业，流程是从Kafka读取JSON字符串，转换成Avro GenericRecord之后，用AvroParquetWriters写入Parquet文件。目前遇到了一个棘手的问题：明明JSON格式合法且和Avro Schema完全匹配，但FlatMap算子里总是抛出Could not forward element to next operator的错误，想请大家帮忙看看哪里出问题了。

先给大家贴一下我的相关代码和配置：

1. JSON消息格式 #

{"user_id":914,"message":"User login","timestamp":"2025-06-29"}

2. Avro Schema文件（log_schema.avsc） #

{
  "type": "record",
  "name": "UserLog",
  "namespace": "com.example",
  "fields": [
    { "name": "user_id", "type": "int" },
    { "name": "message", "type": "string" },
    { "name": "timestamp", "type": "string" }
  ]
}

3. FlatMap转换代码 #

.flatMap(new FlatMapFunction<String, GenericRecord>() {
    private final ObjectMapper mapper = new ObjectMapper();

    @Override
    public void flatMap(String json, Collector<GenericRecord> out) {
        try {
            JsonNode node = mapper.readTree(json);
            GenericRecord record = new GenericData.Record(schema);
            record.put("user_id", node.get("user_id").asInt());
            record.put("message", node.get("message").asText());
            record.put("timestamp", node.get("timestamp").asText());
            out.collect(record);
        } catch (Exception e) {
            System.err.println("Invalid JSON: " + json + " → " + e.getMessage());
        }
    }
})
.returns(TypeInformation.of(GenericRecord.class));

4. ParquetSink代码 #

FileSink<GenericRecord> sink = FileSink
    .forBulkFormat(new Path("file:///output/"), AvroParquetWriters.forGenericRecord(schema))
    .build();
parsedStream.sinkTo(sink);

现在的问题是，就算传入的JSON完全符合要求，FlatMap里还是会触发异常捕获，输出Invalid JSON: {"user_id":914,"message":"User login","timestamp":"2025-06-29"} → Could not forward element to next operator。我已经排查过JSON的格式、字段类型，都和Schema对应得上，实在找不到问题所在了。

我自己排查后的猜测和尝试 #

• 一开始以为是JSON解析的问题，但单独测试ObjectMapper解析这段JSON是完全正常的，能正确取出各个字段的值；
• 也检查了Schema的加载，确认schema对象是正确加载了log_schema.avsc的内容，没有空指针或者加载失败的情况；

求问大家有没有遇到过类似的问题？可能的原因是什么，该怎么解决？ #

内容来源于stack exchange