迁移至Storm 2.2.0时Kafka偏移量不生效消费重置问题咨询

• 偏移量存储逻辑变更
  Storm 0.9.x版本的旧Kafka Spout默认将消费偏移量存储在Kafka服务端，而Storm 2.x的KafkaTridentSpoutOpaque属于事务型Spout，默认将偏移量存储在Trident状态后端，不会主动向Kafka服务端提交消费位移，因此你在Kafka侧查询到的消费组偏移量从未更新，无法作为重启后的消费恢复依据。
• 本地集群状态未持久化
  Storm本地测试集群默认使用内存型Trident状态后端，集群重启后所有状态数据会被清空，Spout找不到已消费的偏移量记录时，会按照默认策略从头开始拉取消息。
• 缺少必要的消费策略配置
  代码中未显式配置初始消费位点策略、偏移量提交规则，Spout会使用默认值执行消费逻辑，进一步加重了重启位点丢失的问题。

本地测试场景（优先使用Kafka存储偏移量） #

在原有配置中补充以下参数即可：

KafkaTridentSpoutConfig.Builder kafkaSpoutConfigBuilder = KafkaTridentSpoutConfig.builder(bootstrapServers, topic);

kafkaSpoutConfigBuilder.setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, fetchSizeBytes);
kafkaSpoutConfigBuilder.setProp(ConsumerConfig.GROUP_ID_CONFIG, clientId);
kafkaSpoutConfigBuilder.setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
kafkaSpoutConfigBuilder.setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
// 新增配置开始
kafkaSpoutConfigBuilder.setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
kafkaSpoutConfigBuilder.setProp(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
kafkaSpoutConfigBuilder.setFirstPollOffsetStrategy(FirstPollOffsetStrategy.COMMITTED_LATEST);
kafkaSpoutConfigBuilder.setOffsetCommitPeriodMs(5000);
// 新增配置结束

return new KafkaTridentSpoutOpaque(kafkaSpoutConfigBuilder.build());

生产环境场景（保留事务特性） #

如果需要保留KafkaTridentSpoutOpaque的事务恰好一次语义，不要修改偏移量存储到Kafka，改为配置持久化Trident状态后端即可，可选方案包括ZooKeeper、HBase等持久化存储，保证Storm集群重启后偏移量数据不会丢失。

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