能否在Kafka Connect值转换器（反序列化阶段）丢弃记录？

问题解决思路 #

你需要在反序列化阶段就跳过无需处理的消息，避免Protobuf反序列化的额外开销，之前遇到的返回null导致消费者崩溃、后置过滤仍有开销的问题，可以通过以下方案解决：

最优简便方案 #

在自定义值转换器中直接做判断分支处理：

• 当消息头符合处理条件时，正常执行Protobuf反序列化；
• 当消息不需要处理时，返回ByteArrayConverter的转换结果（即原字节数组），完全跳过Protobuf反序列化步骤。

示例代码（Java） #

public class CustomProtobufValueConverter implements ValueConverter {
    private final ByteArrayConverter byteArrayConverter = new ByteArrayConverter();
    private final ProtobufConverter protobufConverter = new ProtobufConverter();

    @Override
    public SchemaAndValue convert(Object input, Schema schema, LogicalTypeRegistry registry) {
        // 根据消息头判断是否需要处理（需结合实际场景获取Kafka消息头）
        boolean shouldProcess = shouldProcessMessage();
        
        if (shouldProcess) {
            // 需要处理的消息，执行正常Protobuf反序列化
            return protobufConverter.convert(input, schema, registry);
        } else {
            // 无需处理的消息，直接返回原字节数组，跳过反序列化
            return byteArrayConverter.convert(input, schema, registry);
        }
    }

    // 自定义消息头校验逻辑
    private boolean shouldProcessMessage() {
        // 示例：检查消息头中"process-type"字段是否为"need-handle"
        Headers headers = getCurrentMessageHeaders(); // 从上下文获取当前消息的headers
        Header typeHeader = headers.lastHeader("process-type");
        return typeHeader != null && "need-handle".equals(new String(typeHeader.value()));
    }

    // 辅助方法：根据使用的消费者框架获取当前消息头（如Spring Kafka、原生Kafka Consumer）
    private Headers getCurrentMessageHeaders() {
        // 此处根据实际框架实现，比如Spring中通过上下文、原生Consumer从ConsumerRecord中获取
        return ...;
    }
}

后续处理说明 #

返回原字节数组后，下游业务逻辑可通过判断SchemaAndValue的实际类型（比如是否为字节数组）直接跳过处理，彻底避免不必要的反序列化和业务逻辑开销，比返回空SchemaAndValue更高效。

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