title: Kafka consumer manual commit offset
body: <p>I'm implementing and dsl spring integration flow which takes messages from Kafka</p>
  
  <p>Code snippet:</p>
  
  <pre><code>return IntegrationFlows.from(
                  Kafka.messageDrivenChannelAdapter(new DefaultKafkaConsumerFactory(kafkaTelemetryDataConsumerConfiguration.getConsumerProperties()),
                          kafkaPropertiesConfiguration.getTelemetryDataTopic()))
                  })
                  .handle(bla.someImportantOperation())
                  //TODO:do manual commit here
                  //.handle(consumer.commitSync())
  
                  .get();
  </code></pre>
  
  <p>I would like to know how can I manually commitSync but only after <code>.handle(bla.someImportantOperation())</code> has successfuly finished.</p>
  
  <p>I don't know how to get the consumer reference since I'm using DefaultKafkaConsumerFactory, would appreciate any help.</p>
  
  <p>These are my consumerProperties I use to create consumer:</p>
  
  <pre><code>consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaPropertiesConfiguration.getBootstrapServers());
  consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
  
  consumerProperties.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, kafkaPropertiesConfiguration.getClientIdConfig());
  consumerProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, kafkaPropertiesConfiguration.getGroupIdConfig());
  
  consumerProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
  </code></pre>