executeInTransaction内部的生产者失败

Question

我们也正在手动更新成功提交后的偏移。KafkaTransactionManager用于维护事务。由于消息是通过RestController发布到firstTopic的，我们的@Transactional从那里开始，在偏移更新时结束。为此，我们使用executeInTransaction（）。

Kafka造型

@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);
    config.put(ProducerConfig.RETRIES_CONFIG, 10);
    config.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 1000);
    DefaultKafkaProducerFactory<String, User> defaultKafkaProducerFactory = new DefaultKafkaProducerFactory<>(config);
    defaultKafkaProducerFactory.setTransactionIdPrefix("trans");
    //defaultKafkaProducerFactory.transactionCapable();
    return defaultKafkaProducerFactory;
    //return new DefaultKafkaProducerFactory<>(config);
}

/**
 * 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(10 * 1000);
    backOffPolicy.setMultiplier(1);
    backOffPolicy.setMaxInterval(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;
}

Rest控制器

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

    @Autowired
    KafkaTemplate<String, User> kafkaTemplate;


    @PostMapping("/users")
    @Transactional
    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";
    }

}

@Service
public class KafkaMessageConsumerService {

    @Autowired
    KafkaTemplate<String, User> kafkaTemplate;

    int index = 0;

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


        kafkaTemplate.executeInTransaction(t -> {
            int number = (int) (Math.random() * 10);


            t.send("secondtopic", user);
            if (number % 2 == 0 || number % 2 == 1) {
                System.out.println("about to ack");
                throw new RuntimeException(" Modulus is zero ");
            }
            acknowledgment.acknowledge();

            return true;
        }); 
    }


}

这是解决生产者失败的正确方法吗？

有没有方法在第10次重试后捕获异常并根据业务规则处理消息？我们可以保留2个RetryTemplates和SeekToErrorHandlers吗？

Answer 1

您不应该在那里使用executeintransaction（除非您确实想在其他事务中执行发布）。侦听器容器已启动事务。

此外，acknowledge()将偏移量发送到容器的事务，而不是模板启动的事务。

这可能就是为什么您在确认上会出现奇怪的错误；我不知道.