Spring Integration Kafka手册致谢

Question

我在使用手动确认与Kafka主题关闭管理器时遇到问题。当调用conifate（）时，主题开始反复收到垃圾邮件。Kafka 已将日志.cleaner.启用设置为 true，并且主题正在使用清理。感谢您的任何帮助。

配置：

@Bean
public ZookeeperConfiguration zookeeperConfiguration() {
    ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration(kafkaConfig.getZookeeperAddress());
    zookeeperConfiguration.setClientId("clientId");
    return zookeeperConfiguration;
}

@Bean
public ConnectionFactory connectionFactory() {
    return new DefaultConnectionFactory(zookeeperConfiguration());
}

@Bean
public TestMessageHandler messageListener() {
    return new TestMessageHandler();
}

@Bean
public OffsetManager offsetManager() {
    ZookeeperConnect zookeeperConnect = new ZookeeperConnect(kafkaConfig.getZookeeperAddress());
    OffsetManager offsetManager = new KafkaTopicOffsetManager(zookeeperConnect, kafkaConfig.getTopic() + "_OFFSET");

    return offsetManager;
}

@Bean
public KafkaMessageListenerContainer kafkaMessageListenerContainer() {
    KafkaMessageListenerContainer kafkaMessageListenerContainer = new KafkaMessageListenerContainer(connectionFactory(), kafkaConfig.getTopic());
    kafkaMessageListenerContainer.setMessageListener(messageListener());
    kafkaMessageListenerContainer.setOffsetManager(offsetManager());
    return kafkaMessageListenerContainer;
}

听者：

public class TestMessageHandler implements AcknowledgingMessageListener {

    private static final Logger logger = LoggerFactory.getLogger(TestMessageHandler.class);

    @Override
    public void onMessage(KafkaMessage message, Acknowledgment acknowledgment) {
        logger.info(message.toString());

        acknowledgment.acknowledge();
    }
}

Answer 1

KafkaTopicOffsetManager需要它自己的主题来维护正在使用的实际主题的偏移量。