登录
当前位置： 首页 > 知识库问答 >
问题：

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

傅志文
2023-03-14

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

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

更正式地说，如果流处理应用程序使用消息A并生成消息B，使得B=F（A），那么恰好一次处理意味着当且仅当成功生成B时才认为A被消耗，反之亦然。来源

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

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

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

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

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-错误处理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

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

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：我的消费者现在将回滚异常，并且不会将偏移提交给Kafka。（万岁！）。我现在正试图通过一个JPA事务链接到一个MySQL数据库。

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

我已将最新更改上传到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);
    }
}

共有1个答案

韩嘉祯
2023-03-14
  1. 不要使用auto。犯罪enabled=true（启用）（带事务）；将KafkaTransactionManager注入容器时，如果侦听器正常退出，容器将向事务发送偏移量

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

 类似资料：
  • 异步提交偏移量时“提交偏移量失败”

    我有一个Kafka消费者，我从它消费数据从一个特定的主题，我看到下面的例外。我使用的是Kafka版本。 我添加了这两个额外的消费者属性，但仍然没有帮助： 那个错误意味着什么？我该如何解决它？我需要添加一些其他消费者属性吗？

  • Kafka-偏移提交

    我目前正在从具有特定偏移量的主题中获取消息。我正在使用寻求（）来实现它。但是当我将enable.auto.commit设置为true或使用手动同步（委托同步（）/委托同步（））时，Seek（）不起作用，因为它没有轮询来自特定偏移量的消息，而是从最后提交的偏移量中选择。 因此，在使用Seek（）时，是否必须将偏移量存储在外部DB中，而不提交给Kafka？Seek和Commit不能并行工作吗？ 客户端

  • kafka中的某些提交偏移调用失败

    Kafka 服务器和客户端 JAR 移至最新库：0.10.0.1 我的消费者和生产者代码使用如上所述的最新kafka jars，但仍然使用旧的消费者API(0 . 8 . 2)。 我在调用commit offset时在消费者端遇到问题。 kafka服务器端配置： 以下 Kafka 消费者的配置： 要创建消费者，我使用以下api： 和提交调用 在从 Kafka 读取消息时，我们使用以下方法来处理超时

  • Spring Kafka多并发使用者-提交偏移量失败

    我正在使用spring with Kafka来消费来自Kafka主题的数据。我已经将并发配置为10。因此不同的线程轮询代理以获取消息并处理消息。即使在一段时间后（成功处理），我们也会收到相同的消息返回给使用者的不同线程。我们能够在配置的max.poll.interval.ms=1500000内处理接收到的消息。 请找到以下配置的Kafka消费者属性。我已经通过Kafka配置了自动提交。 你能帮我解

  • 异步自动提交偏移量失败

    我有一个关于Kafka自动提交机制的问题。我正在使用启用自动提交的Spring-Kafka。作为一个实验，我在系统空闲（主题中没有新消息，没有正在处理的消息）的情况下，将我的消费者与Kafka的连接断开了30秒。重新连接后，我收到了如下几条消息： 第一，我不明白有什么好犯的？系统空闲（所有以前的消息都已提交）。第二，断开时间为30秒，比max.poll.interval.ms的5分钟（300000

  • Kafka中的提交偏移量

    我有一个ReactorKafka项目，它消耗来自Kafka主题的消息，转换消息，然后写入到另一个主题。 我的理解是，只有在Reactor中成功完成所有顺序步骤后，才会提交偏移量。对吗？我想确保不会处理下一条记录，除非当前记录成功发送到目标Kafka主题。

相关问答

异步自动提交偏移量失败Kafka中的提交偏移量Kafka控制台使用者错误“分区上的偏移量提交失败”Spring Kafka@kafkalistener-重试发送失败的消息并手动提交偏移量当手动提交特定偏移失败但序列中的下一个偏移成功提交时会发生什么？

相关文章

在当前事务之外提交事务（例如Oracle中的自主事务）java事务回滚失败问题分析jQuery .on（）提交事件Kafka如何存储每个主题的偏移量？延迟发送邮件，直到事务提交

相关阅读

GitLab压缩提交DBMS事务SQLite 事务PostgreSQL 事务SQL 事务

相关工具

移动短信Cmpp服务端，客户端个人网-个人事务管理系统kafka-monitorkafka-phpKafka Manager

相关文档

了解亚马逊网络服务Python 量化交易教程Apache Kafka 官方文档Kafka 中文文档 v1.0Java 提高篇
快捷导航： 新手教程 算法原理 架构设计 Java进阶 数据库进阶 大厂专栏 面试经验 编程笔记 编程问答 所有专题 文档资料 工具软件 电子书籍 小牛导航
在线工具： 房贷计算器 个税计算器 Linux命令查询 Json格式化 正则表达式 颜色转换 AES加解密 SHA1加密 MD5加密 毒鸡汤 字数统计 随机密码生成 进制转换 Base64编解码 励志句子
Copyright © 2019-2024 小牛知识库@xnip.cn. All Rights Reserved.