Divolte Collector与Kafka事件路由异常排查：自定义事件重复投递至多个主题

你遇到的问题核心原因是两个映射（Mapping）都没有设置事件过滤规则。Divolte Collector的默认行为是：所有来自配置的sources（这里两个mapping都指定了[browser]）的事件，会被所有匹配该source的mapping处理。也就是说，不管你发送的是哪类事件，都会同时被a_mapping和b_mapping解析，然后各自推送到对应的Kafka sink，最终导致每个事件出现在两个主题中。

你需要为每个mapping指定专属的过滤条件，确保只有符合特定特征的事件才会被对应的mapping处理。具体有两种实现方式，选其一即可：

方式1：在Groovy映射脚本中添加过滤逻辑 #

修改你的Groovy映射文件，在mapping块开头加入filter语句，根据事件的唯一标识（比如事件类型、自定义参数）来过滤。

比如，假设你的第一个事件类型是js_error，第二个是user_behavior：

• 对于a_mapping的mapping.groovy：

mapping {
    // 只处理事件类型为js_error的事件
    filter eventType('js_error')
    // 如果是用自定义参数区分，比如event_type=type_a
    // filter eventParameter('event_type') == 'type_a'

    map eventParameter('js_error_msg') onto 'js_error_msg'
    map eventParameter('js_error_level') onto 'js_error_level'
    map eventParameter('js_url') onto 'js_url'
    map eventParameter('js_line') onto 'js_line'
    map eventParameter('js_column') onto 'js_column'
    map eventParameter('js_is_error_caught') onto 'js_is_error_caught'
}

• 对于b_mapping的mapping2.groovy：

mapping {
    // 只处理事件类型为user_behavior的事件
    filter eventType('user_behavior')
    // 或者 filter eventParameter('event_type') == 'type_b'

    // 这里添加对应MyEventRecord2.avsc的字段映射
}

方式2：在配置文件的Mapping块中添加过滤表达式 #

直接在divolte-collector.conf的mapping配置里添加event_filter参数，用Divolte的表达式语言指定过滤规则：

divolte { 
    global { 
        // 原有全局配置不变...
    } 
    mappings { 
        a_mapping = { 
            schema_file = "path-to/conf/MyEventRecord.avsc" 
            mapping_script_file = "path-to/conf/mapping.groovy" 
            sources = [browser] 
            sinks = [kafka1] 
            // 添加过滤：只处理事件类型为js_error的事件
            event_filter = "eventType('js_error')"
            // 自定义参数过滤示例：event_filter = "eventParameter('event_type') == 'type_a'"
        } 
        b_mapping = { 
            schema_file = "path-to/conf/MyEventRecord2.avsc" 
            mapping_script_file = "path-to/conf/mapping2.groovy" 
            sources = [browser] 
            sinks = [kafka2] 
            // 添加对应第二个事件的过滤规则
            event_filter = "eventType('user_behavior')"
        } 
    } 
    sinks { 
        // 原有sink配置不变...
    } 
}

• 确保你的事件确实带有用于区分的标识（比如事件类型、自定义参数），过滤规则要和实际事件的特征匹配。
• 修改配置/脚本后，重启Divolte Collector使变更生效。
• 分别发送两类事件，检查Kafka主题：对应事件应该只出现在其专属的主题中。

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