在spring kafka中发布到主题和手动提交

Question

@Bean
public ProducerFactory producerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

    DefaultKafkaProducerFactory<String, User> defaultKafkaProducerFactory = new DefaultKafkaProducerFactory<>(config);
    defaultKafkaProducerFactory.setTransactionIdPrefix("trans");
    //defaultKafkaProducerFactory.transactionCapable();
    return defaultKafkaProducerFactory;
    //return new DefaultKafkaProducerFactory<>(config);
}


@Bean
public KafkaTemplate<String, User> kafkaTemplate() {
    return new KafkaTemplate<>(producerFactory());
}


@Bean
public KafkaTransactionManager<String, User> transactionManager() {
    KafkaTransactionManager transactionManager = new KafkaTransactionManager(producerFactory());
    transactionManager.setTransactionSynchronization(AbstractPlatformTransactionManager.SYNCHRONIZATION_ALWAYS);
    transactionManager.setNestedTransactionAllowed(true);
    return transactionManager;
}


/**
 * New configuration for the consumerFactory added
 *
 * @return
 */
@Bean
public ConsumerFactory<String, User> consumerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    config.put(ConsumerConfig.GROUP_ID_CONFIG, "firstTopic-group");
    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(), new JsonDeserializer<User>(User.class));
}


@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, User>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, User> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    factory.getContainerProperties().setTransactionManager(transactionManager());
    factory.setRetryTemplate(kafkaRetry());
    factory.setStatefulRetry(true);
    factory.setErrorHandler(getErrorHandler());
    factory.setRecoveryCallback(retryContext -> {
        //implement the logic to decide the action after all retries are over.
        ConsumerRecord consumerRecord = (ConsumerRecord) retryContext.getAttribute("record");
        System.out.println("Recovery is called for message  " + consumerRecord.value());
        return Optional.empty();
    });

    return factory;
}



public RetryTemplate kafkaRetry() {
    RetryTemplate retryTemplate = new RetryTemplate();

    ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
    backOffPolicy.setInitialInterval(60 * 1000);
    backOffPolicy.setMultiplier(3);
    backOffPolicy.setMaxInterval(4 * 60 * 1000);       // original 25 * 60 * 1000
    retryTemplate.setBackOffPolicy(backOffPolicy);

    SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
    retryPolicy.setMaxAttempts(4);
    retryTemplate.setRetryPolicy(retryPolicy);
    return retryTemplate;
}


public SeekToCurrentErrorHandler getErrorHandler() {
    SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler() {

        @Override
        public void handle(Exception thrownException,
                           List<ConsumerRecord<?, ?>> records,
                           Consumer<?, ?> consumer,
                           MessageListenerContainer container) {
            //super.handle(thrownException, records, consumer, container);
            if (!records.isEmpty()) {
                ConsumerRecord<?, ?> record = records.get(0);
                String topic = record.topic();
                long offset = record.offset();
                int partition = record.partition();

                if (thrownException instanceof DeserializationException) {
                    System.out.println("------1111------deserialization exception ");
                } else {
                    System.out.println("------xxxx------Record is empty ");
                    consumer.seek(new TopicPartition(topic, partition), offset);
                }
            } else {
                System.out.println("------4444------Record is empty ");
            }

        }
    };

    return errorHandler;
}

   @Autowired
KafkaTemplate<String, User> kafkaTemplate;


@KafkaListener(topics = "firstTopic", groupId = "firstTopic-group")
@Transactional
public void onCustomerMessage(User user, Acknowledgment acknowledgment) throws Exception {

    /*System.out.println("get the message " +user.getFirstName());
    if (user.getFirstName().equalsIgnoreCase("Test")) {
        throw new RuntimeException("Incompatible message " + user.getFirstName());
    }
    */

    //postToSecondTopic(acknowledgment, user);

    System.out.println("NOT In transaction");
    kafkaTemplate.executeInTransaction(t -> {
        System.out.println("---------------------->");
        int number = (int) (Math.random() * 10);
        t.send("secondtopic", user);
        if (number % 5 == 0)
            throw new RuntimeException("fail");
        acknowledgment.acknowledge();
        return true;
    });


    System.out.println("*** exit ***");
}

日志中的错误

2020-05-28 15:52:53.597错误112469---[nio-8080-exec-1]O.a.C.C.C.[.[.[/].[dispatcherServlet]:servlet.Service（）对于servlet[dispatcherServlet]在路径[]上下文中引发异常[请求处理失败；嵌套异常是java.lang.IllegalStateException：没有事务在处理中；可能的解决方案：在template.executeinTransaction（）操作范围内运行模板操作，在调用模板方法之前使用@transactional启动事务，在使用记录时在侦听器容器启动的事务中运行

IllegalStateException：没有事务在处理中；可能的解决方案：在template.executeintransaction（）操作范围内运行模板操作，在调用模板方法之前用@transactional启动事务，在使用org.springframework.util.assert.state(assert.java:73)~[spring-core-5.2.5.release.jar:5.2.5.release]在org.springframework.kafka.core.kafkatemplate.doSend(kafkatemplate.java:394)~[spring-kafka-2.3.7.release.jar:2.3.7.plate.java:216)~[spring-kafka-2.3.7.release.jar:2.3.7.release]在com.barade.sandesh.springkafka.userresource.postcomments(userresource.java:26)~[classes/:na]在sun.reflect.nativeMethodAccessorimpl.invoke0（原生方法）~[na:1.8.0_252]在在java.lang.reflect.method.invoke(method.java:498)~[NA:1.8.0_252]在org.springframework.web.method.support.invocablehandlermethod.doinvoke(invocablehandlermethod.java:190)~[spring-web-5.2.5.release.jar:5.2.5.release]在AbleHandlerMethod.invokeandHandland(servletInvocableHandlerMethod.java:105)~[spring-webmvc-5.2.5.release.jar:5.2.5.release]在org.springframework.web.servlet.mvc.method.annotation.requestMappingHandlerAdapter.invale(requestMappingHandlerAdapter.invale(requestMappingHandlerAdapter.java:879)~[spring-webmapping-5.2.5.release]在793)~[spring-webmvc-5.2.5.release.jar:5.2.5.release]位于org.springframework.web.servlet.mvc.methOd.abstractHandlerMethodAdapter.handle(abstractHandlerMethodAdapter.java:87)~[spring-webmvc-5.2.5.release.jar:5.2.5.release]在org.springframework.web.servlet.dispatcherServlet.doDispatcherServlet.java:1040)~[spring-webmvc-5.2.5.release]在org.springframework.web.servlet.dispatcherServlet:5.2.5.release]在(frameworkservlet.java:1006)~[spring-webmvc-5.2.5.release.jar:5.2.5.release]在org.springframework.web.servlet.frameworkservlet.doPost（frameworkservlet.java:909)~[spring-webmvc-5.2.5.release.jar:5.2.5.release]

用户资源

@RestController
@RequestMapping("accounts")
public class UserResource {

    @Autowired
    KafkaTemplate <String, User> kafkaTemplate;


    @PostMapping("/users")
    public String postComments(@RequestParam ("firstName") final String firstName,
                                    @RequestParam ("lastName") final String lastName,
                                    @RequestParam ("userName") final String userName )  {

        List<String> accountTypes = new ArrayList<String>();
        kafkaTemplate.send("firstTopic", new User(firstName,lastName,userName));

        return "Message sent to the Error queue";
    }



}

Answer 1

是；参见交易记录。

容器启动事务，监听器中的任何kafkatemplate发送操作都参与事务；容器将偏移量发送到事务并提交（如果监听器正常退出）。

不需要“手动”提交偏移量。