Kafka调试技巧及心得分享

@KafkaListener(topicPartitions = {@TopicPartition(topic = "yourTopic", partitionOffsets = {@PartitionOffset(partition = "指定分区", initialOffset = "初始偏移量")})})
public void forlanConsumer(ConsumerRecord<String, String> record) {
	String messageStr = record.value();
	log.info(“测试触达记录消费:offset = {}, res = {}”,record.offset(), messageStr);
}

@KafkaListener(topicPartitions = {@TopicPartition(topic = "yourTopic", partitionOffsets = {@PartitionOffset(partition = "指定分区", initialOffset = "初始偏移量")})})
public void forlanConsumer(ConsumerRecord<String, String> record) {
	if (record.offset() > 结束偏移量) return;
	String messageStr = record.value();
	log.info(“测试触达记录消费:offset = {}, res = {}”,record.offset(), messageStr);
}

@Autowired
private KafkaConfig kafkaConfig;

public void reconsumeMessage(String topic, int partition, long offset) {
		ConsumerFactory<Integer, String> consumerFactory = kafkaConfig.consumerFactory();
		Map<String, Object> configurationProperties = consumerFactory.getConfigurationProperties();

		Map<String, Object> customProps = new HashMap<>();
		customProps.put(ConsumerConfig.GROUP_ID_CONFIG, "reconsume-temp-group-" + UUID.randomUUID());
		customProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");// 防止 offset 不存在时报错
		customProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
		customProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");// 控制每次 poll 只拉取一条
		// 复用 kafkaConfig 中的基础配置
		Map<String, Object> props = new HashMap<>(configurationProperties);
		props.putAll(customProps);

		try (Consumer<String, String> consumer = new KafkaConsumer<>(props)) {
			TopicPartition topicPartition = new TopicPartition(topic, partition);

			// 分配分区并定位偏移量
			consumer.assign(Collections.singletonList(topicPartition));
			consumer.seek(topicPartition, offset);

			// 拉取消息（设置超时时间）
			ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));

			for (ConsumerRecord<String, String> record : records) {
				if (record.offset() == offset) {
					if (Objects.equals(topic, KafkaTopic.Forlan_MESSAGE_NOTIFY)) {
						// 调用@KafkaListener的方法执行逻辑
						forlanConsumer.userConsumer(record);
					}
					break;
				}
			}
		}
	}

private Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap();
    props.put("bootstrap.servers", this.bootstrapServers);
    props.put("group.id", this.groupid);
    props.put("enable.auto.commit", this.autoCommit);
    props.put("auto.commit.interval.ms", this.interval);
    props.put("session.timeout.ms", this.timeout);
    props.put("key.deserializer", this.keyDeserializer);
    props.put("value.deserializer", this.valueDeserializer);
    props.put("auto.offset.reset", this.offsetReset);
    props.put("max.poll.records", this.maxPollRecords);
    return props;
}