NiFi/Kafka中按message拆分多行文本SplitContent重复返回问题

SplitContent处理器默认按固定字节长度、单换行符规则拆分内容，原生不支持跨多行匹配自定义分隔符。直接用"message":作为分隔符时，处理器无法正确处理第二条记录的多行异常内容，读取指针没有正常向后偏移，反复读取同一段开头位置，才会重复输出第一行日志。

方案1：无代码配置流方案 #

用预处理+拆分的组合逻辑实现，不需要写代码：

1. 先接入ReplaceText处理器做内容归一化，配置项如下：
   • 搜索正则：("offset":\d+,\s*)?"message":
   • 替换内容：\n#UNIQUE_MSG_SPLIT_FLAG#\n
   • 替换范围：全文本替换
   • 开启多行匹配模式
     这一步会把所有日志头（无论是否携带offset字段）统一替换成日志内容里不可能出现的特殊分隔符，从根源避免误匹配。
2. 预处理完成后，接入SplitContent处理器：
   • 分隔符设置为#UNIQUE_MSG_SPLIT_FLAG#
   • 关闭"保留分隔符"选项
   • 开启跨换行匹配
     拆分后丢弃第一段空内容（第一个分隔符位于文本最前端，拆分后首段为空），剩余输出的两个FlowFile就是需要的两条独立记录，第二条会完整保留多行异常栈内容，和预期结果完全一致。

方案2：脚本处理方案（适配复杂边界场景） #

如果日志内容存在特殊字符容易导致正则误匹配，可以直接接入ExecuteScript处理器，引擎选择Groovy，粘贴以下脚本即可一次性完成拆分：

def flowFile = session.get()
if (!flowFile) return
def rawContent = flowFile.read().getText("UTF-8")
// 非贪婪匹配所有message分段，支持跨多行内容
def matcher = rawContent =~ /(?s)("offset":\d+,\s*)?"message":(.*?)(?=("offset":\d+,\s*)?"message":|$)/
def logSegments = []
matcher.each {
    def segment = it[2].trim()
    if (segment) logSegments << segment
}
logSegments.each { segment ->
    def newFlowFile = session.create(flowFile)
    newFlowFile = session.write(newFlowFile, { os ->
        os.write(segment.getBytes("UTF-8"))
    } as OutputStreamCallback)
    session.transfer(newFlowFile, REL_SUCCESS)
}
session.remove(flowFile)

该方案会自动识别所有日志段，不管异常栈有多少行都不会出现截断、重复输出的问题。

• 不要直接用"message":作为拆分分隔符，异常内容里本身存在Exception message这类带message关键词的文本，直接拆分容易把异常栈拆碎。
• 自定义拆分标记时，要选择业务日志里绝对不会出现的字符串，避免误拆分。

内容的提问来源于stack exchange，提问作者Newton