使用 GCP PubSub 后无法提交事务 JPA 存储库 Spring

Question

我有一个 Spring Boot 应用程序，它在 GCP PubSub 队列上发送和接收消息。基本上，我尝试将我在消息中获得的一些信息存储在数据库中。但是，当我尝试提交此信息或 saveAndFlush() 时，我收到错误 RollbackException: Error while committing the transaction。如果我保留应用程序收到入站消息时未触发的任何其他信息，则它可以正常工作。如果我尝试 saveAndFlush 而不是 save 我得到 NullPointerException。

我错过了什么？

谢谢。

错误堆栈

Caused by: org.springframework.transaction.TransactionSystemException: Could not commit JPA transaction; nested exception is javax.persistence.RollbackException: Error while committing the transaction
    at org.springframework.orm.jpa.JpaTransactionManager.doCommit(JpaTransactionManager.java:543)
    at org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager.java:743)
    at org.springframework.transaction.support.AbstractPlatformTransactionManager.commit(AbstractPlatformTransactionManager.java:711)
    at org.springframework.transaction.interceptor.TransactionAspectSupport.commitTransactionAfterReturning(TransactionAspectSupport.java:633)
    at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:386)
    at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:118)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
    at org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
    at org.springframework.data.jpa.repository.support.CrudMethodMetadataPostProcessor$CrudMethodMetadataPopulatingMethodInterceptor.invoke(CrudMethodMetadataPostProcessor.java:178)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
    at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:95)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212)
    at com.sun.proxy.$Proxy153.save(Unknown Source)
    at com.sap.service.deployments.DeploymentService.lambda$saveLogs$16(DeploymentService.java:235)
    at java.util.Optional.map(Optional.java:215)
    at com.sap.service.deployments.DeploymentService.saveLogs(DeploymentService.java:233)
    at com.sap.service.deployments.DeploymentService.processDeploymentExecutionResponse(DeploymentService.java:404)
    at com.sap.service.deployments.DeploymentService.messageReceiver(DeploymentService.java:245)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:171)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:120)
    at org.springframework.integration.handler.support.MessagingMethodInvokerHelper$HandlerMethod.invoke(MessagingMethodInvokerHelper.java:1105)
    at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.invokeHandlerMethod(MessagingMethodInvokerHelper.java:583)
    at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.processInternal(MessagingMethodInvokerHelper.java:478)
    at org.springframework.integration.handler.support.MessagingMethodInvokerHelper.process(MessagingMethodInvokerHelper.java:356)
    at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:108)
    ... 24 more
Caused by: javax.persistence.RollbackException: Error while committing the transaction
    at org.hibernate.internal.ExceptionConverterImpl.convertCommitException(ExceptionConverterImpl.java:81)
    at org.hibernate.engine.transaction.internal.TransactionImpl.commit(TransactionImpl.java:104)
    at org.springframework.orm.jpa.JpaTransactionManager.doCommit(JpaTransactionManager.java:534)
    ... 54 more
Caused by: java.lang.NullPointerException
    at com.sap.config.AuditorAwareImpl.getCurrentAuditor(AuditorAwareImpl.java:13)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:205)
    at com.sun.proxy.$Proxy164.getCurrentAuditor(Unknown Source)
    at org.springframework.data.auditing.AuditingHandler.lambda$touchAuditor$6(AuditingHandler.java:193)
    at java.util.Optional.map(Optional.java:215)
    at org.springframework.data.auditing.AuditingHandler.touchAuditor(AuditingHandler.java:191)
    at org.springframework.data.auditing.AuditingHandler.lambda$touch$0(AuditingHandler.java:165)
    at java.util.Optional.map(Optional.java:215)
    at org.springframework.data.auditing.AuditingHandler.touch(AuditingHandler.java:163)
    at org.springframework.data.auditing.AuditingHandler.markModified(AuditingHandler.java:143)
    at org.springframework.data.jpa.domain.support.AuditingEntityListener.touchForUpdate(AuditingEntityListener.java:112)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.hibernate.jpa.event.internal.ListenerCallback.performCallback(ListenerCallback.java:35)
    at org.hibernate.jpa.event.internal.CallbackRegistryImpl.callback(CallbackRegistryImpl.java:95)
    at org.hibernate.jpa.event.internal.CallbackRegistryImpl.preUpdate(CallbackRegistryImpl.java:69)
    at org.hibernate.event.internal.DefaultFlushEntityEventListener.invokeInterceptor(DefaultFlushEntityEventListener.java:366)
    at org.hibernate.event.internal.DefaultFlushEntityEventListener.handleInterception(DefaultFlushEntityEventListener.java:348)
    at org.hibernate.event.internal.DefaultFlushEntityEventListener.scheduleUpdate(DefaultFlushEntityEventListener.java:299)
    at org.hibernate.event.internal.DefaultFlushEntityEventListener.onFlushEntity(DefaultFlushEntityEventListener.java:170)
    at org.hibernate.event.internal.AbstractFlushingEventListener.flushEntities(AbstractFlushingEventListener.java:232)
    at org.hibernate.event.internal.AbstractFlushingEventListener.flushEverythingToExecutions(AbstractFlushingEventListener.java:92)
    at org.hibernate.event.internal.DefaultFlushEventListener.onFlush(DefaultFlushEventListener.java:39)
    at org.hibernate.event.service.internal.EventListenerGroupImpl.fireEventOnEachListener(EventListenerGroupImpl.java:102)
    at org.hibernate.internal.SessionImpl.doFlush(SessionImpl.java:1360)
    at org.hibernate.internal.SessionImpl.managedFlush(SessionImpl.java:451)
    at org.hibernate.internal.SessionImpl.flushBeforeTransactionCompletion(SessionImpl.java:3210)
    at org.hibernate.internal.SessionImpl.beforeTransactionCompletion(SessionImpl.java:2378)
    at org.hibernate.engine.jdbc.internal.JdbcCoordinatorImpl.beforeTransactionCompletion(JdbcCoordinatorImpl.java:447)
    at org.hibernate.resource.transaction.backend.jdbc.internal.JdbcResourceLocalTransactionCoordinatorImpl.beforeCompletionCallback(JdbcResourceLocalTransactionCoordinatorImpl.java:183)
    at org.hibernate.resource.transaction.backend.jdbc.internal.JdbcResourceLocalTransactionCoordinatorImpl.access$300(JdbcResourceLocalTransactionCoordinatorImpl.java:40)
    at org.hibernate.resource.transaction.backend.jdbc.internal.JdbcResourceLocalTransactionCoordinatorImpl$TransactionDriverControlImpl.commit(JdbcResourceLocalTransactionCoordinatorImpl.java:281)
    at org.hibernate.engine.transaction.internal.TransactionImpl.commit(TransactionImpl.java:101)
    ... 55 more

AuditorAwareImpl

import org.springframework.data.domain.AuditorAware;
import org.springframework.security.core.context.SecurityContextHolder;

import java.util.Optional;

class AuditorAwareImpl implements AuditorAware<String> {

    @Override
    public Optional<String> getCurrentAuditor() {
        SamlUserDetails userDetails = (SamlUserDetails) SecurityContextHolder.getContext().getAuthentication().getPrincipal(); //NullPointerExc on this line
        return Optional.of(userDetails.getFullname() + " (" + userDetails.getAttribute("uid") + ")");
    }
}

这是 Inbound PubSub 的配置

@Component
public class PubSubInbound {

    @Autowired
    private DeploymentService deploymentService;

    @Bean
    public MessageChannel inputMessageChannel() {
        return new PublishSubscribeChannel();
    }

    // Create an inbound channel adapter to listen to the subscription `sub-one` and send
    // messages to the input message channel.
    @Bean
    public PubSubInboundChannelAdapter inboundChannelAdapter(
            @Qualifier("inputMessageChannel") MessageChannel messageChannel,
            PubSubTemplate pubSubTemplate) {
        PubSubInboundChannelAdapter adapter =
                new PubSubInboundChannelAdapter(pubSubTemplate, "cerberusSubscription");
        adapter.setOutputChannel(messageChannel);
        adapter.setAckMode(AckMode.MANUAL);
        adapter.setPayloadType(String.class);
        return adapter;
    }
    @Bean
    @ServiceActivator(inputChannel = "pubsubInputChannel")
    public MessageHandler messageReceiver() {
        return message -> {
            System.out.println("Message arrived! Payload: " + new String((byte[]) message.getPayload()));
            BasicAcknowledgeablePubsubMessage originalMessage =
                    message.getHeaders().get(GcpPubSubHeaders.ORIGINAL_MESSAGE, BasicAcknowledgeablePubsubMessage.class);
            originalMessage.ack();
            deploymentService.persistLogs(
               Long.parseLong(message.getHeaders().get("deploymentId", String.class)),
               new String((byte[]) message.getPayload()));
        };
    }


}

部署服务

@Service
public class DeploymentService {

    @Autowired
    private DeploymentRepository deploymentRepository;

....

    public Optional<Deployment> persistLogs(Long deploymentId, String logs){
        Optional<Deployment> deploymentOpt = deploymentRepository.findById(deploymentId);
        return deploymentOpt.map(deployment -> {
            deployment.setLogs(logs);
            return deploymentRepository.save(deployment); //ERROR HERE
        });
    }

....

部署存储库

@Repository
public interface DeploymentRepository extends JpaRepository<Deployment, Long> { }

Answer 1

解决了。

SecurityContextHolder.getContext().getAuthentication()

当消息来自 GCP PubSub 时为 null。重做 getCurrentAuditor 就成功了。

    @Override
    public Optional<String> getCurrentAuditor() {
        if(SecurityContextHolder.getContext().getAuthentication()!=null){
            SamlUserDetails userDetails = (SamlUserDetails) SecurityContextHolder.getContext().getAuthentication().getPrincipal();
            return Optional.of(userDetails.getFullname() + " (" + userDetails.getAttribute("uid") + ")");
        }else return Optional.empty();
    }