java Kafka事务失败，但仍要提交偏移量

Question

我正试着把我的头绕在卡夫卡的交易上，而且只绕了一次

我已经创建了一个事务性消费者，我想确保阅读和处理某个主题的所有消息。如果事务失败，消息因此丢失，卡夫卡仍会提交偏移量

More formally, if a stream processing application consumes message A and produces message B such that B = F(A), then exactly once processing means that A is considered consumed if and only if B is successfully produced, and vice versa. Source

基于此，我假设消息A没有被消费，因此将再次被重新处理。但这条信息将如何重新处理

我发现有很多信息来源表明，如果处理失败，那么消息就不会被使用。但我找不到任何来源提到如果消息没有被消费，如何重新处理它。我认为，如果事务消费者失败，那么Kafka将回滚，但会继续提交偏移量，以便它可以处理下一条消息。但是如果卡夫卡提交了偏移量，那么之前的消息就丢失了

我发现一些Spring Kafka页面描述了如何处理消费者中的异常。因此，基本上要么停止容器，要么停止整个应用程序。我认为卡夫卡有一些内部机制来处理这种行为，因为文档说明，只有在成功生成消息B时，消息A才会被使用。但是如果消息A没有被使用，那么卡夫卡仍然会提交偏移量并继续下一条消息

感觉只有一次适用于永远不会发生错误的情况。老实说，我不在乎消息是否被消费，卡夫卡是否会提交偏移量。即使消息未被使用，消息也会丢失，因此我似乎必须停止容器或应用程序，以确保不会丢失任何消息

https://spring.io/blog/2017/12/01/spring-for-apache-kafka-2-1-0-release-and-1-3-2-2-0-2-available https://docs.spring.io/spring-kafka/reference/html/_reference.html#annotation-error-handling https://github.com/spring-cloud/spring-cloud-stream/issues/1114

编辑1：我在下面添加了一些示例代码

我有一个非常简单的Spring boot应用程序

@SpringBootApplication
@EnableTransactionManagement
public class KafkaTransactionMysteryApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaTransactionMysteryApplication.class, args);
    }

    @Bean
    @Primary
    public JpaTransactionManager transactionManager() {
        return new JpaTransactionManager();
    }

    @Bean
    public ChainedKafkaTransactionManager chainedTxM(JpaTransactionManager jpa, KafkaTransactionManager<?, ?> kafka) {
        return new ChainedKafkaTransactionManager(kafka, jpa);
    }

    @KafkaListener(topics = "trans-topic")
    @Transactional(propagation = Propagation.REQUIRED, transactionManager = "chainedTxM", rollbackFor = Exception.class)
    public void listen(ConsumerRecord record) throws Exception {
        System.out.println(record.value());
        if (true) {
            throw new Exception("Force rollback");
        }
    }
}

这是我的应用程序属性：

spring.kafka.producer.bootstrap-servers=127.0.0.1:9092
spring.kafka.consumer.bootstrap-servers=127.0.0.1:9092

spring.kafka.producer.transaction-id-prefix=mytrans
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

spring.kafka.consumer.group-id=trans-topic-grp1
spring.kafka.consumer.properties.isolation.level=read_committed
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.ack-mode=RECORD
spring.kafka.consumer.enable-auto-commit=false

这是从控制台制作人发送卡夫卡消息时上述代码的日志：

2018-06-20 10:17:43.494 DEBUG 47970 --- [ntainer#0-0-C-1] o.s.k.t.KafkaTransactionManager          : Created Kafka transaction on producer [CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@55ef3ce1]]
2018-06-20 10:17:43.494 DEBUG 47970 --- [ntainer#0-0-C-1] o.s.orm.jpa.JpaTransactionManager        : Creating new transaction with name [dk.kkflf.kafka.transaction.mystery.KafkaTransactionMysteryApplication$$EnhancerBySpringCGLIB$$67f55772.listen]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT; 'chainedTxM',-java.lang.Exception
2018-06-20 10:17:43.537 DEBUG 47970 --- [ntainer#0-0-C-1] o.h.stat.internal.StatisticsInitiator    : Statistics initialized [enabled=false]
2018-06-20 10:17:43.541 DEBUG 47970 --- [ntainer#0-0-C-1] o.s.orm.jpa.JpaTransactionManager        : Opened new EntityManager [SessionImpl(PersistenceContext[entityKeys=[],collectionKeys=[]];ActionQueue[insertions=ExecutableList{size=0} updates=ExecutableList{size=0} deletions=ExecutableList{size=0} orphanRemovals=ExecutableList{size=0} collectionCreations=ExecutableList{size=0} collectionRemovals=ExecutableList{size=0} collectionUpdates=ExecutableList{size=0} collectionQueuedOps=ExecutableList{size=0} unresolvedInsertDependencies=null])] for JPA transaction
2018-06-20 10:17:43.543 DEBUG 47970 --- [ntainer#0-0-C-1] o.h.e.t.internal.TransactionImpl         : begin
2018-06-20 10:17:43.545 DEBUG 47970 --- [ntainer#0-0-C-1] o.s.orm.jpa.JpaTransactionManager        : Exposing JPA transaction as JDBC transaction [org.springframework.orm.jpa.vendor.HibernateJpaDialect$HibernateConnectionHandle@4d13cb21]
hello stackoverflow
2018-06-20 10:17:43.545 DEBUG 47970 --- [ntainer#0-0-C-1] o.s.orm.jpa.JpaTransactionManager        : Initiating transaction rollback
2018-06-20 10:17:43.545 DEBUG 47970 --- [ntainer#0-0-C-1] o.s.orm.jpa.JpaTransactionManager        : Rolling back JPA transaction on EntityManager [SessionImpl(PersistenceContext[entityKeys=[],collectionKeys=[]];ActionQueue[insertions=ExecutableList{size=0} updates=ExecutableList{size=0} deletions=ExecutableList{size=0} orphanRemovals=ExecutableList{size=0} collectionCreations=ExecutableList{size=0} collectionRemovals=ExecutableList{size=0} collectionUpdates=ExecutableList{size=0} collectionQueuedOps=ExecutableList{size=0} unresolvedInsertDependencies=null])]
2018-06-20 10:17:43.545 DEBUG 47970 --- [ntainer#0-0-C-1] o.h.e.t.internal.TransactionImpl         : rolling back
2018-06-20 10:17:43.547 DEBUG 47970 --- [ntainer#0-0-C-1] o.s.orm.jpa.JpaTransactionManager        : Closing JPA EntityManager [SessionImpl(PersistenceContext[entityKeys=[],collectionKeys=[]];ActionQueue[insertions=ExecutableList{size=0} updates=ExecutableList{size=0} deletions=ExecutableList{size=0} orphanRemovals=ExecutableList{size=0} collectionCreations=ExecutableList{size=0} collectionRemovals=ExecutableList{size=0} collectionUpdates=ExecutableList{size=0} collectionQueuedOps=ExecutableList{size=0} unresolvedInsertDependencies=null])] after transaction
2018-06-20 10:17:43.547 DEBUG 47970 --- [ntainer#0-0-C-1] o.s.orm.jpa.EntityManagerFactoryUtils    : Closing JPA EntityManager
2018-06-20 10:17:43.548 DEBUG 47970 --- [ntainer#0-0-C-1] o.s.k.t.KafkaTransactionManager          : Initiating transaction rollback
2018-06-20 10:17:43.549 DEBUG 47970 --- [ntainer#0-0-C-1] o.a.k.c.p.internals.TransactionManager   : [Producer clientId=producer-1, transactionalId=mytrans0] Transition from state IN_TRANSACTION to ABORTING_TRANSACTION
2018-06-20 10:17:43.549 DEBUG 47970 --- [ntainer#0-0-C-1] o.a.k.c.p.internals.TransactionManager   : [Producer clientId=producer-1, transactionalId=mytrans0] Enqueuing transactional request (type=EndTxnRequest, transactionalId=mytrans0, producerId=0, producerEpoch=21, result=ABORT)
2018-06-20 10:17:43.549 DEBUG 47970 --- [ad | producer-1] o.a.k.c.p.internals.TransactionManager   : [Producer clientId=producer-1, transactionalId=mytrans0] Not sending EndTxn for completed transaction since no partitions or offsets were successfully added
2018-06-20 10:17:43.549 DEBUG 47970 --- [ad | producer-1] o.a.k.c.p.internals.TransactionManager   : [Producer clientId=producer-1, transactionalId=mytrans0] Transition from state ABORTING_TRANSACTION to READY
2018-06-20 10:17:43.550 DEBUG 47970 --- [ntainer#0-0-C-1] o.s.k.t.KafkaTransactionManager          : Resuming suspended transaction after completion of inner transaction
2018-06-20 10:17:43.552 ERROR 47970 --- [ntainer#0-0-C-1] o.s.kafka.listener.LoggingErrorHandler   : Error while processing: ConsumerRecord(topic = trans-topic, partition = 0, offset = 39, CreateTime = 1529479063351, serialized key size = -1, serialized value size = 19, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello stackoverflow)

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void dk.kkflf.kafka.transaction.mystery.KafkaTransactionMysteryApplication.listen(org.apache.kafka.clients.consumer.ConsumerRecord) throws java.lang.Exception' threw exception; nested exception is java.lang.Exception: Force rollback
    at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:267) ~[spring-kafka-2.1.6.RELEASE.jar:2.1.6.RELEASE]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:80) ~[spring-kafka-2.1.6.RELEASE.jar:2.1.6.RELEASE]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:51) ~[spring-kafka-2.1.6.RELEASE.jar:2.1.6.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1071) [spring-kafka-2.1.6.RELEASE.jar:2.1.6.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1051) [spring-kafka-2.1.6.RELEASE.jar:2.1.6.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:998) [spring-kafka-2.1.6.RELEASE.jar:2.1.6.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:866) [spring-kafka-2.1.6.RELEASE.jar:2.1.6.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:724) [spring-kafka-2.1.6.RELEASE.jar:2.1.6.RELEASE]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_172]
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_172]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_172]
Caused by: java.lang.Exception: Force rollback
    at dk.kkflf.kafka.transaction.mystery.KafkaTransactionMysteryApplication.listen(KafkaTransactionMysteryApplication.java:39) ~[classes/:na]
    at dk.kkflf.kafka.transaction.mystery.KafkaTransactionMysteryApplication$$FastClassBySpringCGLIB$$7b3cab85.invoke(<generated>) ~[classes/:na]
    at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204) ~[spring-core-5.0.6.RELEASE.jar:5.0.6.RELEASE]
    at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746) ~[spring-aop-5.0.6.RELEASE.jar:5.0.6.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-5.0.6.RELEASE.jar:5.0.6.RELEASE]
    at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:294) ~[spring-tx-5.0.6.RELEASE.jar:5.0.6.RELEASE]
    at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:98) ~[spring-tx-5.0.6.RELEASE.jar:5.0.6.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185) ~[spring-aop-5.0.6.RELEASE.jar:5.0.6.RELEASE]
    at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688) ~[spring-aop-5.0.6.RELEASE.jar:5.0.6.RELEASE]
    at dk.kkflf.kafka.transaction.mystery.KafkaTransactionMysteryApplication$$EnhancerBySpringCGLIB$$1375ca24.listen(<generated>) ~[classes/:na]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_172]
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_172]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_172]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_172]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:181) ~[spring-messaging-5.0.6.RELEASE.jar:5.0.6.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:114) ~[spring-messaging-5.0.6.RELEASE.jar:5.0.6.RELEASE]
    at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:48) ~[spring-kafka-2.1.6.RELEASE.jar:2.1.6.RELEASE]
    at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:248) ~[spring-kafka-2.1.6.RELEASE.jar:2.1.6.RELEASE]
    ... 10 common frames omitted

2018-06-20 10:17:43.553 DEBUG 47970 --- [ntainer#0-0-C-1] essageListenerContainer$ListenerConsumer : Commit list: {trans-topic-0=OffsetAndMetadata{offset=40, metadata=''}}
2018-06-20 10:17:43.553 DEBUG 47970 --- [ntainer#0-0-C-1] essageListenerContainer$ListenerConsumer : Committing: {trans-topic-0=OffsetAndMetadata{offset=40, metadata=''}}
2018-06-20 10:17:43.554 DEBUG 47970 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-1, groupId=trans-topic-grp1] Committed offset 40 for partition trans-topic-0
2018-06-20 10:17:43.693 DEBUG 47970 --- [rans-topic-grp1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=trans-topic-grp1] Sending Heartbeat request to coordinator localhost:9092 (id: 2147483647 rack: null)
2018-06-20 10:17:43.696 DEBUG 47970 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=trans-topic-grp1] Received successful Heartbeat response
2018-06-20 10:17:43.747 DEBUG 47970 --- [ntainer#0-0-C-1] o.a.k.c.consumer.internals.Fetcher       : [Consumer clientId=consumer-1, groupId=trans-topic-grp1] Fetch READ_COMMITTED at offset 40 for partition trans-topic-0 returned fetch data (error=NONE, highWaterMark=40, lastStableOffset = 40, logStartOffset = 0, abortedTransactions = [], recordsSizeInBytes=0)
2018-06-20 10:17:43.747 DEBUG 47970 --- [ntainer#0-0-C-1] o.a.k.c.consumer.internals.Fetcher       : [Consumer clientId=consumer-1, groupId=trans-topic-grp1] Added READ_COMMITTED fetch request for partition trans-topic-0 at offset 40 to node localhost:9092 (id: 0 rack: null)
2018-06-20 10:17:43.748 DEBUG 47970 --- [ntainer#0-0-C-1] o.a.k.c.consumer.internals.Fetcher       : [Consumer clientId=consumer-1, groupId=trans-topic-grp1] Sendin

源代码也可以在github上访问 https://github.com/kkflf/kafka-transaction-mystery/tree/SO-50922050

编辑2：我的消费者现在将回滚异常，并且不会将偏移提交给卡夫卡。（万岁！）。我现在正试图通过一个JPA事务链接到一个MySQL数据库

下面的代码将MyMessage对象持久化到我的MySQL数据库中，但它将回滚Kafka事务并再次重播相同的Kafka消息（这是一件好事）。这将导致MySQL数据库中出现重复条目，因为Kafka做了回滚，而JPA没有回滚。我该如何克服这种情况——卡夫卡利斯汀和我的存储库似乎不共享同一个事务会话

我已将最新的更改上传到https://github.com/kkflf/kafka-transaction-mystery/tree/SO-50922050-edit2

这篇文章越来越长了，所以我把日志文件推到这里： https://github.com/kkflf/kafka-transaction-mystery/blob/SO-50922050-edit2/log-SO-50922050-edit2.log

主要用途：

@SpringBootApplication
@EnableTransactionManagement
public class KafkaTransactionMysteryApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaTransactionMysteryApplication.class, args);
    }

    @Autowired
    private MessageRepository messageRepository;

    @Bean
    public JpaTransactionManager transactionManager() {
        return new JpaTransactionManager();
    }

    @Bean
    @Primary
    public ChainedKafkaTransactionManager chainedTxM(JpaTransactionManager jpa, KafkaTransactionManager<?, ?> kafka) {
        return new ChainedKafkaTransactionManager(kafka, jpa);
    }

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
    kafkaListenerContainerFactory(JpaTransactionManager jpa, KafkaTransactionManager<?, ?> kafka) {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        factory.getContainerProperties().setTransactionManager(chainedTxM(jpa, kafka));
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerProps());
    }

    private Map<String, Object> consumerProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "trans-topic-grp1");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

    @KafkaListener(topics = "trans-topic", containerFactory = "kafkaListenerContainerFactory")
    @Transactional(propagation = Propagation.REQUIRED, transactionManager = "chainedTxM", rollbackFor = Exception.class)
    public void listen(List<String> records) throws Exception {
        for (String record : records) {
            MyMessage message = new MyMessage(record);
            messageRepository.save(message);
            if (record.equals("fail")) {
                throw new Exception("Forced rollback - msg: " + record);
            }
        }
    }
}

将MyMessage对象持久化到数据库的存储库：

@Repository
public class MessageRepository {

    @Autowired
    EntityManager em;

    //Mandatory to make sure this is never executed outside a transaction
    @Transactional(propagation = Propagation.MANDATORY)
    public void save(MyMessage message) {
        em.persist(message);
    }

    public MyMessage findById(Long id) {
        return em.find(MyMessage.class, id);
    }
}

Answer 1

1. 不要在事务中使用auto.commit.enabled=true；当KafkaTransactionManager被注入容器时，如果侦听器正常退出，容器将向事务发送偏移量
2. 有关回滚，请参见AfterRollbackProcessor。回滚事务后调用DefaultAfterRollbackProcessor；它会重新查找所有未处理的记录，以便卡夫卡在下一个consumer.poll()再次返回这些记录

你找到的关于停止/启动容器的参考资料是关于spring kafka的旧版本。例如，当不使用事务时，我们现在提供SeekToCurrentErrorHandler（在使用事务时，它执行与AfterRollbackProcessor类似的功能）