我们正试图用下面的代码用springdsl-kafka实现手动提交。我们找不到它的任何参考资料。
我们可以将consumer porperty添加为“auto.commit”或“false”，但我们希望在成功处理消息后提交消息。有人能帮忙吗？

@Bean
        IntegrationFlow consumer()  {
    KafkaHighLevelConsumerMessageSourceSpec messageSourceSpec = Kafka
                    .inboundChannelAdapter(new ZookeeperConnect(this.kafkaConfig.getZookeeperAddress()))
                    .consumerProperties(
                            props -> props.put("zookeeper.session.timeout.ms", "500").put("zookeeper.sync.time.ms", "250").
                            put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer").
                            put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer").
                            put("auto.offset.reset", "smallest").
                            put("auto.commit.interval.ms", "100")
                            )
                    .addConsumer(this.kafkaConfig.getConsumerGroup(),
                            metadata -> metadata.consumerTimeout(100)
                                    .topicStreamMap(m -> m.put(this.kafkaConfig.getTopicRead(), 1)).maxMessages(1));

            Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer = e -> e.poller(p -> p.fixedDelay(100));
            return IntegrationFlows.from(messageSourceSpec, endpointConfigurer)
                    .<Map<String, List<byte[]>>> handle((payload, headers) -> {
                        payload.entrySet().forEach(e -> processMessage((ConcurrentHashMap<Integer, List<byte[]>>) e.getValue()));
                        return null;
                    }).get();
        }

版本：

<dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-java-dsl</artifactId>
            <version>1.1.0.RELEASE</version>
        </dependency>