title: Spring Integration Kafka Manual Acknowledgment
body: <p>I'm having issues using manual acknowledgements with the KafkaTopicOffsetManager. When acknowledge() is called, the topic begins to get spammed repeatedly. Kafka has log.cleaner.enable set to true and the topic is using cleanup.policy=compact. Thanks for any help.</p>
  
  <p>Config:</p>
  
  <pre><code>@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;
  }
  </code></pre>
  
  <p>Listener:</p>
  
  <pre><code>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();
      }
  }
  </code></pre>