Missing synchronizations but resources already present if exception occurs in "commitAll"

[kuldeepkalassts]

We are getting this exception randomly at our code when send message flow is in spring transaction and message are not commit:

"com.rabbitmq.client.ChannelContinuationTimeoutException: Continuation call for method #method<tx.commit>() on channel AMQChannel(amqp://server@ip:5672/server,1) (#1) timed out": ["AMQChannel.java", 313, "com.rabbitmq.client.impl.AMQChannel.wrapTimeoutException", "AMQChannel.java", 295, "com.rabbitmq.client.impl.AMQChannel.privateRpc", "AMQChannel.java", 141, "com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc", "ChannelN.java", 1540, "com.rabbitmq.client.impl.ChannelN.txCommit", "ChannelN.java", 52, "com.rabbitmq.client.impl.ChannelN.txCommit", "", -1, "sun.reflect.GeneratedMethodAccessor452.invoke", "DelegatingMethodAccessorImpl.java", 43, "sun.reflect.DelegatingMethodAccessorImpl.invoke", "Method.java", 498, "java.lang.reflect.Method.invoke", "CachingConnectionFactory.java", 1190, "org.springframework.amqp.rabbit.connection.CachingConnectionFactory$CachedChannelInvocationHandler.invoke", "", -1, "com.sun.proxy.$Proxy941.txCommit", "RabbitResourceHolder.java", 153, "org.springframework.amqp.rabbit.connection.RabbitResourceHolder.commitAll"],

After this error occurred we are seeing that the spring transaction synchronizations are missing, but resource for rabbitResourceHolder is present , we are verifying this with a log we printed

Logging Code

	private void logSpringTransactionDetailsForRabbitFlow(String routingKey, RabbitTemplate template) {
		if (template.isChannelTransacted() && TransactionSynchronizationManager.isActualTransactionActive() && isRabbitResourceOrSyncNotPresent(template)) {
			LOGGER.error("For routing key {} current spring transaction {}, have these synchronizations : {} and this resourcemap {}",
					routingKey, TransactionSynchronizationManager.getCurrentTransactionName(),
					TransactionSynchronizationManager.getSynchronizations(),
					TransactionSynchronizationManager.getResourceMap());
		}
	}


	private boolean isRabbitResourceOrSyncNotPresent(RabbitTemplate template) {
		return !TransactionSynchronizationManager.hasResource(template.getConnectionFactory()) || !isRabbitSyncPresent();
	}

	private boolean isRabbitSyncPresent() {
		for (TransactionSynchronization transactionSynchronization : TransactionSynchronizationManager.getSynchronizations()) {
			if(transactionSynchronization.getClass().getName().contains("RabbitResourceSynchronization")) {
				return true;
			}
		}
		return false;
	}

Output :
For routing key tk.news.alerts current spring transaction name_of_transaction, have these synchronizations : [] and this resourcemap {org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean@616f930f=org.springframework.orm.jpa.EntityManagerHolder@61768137,CachingConnectionFactory [channelCacheSize=25, host=our_host_url, port=5672, active=true com.sample.messaging.impl.connection.CachingConnectionFactory@240953f5]=org.springframework.amqp.rabbit.connection.RabbitResourceHolder@692ba9fb}

From logs we can see that synchronization was not added but resources are already present.

So I suspect that because of ChannelContinuationTimeoutException, the first commit was not success, but resources added by this were still present in TransactionSynchronizationManager.getResource

And next time when we send new message in a new transaction, the condition TransactionSynchronizationManager.hasResource return true in below code, thus it do not add the new synchronization for the new message.

//class org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils
	public static RabbitResourceHolder bindResourceToTransaction(RabbitResourceHolder resourceHolder,
			ConnectionFactory connectionFactory, boolean synched) {
		if (TransactionSynchronizationManager.hasResource(connectionFactory)
				|| !TransactionSynchronizationManager.isActualTransactionActive() || !synched) {
			return (RabbitResourceHolder) TransactionSynchronizationManager.getResource(connectionFactory); // NOSONAR never null
		}
		TransactionSynchronizationManager.bindResource(connectionFactory, resourceHolder);
		resourceHolder.setSynchronizedWithTransaction(true);
		if (TransactionSynchronizationManager.isSynchronizationActive()) {
			TransactionSynchronizationManager.registerSynchronization(new RabbitResourceSynchronization(resourceHolder,
					connectionFactory));
		}
		return resourceHolder;
	}

So can this code be enhanced so that orphan resources could be remove ? or is there any work around to handle such cases?

[garyrussell]

Can you provide more information or some way to reproduce this; this test passes fine:

	@Test
	void resourcesClearedAfterTxFails() throws IOException, TimeoutException {
		ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class);
		Connection mockConnection = mock(Connection.class);
		Channel mockChannel = mock(Channel.class);

		given(mockConnectionFactory.newConnection(any(ExecutorService.class), anyString())).willReturn(mockConnection);
		given(mockConnection.isOpen()).willReturn(true);
		given(mockConnection.createChannel()).willReturn(mockChannel);
		given(mockChannel.txSelect()).willReturn(mock(SelectOk.class));
		given(mockChannel.txCommit()).willThrow(IllegalStateException.class);
		SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory);
		connectionFactory.setExecutor(mock(ExecutorService.class));
		RabbitTemplate template = new RabbitTemplate(connectionFactory);
		template.setChannelTransacted(true);
		TransactionTemplate tt = new TransactionTemplate(new RabbitTransactionManager(connectionFactory));
		assertThatIllegalStateException()
			.isThrownBy(() ->
				tt.execute(status -> {
					template.convertAndSend("foo", "bar");
				return null;
				}));
		assertThat(TransactionSynchronizationManager.hasResource(connectionFactory)).isFalse();
		assertThatIllegalStateException()
			.isThrownBy(() -> (TransactionSynchronizationManager.getSynchronizations()).isEmpty())
			.withMessage("Transaction synchronization is not active");
	}

When running in a debugger, I see the synchronization being cleared in triggerAfterCompletion

	/**
	 * Trigger {@code afterCompletion} callbacks.
	 * @param status object representing the transaction
	 * @param completionStatus completion status according to TransactionSynchronization constants
	 */
	private void triggerAfterCompletion(DefaultTransactionStatus status, int completionStatus) {
		if (status.isNewSynchronization()) {
			List<TransactionSynchronization> synchronizations = TransactionSynchronizationManager.getSynchronizations();
			TransactionSynchronizationManager.clearSynchronization(); <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<
			if (!status.hasTransaction() || status.isNewTransaction()) {
				if (status.isDebug()) {
					logger.trace("Triggering afterCompletion synchronization");
				}
				// No transaction or new transaction for the current scope ->
				// invoke the afterCompletion callbacks immediately
				invokeAfterCompletion(synchronizations, completionStatus);
			}
			else if (!synchronizations.isEmpty()) {
				// Existing transaction that we participate in, controlled outside
				// of the scope of this Spring transaction manager -> try to register
				// an afterCompletion callback with the existing (JTA) transaction.
				registerAfterCompletionWithExistingTransaction(status.getTransaction(), synchronizations);
			}
		}
	}

and the resource removed in

	@Override
	protected void doCleanupAfterCompletion(Object transaction) {
		RabbitTransactionObject txObject = (RabbitTransactionObject) transaction;
		TransactionSynchronizationManager.unbindResource(getConnectionFactory());
		txObject.getResourceHolder().closeAll();
		txObject.getResourceHolder().clear();
	}

[garyrussell]

This test works too - local listener container factory transaction; the resource is unbound (in a finally block) here too

public class MessageListenerContainerTxSynchTests {

	@Test
	void resourcesClearedAfterTxFails() throws IOException, TimeoutException, InterruptedException, ExecutionException {
		ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class);
		Connection mockConnection = mock(Connection.class);
		Channel mockChannel = mock(Channel.class);

		given(mockConnectionFactory.newConnection(any(ExecutorService.class), anyString())).willReturn(mockConnection);
		given(mockConnection.isOpen()).willReturn(true);
		given(mockConnection.createChannel()).willReturn(mockChannel);
		given(mockChannel.txSelect()).willReturn(mock(SelectOk.class));
		given(mockChannel.txCommit()).willThrow(IllegalStateException.class);
		AtomicReference<Consumer> consumer = new AtomicReference<>();
		AtomicReference<String> tag = new AtomicReference<>();
		CountDownLatch latch1 = new CountDownLatch(1);
		willAnswer(inv -> {
			consumer.set(inv.getArgument(6));
			consumer.get().handleConsumeOk(inv.getArgument(2));
			tag.set(inv.getArgument(2));
			latch1.countDown();
			return null;
		}).given(mockChannel).basicConsume(any(), anyBoolean(), any(), anyBoolean(), anyBoolean(), any(), any());
		SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory);
		connectionFactory.setExecutor(mock(ExecutorService.class));
		RabbitTemplate template = new RabbitTemplate(connectionFactory);
		template.setChannelTransacted(true);
		ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
		exec.initialize();
		SimpleMessageListenerContainer mlc = new SimpleMessageListenerContainer();
		mlc.setConnectionFactory(connectionFactory);
		mlc.setQueueNames("foo");
		mlc.setTaskExecutor(exec);
		mlc.setChannelTransacted(true);
		CountDownLatch latch2 = new CountDownLatch(1);
		mlc.setMessageListener(msg -> {
			template.convertAndSend("foo", "bar");
			latch2.countDown();
		});
		mlc.start();
		assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
		consumer.get().handleDelivery(tag.get(), new Envelope(1, false, "", ""), new AMQP.BasicProperties(),
				new byte[0]);
		assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue();
		mlc.stop();
		Future<Exception> exception = exec.submit(() -> {
                         // verify no lingering resources bound to the executor's single thread
			try {
				assertThat(TransactionSynchronizationManager.hasResource(connectionFactory)).isFalse();
				assertThatIllegalStateException()
					.isThrownBy(() -> (TransactionSynchronizationManager.getSynchronizations()).isEmpty())
					.withMessage("Transaction synchronization is not active");
				return null;
			}
			catch (Exception e) {
				return e;
			}
		});
		assertThat(exception.get(10, TimeUnit.SECONDS)).isNull();
		exec.shutdown();
	}

}

[kuldeepkalassts]

As we have seen in the server logs trace, exception was on org.springframework.amqp.rabbit.connection.RabbitResourceHolder.commitAll.

"logger": "org.springframework.transaction.support.TransactionSynchronizationUtils",
  "msg": "TransactionSynchronization.afterCompletion threw exception",
  "ERR": {
    "org.springframework.amqp.AmqpException: failed to commit RabbitMQ transaction": [
      ["RabbitResourceHolder.java", 157, "org.springframework.amqp.rabbit.connection.RabbitResourceHolder.commitAll"],
      ["ConnectionFactoryUtils.java", 333, "org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils$RabbitResourceSynchronization.afterCompletion"],
      ["TransactionSynchronizationUtils.java", 171, "org.springframework.transaction.support.TransactionSynchronizationUtils.invokeAfterCompletion"],
      ["AbstractPlatformTransactionManager.java", 990, "org.springframework.transaction.support.AbstractPlatformTransactionManager.invokeAfterCompletion"],
      ["AbstractPlatformTransactionManager.java", 965, "org.springframework.transaction.support.AbstractPlatformTransactionManager.triggerAfterCompletion"],
      ["AbstractPlatformTransactionManager.java", 786, "org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit"],
      ["AbstractPlatformTransactionManager.java", 712, "org.springframework.transaction.support.AbstractPlatformTransactionManager.commit"],
      ["TransactionAspectSupport.java", 631, "org.springframework.transaction.interceptor.TransactionAspectSupport.commitTransactionAfterReturning"],
      ["TransactionAspectSupport.java", 385, "org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction"],
      ["TransactionInterceptor.java", 99, "org.springframework.transaction.interceptor.TransactionInterceptor.invoke"],
      ["ReflectiveMethodInvocation.java", 186, "org.springframework.aop.framework.ReflectiveMethodInvocation.proceed"],
      ["CglibAopProxy.java", 747, "org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed"],
      ["CglibAopProxy.java", 689, "org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept"],
      ["<generated>", -1, "com.myorg.messaging.impl.daemon.handler.MessageSenderDaemon$$EnhancerBySpringCGLIB$$5dc0bb6c.run"],<<tx
      ["Executors.java", 511, "java.util.concurrent.Executors$RunnableAdapter.call"],
      ["FutureTask.java", 308, "java.util.concurrent.FutureTask.runAndReset"],
      ["ScheduledThreadPoolExecutor.java", 180, "java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301"],
      ["ScheduledThreadPoolExecutor.java", 294, "java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run"],
      ["ThreadPoolExecutor.java", 1149, "java.util.concurrent.ThreadPoolExecutor.runWorker"],
      ["ThreadPoolExecutor.java", 624, "java.util.concurrent.ThreadPoolExecutor$Worker.run"],
      ["Thread.java", 748, "java.lang.Thread.run"]
    ],
    "Caused By": {
      "com.rabbitmq.client.ChannelContinuationTimeoutException: Continuation call for method #method<tx.commit>() on channel AMQChannel(amqp://amqp-server@node_ip:5672/cust01-prd04-ins01-wfm10-fnt.int.prd.mykronos.com,1) (#1) timed out": ["AMQChannel.java", 313, "com.rabbitmq.client.impl.AMQChannel.wrapTimeoutException", "AMQChannel.java", 295, "com.rabbitmq.client.impl.AMQChannel.privateRpc", "AMQChannel.java", 141, "com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc", "ChannelN.java", 1540, "com.rabbitmq.client.impl.ChannelN.txCommit", "ChannelN.java", 52, "com.rabbitmq.client.impl.ChannelN.txCommit", "", -1, "sun.reflect.GeneratedMethodAccessor452.invoke", "DelegatingMethodAccessorImpl.java", 43, "sun.reflect.DelegatingMethodAccessorImpl.invoke", "Method.java", 498, "java.lang.reflect.Method.invoke", "CachingConnectionFactory.java", 1190, "org.springframework.amqp.rabbit.connection.CachingConnectionFactory$CachedChannelInvocationHandler.invoke", "", -1, "com.sun.proxy.$Proxy941.txCommit", "RabbitResourceHolder.java", 153, "org.springframework.amqp.rabbit.connection.RabbitResourceHolder.commitAll"],
      "Caused By": {
        "java.util.concurrent.TimeoutException": ["BlockingCell.java", 77, "com.rabbitmq.utility.BlockingCell.get", "BlockingCell.java", 120, "com.rabbitmq.utility.BlockingCell.uninterruptibleGet", "BlockingValueOrException.java", 36, "com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue", "AMQChannel.java", 502, "com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply", "AMQChannel.java", 293, "com.rabbitmq.client.impl.AMQChannel.privateRpc"]
      }
    }
  }

I checked this code area org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.RabbitResourceSynchronization.afterCompletion(int)

                @Override
		public void afterCompletion(int status) {
			if (status == TransactionSynchronization.STATUS_COMMITTED) {
				this.resourceHolder.commitAll();
			}
			else {
				this.resourceHolder.rollbackAll();
			}

			if (this.resourceHolder.isReleaseAfterCompletion()) {
				this.resourceHolder.setSynchronizedWithTransaction(false);
			}
			super.afterCompletion(status);
		}

And I guess that as exception occurred on this.resourceHolder.commitAll();
because of it, super.afterCompletion(status); never get called. Thus, it was failed to remove the resources.

[garyrussell]

Ah; you are using transaction synchronization, not the RabbitTransactionManager or local container transactions. The full stack trace would have been useful in the original post. In future, please provide as much information as possible.

However, I added another test:

	@Test
	void resourcesClearedAfterTxFailsWithSync() throws IOException, TimeoutException {
		ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class);
		Connection mockConnection = mock(Connection.class);
		Channel mockChannel = mock(Channel.class);

		given(mockConnectionFactory.newConnection(any(ExecutorService.class), anyString())).willReturn(mockConnection);
		given(mockConnection.isOpen()).willReturn(true);
		given(mockConnection.createChannel()).willReturn(mockChannel);
		given(mockChannel.txSelect()).willReturn(mock(SelectOk.class));
		given(mockChannel.txCommit()).willThrow(IllegalStateException.class);
		SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory);
		connectionFactory.setExecutor(mock(ExecutorService.class));
		RabbitTemplate template = new RabbitTemplate(connectionFactory);
		template.setChannelTransacted(true);
		TransactionTemplate tt = new TransactionTemplate(new TestTransactionManager());
		tt.execute(status -> {
			template.convertAndSend("foo", "bar");
			return null;
		});
		assertThat(TransactionSynchronizationManager.hasResource(connectionFactory)).isFalse();
		assertThatIllegalStateException()
			.isThrownBy(() -> (TransactionSynchronizationManager.getSynchronizations()).isEmpty())
			.withMessage("Transaction synchronization is not active");
	}

Again, the synchronizations are cleared in triggerAfterCompletion(), and I see the resource being unbound in triggerBeforeCompletion() because shouldUnbindAtCompletion() always returns true.

So something else must be going on in your case.

I will, however, move that cleanup code to a finally block.

[kuldeepkalassts]

We checked server logs, and at sever log we are seeing this exception at the time of commit failure

Error on AMQP connection <0.14285.4931> (##IP:50562 -> ##IP:5672 - connectionFactory#722cd0a9:3, vhost: '<OUR_URL>', user: '<OUR_URL>', state: running), channel 33: operation tx.rollback caused a connection exception channel_error: "unexpected command while processing 'tx.commit'"

Networking team also told us about some network glitches around that time.

I hope these information could help you to investigate.

[garyrussell]

That is most likely downstream of the com.rabbitmq.client.ChannelContinuationTimeoutException on the txCommit - in the non-synchronization case, the transaction manager will perform a rollback if the commit fails.

I don't see that in the synchronization case, however.