Apache Storm ETL项目中Kafka 0.9.0读、3.1.0写的配置咨询

一、Kafka 3.1.0 Producer核心配置项 #

针对Storm ETL场景，写入Kafka 3.1.0需重点配置以下参数：

• bootstrap.servers：必填，传入Kafka集群地址，格式为host1:9092,host2:9092
• acks：建议设为all（或-1），确保消息写入所有同步副本后再确认，避免数据丢失；追求吞吐量可设为1
• retries：消息发送失败后的重试次数，建议设为3及以上，配合retry.backoff.ms（默认100ms）使用
• batch.size：批量发送的消息大小阈值，默认16384字节，可根据业务调至32768提升吞吐量
• linger.ms：发送前等待时长，默认0，设为5-10ms可让更多消息批量发送，提升传输效率
• key.serializer/value.serializer：指定为org.apache.kafka.common.serialization.StringSerializer（或匹配业务数据类型的序列化器）
• enable.idempotence：设为true开启幂等性，避免重复发送消息，适配ETL场景的数据一致性要求
• transactional.id：如需实现精确一次语义（Exactly-Once），需配置该参数，结合Storm事务拓扑使用

二、CPPKafkaBolt代码优化点 #

你的自定义Bolt存在几处适配Kafka 3.1.0及稳定性优化的点：

1. 修复配置合并逻辑 #

原代码未合并用户通过withProducerProperties传入的配置，导致自定义配置失效，修改如下：

@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
    // 保留兼容逻辑
    if(mapper == null) {
        this.mapper = new FieldNameBasedTupleToKafkaMapper<K,V>();
    }
    if(topicSelector == null) {
        this.topicSelector = new DefaultTopicSelector((String) stormConf.get(TOPIC));
    }

    // 初始化基础配置
    Properties props = new Properties();
    props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
    props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
    props.setProperty(ProducerConfig.ACKS_CONFIG, "all");
    
    // 合并用户自定义配置
    if (boltSpecfiedProperties != null) {
        props.putAll(boltSpecfiedProperties);
    }

    this.producer = new KafkaProducer<>(props);
    this.collector = collector;
    // 删除原代码中覆盖用户mapper的语句
}

2. 修复ProducerRecord参数硬编码问题 #

原代码写死了topic、key、message值，需替换为从Tuple中获取的真实数据：

// 错误写法
// Future<RecordMetadata> result = producer.send(new ProducerRecord("topic", "key", "message"), (Callback)callback);
// 修改为
Future<RecordMetadata> result = producer.send(new ProducerRecord<>(topic, key, message), callback);

3. 泛型与Producer类型一致性优化 #

将Producer改为泛型，匹配Bolt的<K, V>定义：

private Producer<K, V> producer;

// prepare方法中创建Producer时
this.producer = new KafkaProducer<>(props);

4. 资源关闭优化 #

在cleanup方法中指定关闭超时，确保未发送消息完成处理：

@Override
public void cleanup() {
    if (producer != null) {
        producer.close(Duration.ofSeconds(10)); // 等待10秒完成剩余消息发送
    }
}

5. Tick Tuple处理增强 #

增加日志便于调试：

if (TupleUtils.isTick(input)) {
    LOG.debug("收到Tick Tuple，跳过Kafka发送");
    collector.ack(input);
    return;
}

三、Kafka 0.9.0读取配置注意事项 #

从Kafka 0.9.0读取数据时，Storm Kafka Spout需使用兼容版本：

• 依赖使用org.apache.storm:storm-kafka:对应Storm版本，该版本兼容Kafka 0.9.x
• Spout配置需指定Kafka 0.9.0集群的bootstrap.servers、消费组group.id，以及auto.offset.reset（设为earliest或latest）

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