如果您确实想手动提交偏移量，那么类似的方法可以帮助您。
如果你不想要它，那就换 setAckMode 其他价值。
这件事是按Spring的方式做的。 CoreAutoConfiguration 班级：

@Configuration
@Import({KafkaAutoConfiguration.class})
public class CoreAutoConfiguration {

@Bean("batchKafkaListenerContainerFactory")
public ConcurrentKafkaListenerContainerFactory<?, ?> batchKafkaListenerContainerFactory(ConcurrentKafkaListenerContainerFactoryConfigurer configurer, ConsumerFactory<Object, Object> kafkaConsumerFactory) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, kafkaConsumerFactory);
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    factory.setBatchListener(true);
    return factory;
    }
}

然后你就走了 Config 班级：

@Configuration
@Import({
        CoreAutoConfiguration.class,
        KafkaAutoConfiguration.class,
})
@EnableKafka
@EnableRetry
public class Config {
}

最后，消费者：

@KafkaListener(
        topics = "MessagesTopic",
        containerFactory = "batchKafkaListenerContainerFactory"
)
public void dataReceived(@Payload List<String> payload) throws RuntimeException {
    yourService.processIncomingData(payload);
}

最后，属性：

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=helloworld
spring.kafka.listener.type=batch
spring.kafka.consumer.enable-auto-commit=false

# this is size of incoming list if broker has this many entries, can be lower eventually

spring.kafka.consumer.max-poll-records=100
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer