feat(client): support custom kafka consumer properties (SASL/SSL) & fix rollback NPE in flatMessage mode - #5617

背景 / Motivation

服务端（CanalKafkaProducer）和 client-adapter（CanalKafkaConsumer）都支持通过 kafka.* 前缀把任意配置透传给 Kafka 客户端，canal.properties 里也有 SASL 的示例：

kafka.sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required username="alice" password="alice-secret";
kafka.sasl.mechanism = SCRAM-SHA-512
kafka.security.protocol = SASL_PLAINTEXT

但 canal.client 模块的 KafkaCanalConnector 只有一个构造方法，consumer 配置在构造方法里写死，没有任何注入自定义配置的入口，导致 client 无法直接连接启用了 SASL / SSL 认证的 Kafka 集群（参考 #3066，只能自己改源码）。

Server side and client-adapter already pass kafka.* properties through to the Kafka client, but KafkaCanalConnector in canal.client hardcodes all consumer properties and offers no way to inject security.protocol / sasl.mechanism / sasl.jaas.config, so it cannot connect to a SASL/SSL secured cluster without patching the source.

改动 / Changes

1. KafkaCanalConnector / KafkaOffsetCanalConnector 新增重载构造方法，多接一个 Properties customProperties，在默认配置之后 putAll，可用于传 SASL / SSL 或其他任意 consumer 参数。原构造方法保留并委托给新方法，完全向后兼容。
   KafkaOffsetCanalConnector 仅在 customProperties 未指定 auto.offset.reset 时才写入默认的 earliest。

   Properties props = new Properties();
   props.put("security.protocol", "SASL_PLAINTEXT");
   props.put("sasl.mechanism", "SCRAM-SHA-512");
   props.put("sasl.jaas.config",
       "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"alice\" password=\"alice-secret\";");
   KafkaCanalConnector connector = new KafkaCanalConnector(servers, topic, null, groupId, 100, true, props);

2. 修复 KafkaCanalConnector.rollback() 在 flatMessage 模式下的 NPE：kafkaConsumer2 分支里 seek 的是 kafkaConsumer2，但 commitSync() 调的是 kafkaConsumer（flatMessage 模式下为 null）。
   Fix NPE in rollback(): the kafkaConsumer2 branch seeks on kafkaConsumer2 but committed on kafkaConsumer, which is null in flatMessage mode.

验证 / Verification

• mvn -pl client -am compile (JDK 8) 通过