Fix testing queue/queue-rabbitmq - RabbitMQMailQueueTest.deQueueMails - RabbitFluxException #1075

vttranlina · 2022-07-04T03:59:34Z

CI sometimes pass, sometimes fail at org.apache.james.queue.rabbitmq.RabbitMQMailQueueTest

Error logs

reactor.rabbitmq.RabbitFluxException: Not retryable exception, cannot retry

	at reactor.rabbitmq.ExceptionHandlers$SimpleRetryTemplate.retry(ExceptionHandlers.java:125)
	at reactor.rabbitmq.ExceptionHandlers$RetryAcknowledgmentExceptionHandler.accept(ExceptionHandlers.java:143)
	at reactor.rabbitmq.ExceptionHandlers$RetryAcknowledgmentExceptionHandler.accept(ExceptionHandlers.java:130)
	at reactor.rabbitmq.AcknowledgableDelivery.retry(AcknowledgableDelivery.java:130)
	at reactor.rabbitmq.AcknowledgableDelivery.ack(AcknowledgableDelivery.java:64)
	at reactor.rabbitmq.AcknowledgableDelivery.ack(AcknowledgableDelivery.java:76)
	at org.apache.james.queue.rabbitmq.Dequeuer.lambda$ack$4(Dequeuer.java:130)
	at com.github.fge.lambdas.consumers.ThrowingConsumer.accept(ThrowingConsumer.java:22)
	at org.apache.james.queue.rabbitmq.Dequeuer$RabbitMQMailQueueItem.done(Dequeuer.java:75)
	at org.apache.james.queue.api.RawMailQueueItem.done(RawMailQueueItem.java:39)
	at org.apache.james.queue.rabbitmq.RabbitMQMailQueueTest$MailQueueSizeMetricsEnabled.lambda$dequeueMails$15(RabbitMQMailQueueTest.java:655)
	at reactor.core.publisher.MonoCallable.call(MonoCallable.java:92)
	at reactor.core.publisher.FluxSubscribeOnCallable$CallableSubscribeOnSubscription.run(FluxSubscribeOnCallable.java:227)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
	Suppressed: java.lang.Exception: #block terminated with an error
		at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:99)
		at reactor.core.publisher.Flux.blockLast(Flux.java:2645)
		at org.apache.james.queue.rabbitmq.RabbitMQMailQueueTest$MailQueueSizeMetricsEnabled.dequeueMails(RabbitMQMailQueueTest.java:658)
		at org.apache.james.queue.rabbitmq.RabbitMQMailQueueTest$MailQueueSizeMetricsEnabled.browseAndDequeueShouldCombineWellWhenDifferentSlices(RabbitMQMailQueueTest.java:291)
		at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
		at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
		at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
		at java.base/java.lang.reflect.Method.invoke(Method.java:566)
		at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:725)
		at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
		at org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
		at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:149)
		at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:140)
		at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:84)
		at org.junit.jupiter.engine.execution.ExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(ExecutableInvoker.java:115)
		at org.junit.jupiter.engine.execution.ExecutableInvoker.lambda$invoke$0(ExecutableInvoker.java:105)
		at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
		at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
		at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
		at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
		at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:104)
		at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:98)
		at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$7(TestMethodTestDescriptor.java:214)
		at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
		at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeTestMethod(TestMethodTestDescriptor.java:210)
		at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:135)
		at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:66)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:151)
		at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
		at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
		at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
		at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
		at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:41)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
		at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
		at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
		at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
		at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
		at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:41)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
		at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
		at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
		at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
		at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
		at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:41)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
		at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
		at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
		at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
		at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
		at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.submit(SameThreadHierarchicalTestExecutorService.java:35)
		at org.junit.platform.engine.support.hierarchical.HierarchicalTestExecutor.execute(HierarchicalTestExecutor.java:57)
		at org.junit.platform.engine.support.hierarchical.HierarchicalTestEngine.execute(HierarchicalTestEngine.java:54)
		at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:107)
		at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:88)
		at org.junit.platform.launcher.core.EngineExecutionOrchestrator.lambda$execute$0(EngineExecutionOrchestrator.java:54)
		at org.junit.platform.launcher.core.EngineExecutionOrchestrator.withInterceptedStreams(EngineExecutionOrchestrator.java:67)
		at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:52)
		at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:114)
		at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:86)
		at org.junit.platform.launcher.core.DefaultLauncherSession$DelegatingLauncher.execute(DefaultLauncherSession.java:86)
		at org.junit.platform.launcher.core.SessionPerRequestLauncher.execute(SessionPerRequestLauncher.java:53)
		at com.intellij.junit5.JUnit5IdeaTestRunner.startRunnerWithArgs(JUnit5IdeaTestRunner.java:71)
		at com.intellij.rt.junit.IdeaTestRunner$Repeater$1.execute(IdeaTestRunner.java:38)
		at com.intellij.rt.execution.junit.TestsRepeater.repeat(TestsRepeater.java:30)
		at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:35)
		at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:235)
		at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54)
Caused by: com.rabbitmq.client.AlreadyClosedException: channel is already closed due to clean channel shutdown; protocol method: #method<channel.close>(reply-code=200, reply-text=OK, class-id=0, method-id=0)
	at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:258)
	at com.rabbitmq.client.impl.AMQChannel.transmit(AMQChannel.java:427)
	at com.rabbitmq.client.impl.AMQChannel.transmit(AMQChannel.java:421)
	at com.rabbitmq.client.impl.recovery.RecoveryAwareChannelN.basicAck(RecoveryAwareChannelN.java:93)
	at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.basicAck(AutorecoveringChannel.java:439)
	at reactor.rabbitmq.AcknowledgableDelivery.basicAck(AcknowledgableDelivery.java:110)
	at reactor.rabbitmq.AcknowledgableDelivery.ack(AcknowledgableDelivery.java:62)
	... 15 more

… - RabbitFluxException

...ueue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java

…meoutException - Regression: org.apache.james.queue.rabbitmq.RabbitMQMailQueueTest$MailQueueSizeMetricsEnabled.dequeueShouldBeConcurrent (from org.apache.james.queue.rabbitmq.RabbitMQMailQueueTest) Condition with lambda expression in org.apache.james.queue.api.MailQueueContract that uses java.util.concurrent.ConcurrentLinkedDeque, java.util.concurrent.ConcurrentLinkedDequeint was not fulfilled within 1 minutes.

quantranhong1999 · 2022-07-04T09:37:04Z

...ueue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java

-                }).subscribeOn(Schedulers.elastic()))
+                }).subscribeOn(Schedulers.newBoundedElastic(2, 20, "DeQueueSchedulerTest")))


Failing test percent (Mean of 3)
Before: 7 %
After: 2.333 %

It would be better if we can understand the root cause IMO

Have you tried single scheduler?

Have you tried single scheduler?

Schedulers.single() -> .block is not allowed on single scheduler.

reduce newBoundedElastic thread size to 1 -> the failing probability seems reduce a bit. about <= 2%

2 will be easier to detect the parallel vulnerable. Furthermore here is the test code, IMO, no need to be too strict 1 or 2

I added some tests for easier debugging issues with reactor-rabbitmq (maybe?)
See d40cdea

Can you see in these tests if using boundedElastic/single helps?

Open a ticket upstream too maybe.

Open a ticket upstream too maybe.

Where I can? https://issues.apache.org/jira/projects/JAMES ?

Can you see in these tests if using boundedElastic/single helps?

single: No, it fails as immediately

boundedElastic: it makes the % fail decreasingly

quantranhong1999 · 2022-07-04T09:58:56Z

server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueContract.java

+        Scheduler newBoundedElastic = Schedulers.newBoundedElastic(10 * Runtime.getRuntime().availableProcessors(), 10, "MailQueueTest");
+


Can you explain this?

#1075 (comment)

chibenwa

I do not see written answers to quan questions.

I do not understand the rationnals behind this.

Maybe the test should just be disabled?

vttranlina · 2022-07-07T02:03:17Z

I do not see written answers to quan questions.

I do not understand the rationnals behind this.

Maybe the test should just be disabled?

I'm not sure for threads deeper here, I guess the creation threads as "elastic" make test cases sometimes fail (rarely).
So When I tried to use boundedElastic threads, I saw the test result is better. (meaning of test case still does not change).

…anual ack

vttranlina · 2022-07-08T10:19:55Z

I added some tests for easier debugging issues with reactor-rabbitmq (maybe?)
See d40cdea

chibenwa · 2022-07-09T05:43:59Z

My feeling is that this could explain crashing consumers cc @Arsnael

Arsnael · 2022-07-11T04:09:00Z

The crashing consumers we have sometimes in prod? Could be yes...

chibenwa · 2022-07-11T10:15:27Z

Does usenio in rabbitmq client setting affects this?

chibenwa · 2022-08-01T02:06:28Z

CF reactor/reactor-rabbitmq#176

chibenwa · 2022-08-01T02:58:13Z

See #1099

Fix testing queue/queue-rabbitmq - RabbitMQMailQueueTest.deQueueMails…

7e309ae

… - RabbitFluxException

Arsnael reviewed Jul 4, 2022

View reviewed changes

...ueue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java Show resolved Hide resolved

Arsnael approved these changes Jul 4, 2022

View reviewed changes

quantranhong1999 reviewed Jul 4, 2022

View reviewed changes

quantranhong1999 approved these changes Jul 5, 2022

View reviewed changes

chibenwa requested changes Jul 5, 2022

View reviewed changes

rabbitmq - Add unit test for concurrent test consuming message with m…

d40cdea

…anual ack

vttranlina mentioned this pull request Jul 14, 2022

James 3779 Tasks are doing sneaky blocking calls #1049

Merged

chibenwa mentioned this pull request Aug 1, 2022

Fix testing queue/queue-rabbitmq - RabbitMQMailQueueTest.deQueueMails - RabbitFluxException #1099

Merged

chibenwa closed this Aug 1, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix testing queue/queue-rabbitmq - RabbitMQMailQueueTest.deQueueMails - RabbitFluxException #1075

Fix testing queue/queue-rabbitmq - RabbitMQMailQueueTest.deQueueMails - RabbitFluxException #1075

vttranlina commented Jul 4, 2022

quantranhong1999 Jul 4, 2022

quantranhong1999 Jul 4, 2022

chibenwa Jul 5, 2022

quantranhong1999 Jul 7, 2022

vttranlina Jul 7, 2022

vttranlina Jul 8, 2022

chibenwa Jul 9, 2022

chibenwa Jul 9, 2022

vttranlina Jul 11, 2022

vttranlina Jul 11, 2022

quantranhong1999 Jul 4, 2022

vttranlina Jul 7, 2022

chibenwa left a comment

vttranlina commented Jul 7, 2022 •

edited

Loading

vttranlina commented Jul 8, 2022

chibenwa commented Jul 9, 2022

Arsnael commented Jul 11, 2022

chibenwa commented Jul 11, 2022

chibenwa commented Aug 1, 2022

chibenwa commented Aug 1, 2022

		}).subscribeOn(Schedulers.elastic()))
		}).subscribeOn(Schedulers.newBoundedElastic(2, 20, "DeQueueSchedulerTest")))

		Scheduler newBoundedElastic = Schedulers.newBoundedElastic(10 * Runtime.getRuntime().availableProcessors(), 10, "MailQueueTest");

Fix testing queue/queue-rabbitmq - RabbitMQMailQueueTest.deQueueMails - RabbitFluxException #1075

Fix testing queue/queue-rabbitmq - RabbitMQMailQueueTest.deQueueMails - RabbitFluxException #1075

Conversation

vttranlina commented Jul 4, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chibenwa left a comment

Choose a reason for hiding this comment

vttranlina commented Jul 7, 2022 • edited Loading

vttranlina commented Jul 8, 2022

chibenwa commented Jul 9, 2022

Arsnael commented Jul 11, 2022

chibenwa commented Jul 11, 2022

chibenwa commented Aug 1, 2022

chibenwa commented Aug 1, 2022

vttranlina commented Jul 7, 2022 •

edited

Loading