Fix rabbitmq container deprecations and test code cleanup. (#619)

* Refactor mutable static state. * Use RabbitMQContainer API to replace deprecated use. * Use latest official rabbitmq Docker container. * Share rabbit version between tests * nit: formatting * Use RabbitContainer and abhor statics * Don't use Hamcrest * polling conditions need asserts * Fix tests for rabbit 3.9+ and improve logging a bit * Remove deprecated container and use same rabbit version * Use the same rabbit version for all tests --------- Co-authored-by: Tim Yates <[email protected]>
micronaut-projects · Apr 16, 2024 · 3e06a61 · 3e06a61
1 parent c072219
commit 3e06a61
Show file tree

Hide file tree

Showing 6 changed files with 71 additions and 66 deletions.
diff --git a/rabbitmq/src/test/groovy/io/micronaut/rabbitmq/AbstractRabbitMQClusterTest.groovy b/rabbitmq/src/test/groovy/io/micronaut/rabbitmq/AbstractRabbitMQClusterTest.groovy
@@ -24,36 +24,27 @@ abstract class AbstractRabbitMQClusterTest extends Specification {
     private static final Logger log = LoggerFactory.getLogger(AbstractRabbitMQClusterTest)
 
     private static final int AMQP_PORT = 5672
-    private static final DockerImageName RABBIT_IMAGE = DockerImageName.parse("rabbitmq:3.8-management")
-    private static final String CLUSTER_COOKIE = "test-cluster"
-    private static final String RABBIT_CONFIG_PATH = ClassLoader.getSystemResource("rabbit/rabbitmq.conf").path
-    private static final String RABBIT_DEFINITIONS_PATH = ClassLoader.getSystemResource("rabbit/definitions.json").path
+    private static final String DOCKER_IMAGE_NAME = "rabbitmq:$AbstractRabbitMQTest.RABBIT_CONTAINER_VERSION-management"
 
     protected ApplicationContext applicationContext
 
-    static final String EXCHANGE = "test-exchange"
-    static final String QUEUE = "test-durable-queue"
-
     @Shared
     @AutoCleanup
     Network mqClusterNet = Network.newNetwork()
 
     @Shared
     @AutoCleanup
-    RabbitMQContainer node1 = new RabbitMQContainer(RABBIT_IMAGE)
+    RabbitMQContainer node1 = new RabbitMQContainer(DockerImageName.parse(DOCKER_IMAGE_NAME))
 
     @Shared
     @AutoCleanup
-    RabbitMQContainer node2 = new RabbitMQContainer(RABBIT_IMAGE)
+    RabbitMQContainer node2 = new RabbitMQContainer(DockerImageName.parse(DOCKER_IMAGE_NAME))
 
     @Shared
     @AutoCleanup
-    RabbitMQContainer node3 = new RabbitMQContainer(RABBIT_IMAGE)
+    RabbitMQContainer node3 = new RabbitMQContainer(DockerImageName.parse(DOCKER_IMAGE_NAME))
 
     def setupSpec() {
-        log.info("rabbit.conf path: {}", RABBIT_CONFIG_PATH)
-        log.info("rabbit definitions path: {}", RABBIT_DEFINITIONS_PATH)
-
         configureContainer(node1, "rabbitmq1")
                 .waitingFor(Wait.forHealthcheck().withStartupTimeout(Duration.ofMinutes(1)))
                 .start()
@@ -95,7 +86,8 @@ abstract class AbstractRabbitMQClusterTest extends Specification {
                 "spec.name"                  : getClass().simpleName,
                 "rabbitmq.servers.node1.port": node1Port,
                 "rabbitmq.servers.node2.port": node2Port,
-                "rabbitmq.servers.node3.port": node3Port] << additionalConfig
+                "rabbitmq.servers.node3.port": node3Port
+        ] << additionalConfig
 
         log.info("context properties: {}", properties)
         applicationContext = ApplicationContext.run(properties, "test")
@@ -106,10 +98,16 @@ abstract class AbstractRabbitMQClusterTest extends Specification {
     }
 
     private configureContainer(RabbitMQContainer mqContainer, String hostname) {
+        String rabbitConfigPath = ClassLoader.getSystemResource("rabbit/rabbitmq.conf").path
+        log.info("rabbit.conf path: {}", rabbitConfigPath)
+
+        String rabbitDefinitionsPath = ClassLoader.getSystemResource("rabbit/definitions.json").path
+        log.info("rabbit definitions.json path: {}", rabbitDefinitionsPath)
+
         mqContainer
-                .withEnv("RABBITMQ_ERLANG_COOKIE", CLUSTER_COOKIE)
-                .withCopyFileToContainer(forHostPath(RABBIT_CONFIG_PATH), "/etc/rabbitmq/rabbitmq.conf")
-                .withCopyFileToContainer(forHostPath(RABBIT_DEFINITIONS_PATH), "/etc/rabbitmq/definitions.json")
+                .withEnv("RABBITMQ_ERLANG_COOKIE", "test-cluster")
+                .withRabbitMQConfigSysctl(forHostPath(rabbitConfigPath))
+                .withCopyFileToContainer(forHostPath(rabbitDefinitionsPath), "/etc/rabbitmq/definitions.json")
                 .withNetwork(mqClusterNet)
                 .withLogConsumer(new Slf4jLogConsumer(log).withPrefix(hostname))
                 .withCreateContainerCmdModifier(cmd -> cmd

diff --git a/rabbitmq/src/test/groovy/io/micronaut/rabbitmq/AbstractRabbitMQTest.groovy b/rabbitmq/src/test/groovy/io/micronaut/rabbitmq/AbstractRabbitMQTest.groovy
@@ -1,19 +1,21 @@
 package io.micronaut.rabbitmq
 
 import io.micronaut.context.ApplicationContext
-import org.testcontainers.containers.GenericContainer
-import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy
+import org.testcontainers.containers.RabbitMQContainer
+import spock.lang.AutoCleanup
+import spock.lang.Shared
 import spock.lang.Specification
 import spock.util.concurrent.PollingConditions
 
 abstract class AbstractRabbitMQTest extends Specification {
 
-    static GenericContainer rabbitContainer =
-            new GenericContainer("library/rabbitmq:3.7")
-                    .withExposedPorts(5672)
-                    .waitingFor(new LogMessageWaitStrategy().withRegEx("(?s).*Server startup complete.*"))
+    static String RABBIT_CONTAINER_VERSION = "3.13.1"
 
-    static {
+    @Shared
+    @AutoCleanup
+    RabbitMQContainer rabbitContainer = new RabbitMQContainer("rabbitmq:" + RABBIT_CONTAINER_VERSION)
+
+    def setupSpec() {
         rabbitContainer.start()
     }
 

diff --git a/rabbitmq/src/test/groovy/io/micronaut/rabbitmq/connect/MultipleServersSpec.groovy b/rabbitmq/src/test/groovy/io/micronaut/rabbitmq/connect/MultipleServersSpec.groovy
@@ -3,8 +3,8 @@ package io.micronaut.rabbitmq.connect
 import com.rabbitmq.client.Connection
 import io.micronaut.context.ApplicationContext
 import io.micronaut.inject.qualifiers.Qualifiers
-import org.testcontainers.containers.GenericContainer
-import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy
+import io.micronaut.rabbitmq.AbstractRabbitMQTest
+import org.testcontainers.containers.RabbitMQContainer
 import spock.lang.Specification
 
 import java.time.Duration
@@ -13,24 +13,21 @@ class MultipleServersSpec extends Specification {
 
     void "test multiple server configuration"() {
         given:
-        GenericContainer rabbit1 = new GenericContainer("library/rabbitmq:3.7")
-                .withExposedPorts(5672)
-                .waitingFor(new LogMessageWaitStrategy().withRegEx("(?s).*Server startup complete.*"))
-        GenericContainer rabbit2 = new GenericContainer("library/rabbitmq:3.7")
-                .withExposedPorts(5672)
-                .waitingFor(new LogMessageWaitStrategy().withRegEx("(?s).*Server startup complete.*"))
+        RabbitMQContainer rabbit1 = new RabbitMQContainer("rabbitmq:" + AbstractRabbitMQTest.RABBIT_CONTAINER_VERSION)
+        RabbitMQContainer rabbit2 = new RabbitMQContainer("rabbitmq:" + AbstractRabbitMQTest.RABBIT_CONTAINER_VERSION)
 
         when:
         rabbit1.start()
         rabbit2.start()
-        ApplicationContext context = ApplicationContext.run(
-                ["spec.name": getClass().simpleName,
-                 "rabbitmq.servers.one.uri": "amqp://localhost:${rabbit1.getMappedPort(5672)}",
-                 "rabbitmq.servers.one.channel-pool.max-idle-channels": "10",
-                 "rabbitmq.servers.one.rpc.timeout": "10s",
-                 "rabbitmq.servers.two.uri": "amqp://localhost:${rabbit2.getMappedPort(5672)}",
-                 "rabbitmq.servers.two.channel-pool.max-idle-channels": "20",
-                 "rabbitmq.servers.two.rpc.timeout": "20s"])
+        ApplicationContext context = ApplicationContext.run([
+                "spec.name"                                          : getClass().simpleName,
+                "rabbitmq.servers.one.uri"                           : "amqp://localhost:${rabbit1.getMappedPort(5672)}",
+                "rabbitmq.servers.one.channel-pool.max-idle-channels": "10",
+                "rabbitmq.servers.one.rpc.timeout"                   : "10s",
+                "rabbitmq.servers.two.uri"                           : "amqp://localhost:${rabbit2.getMappedPort(5672)}",
+                "rabbitmq.servers.two.channel-pool.max-idle-channels": "20",
+                "rabbitmq.servers.two.rpc.timeout"                   : "20s"
+        ])
 
         then:
         context.getBean(RabbitConnectionFactoryConfig, Qualifiers.byName("one")).port == rabbit1.getMappedPort(5672)

diff --git a/rabbitmq/src/test/groovy/io/micronaut/rabbitmq/connect/TemporarilyDownConsumersSpec.groovy b/rabbitmq/src/test/groovy/io/micronaut/rabbitmq/connect/TemporarilyDownConsumersSpec.groovy
@@ -5,30 +5,31 @@ import com.rabbitmq.client.ShutdownListener
 import io.micronaut.context.ApplicationContext
 import io.micronaut.context.annotation.Requires
 import io.micronaut.core.io.socket.SocketUtils
+import io.micronaut.rabbitmq.AbstractRabbitMQTest
 import io.micronaut.rabbitmq.annotation.Binding
 import io.micronaut.rabbitmq.annotation.Queue
 import io.micronaut.rabbitmq.annotation.RabbitClient
 import io.micronaut.rabbitmq.annotation.RabbitListener
-import io.micronaut.rabbitmq.connect.recovery.TemporarilyDownException
 import io.micronaut.rabbitmq.connect.recovery.TemporarilyDownIOException
 import io.micronaut.rabbitmq.connect.recovery.TemporarilyDownRuntimeException
 import io.micronaut.rabbitmq.exception.RabbitClientException
 import io.micronaut.rabbitmq.exception.RabbitListenerException
 import io.micronaut.rabbitmq.exception.RabbitListenerExceptionHandler
-import org.testcontainers.containers.FixedHostPortGenericContainer
-import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy
+import org.testcontainers.containers.RabbitMQContainer
 import spock.lang.Specification
 import spock.util.concurrent.PollingConditions
 
 class TemporarilyDownConsumersSpec extends Specification {
 
-    static PollingConditions conditions = new PollingConditions(timeout: 10)
+    PollingConditions conditions = new PollingConditions(timeout: 10)
 
     void "test temporarily down consumer"() {
+        given:
         int port = SocketUtils.findAvailableTcpPort()
-        FixedHostPortGenericContainer rabbitmq = new FixedHostPortGenericContainer("library/rabbitmq:3.7")
-                .withFixedExposedPort(port,5672)
-                .waitingFor(new LogMessageWaitStrategy().withRegEx("(?s).*Server startup complete.*"))
+        RabbitMQContainer rabbitmq = new RabbitMQContainer("rabbitmq:" + AbstractRabbitMQTest.RABBIT_CONTAINER_VERSION).tap {
+            portBindings = ["$port:5672".toString()]
+        }
+
         ApplicationContext applicationContext = ApplicationContext.run(["rabbitmq.port": port, "spec.name": "TemporarilyDownConsumersSpec"], "test")
 
         when: "consumer is instantiated when the server is down"
@@ -56,23 +57,24 @@ class TemporarilyDownConsumersSpec extends Specification {
     }
 
     void "test temporarily down producer"() {
+        given:
         int port = SocketUtils.findAvailableTcpPort()
-        FixedHostPortGenericContainer rabbitmq = new FixedHostPortGenericContainer("library/rabbitmq:3.7")
-                .withFixedExposedPort(port,5672)
-                .waitingFor(new LogMessageWaitStrategy().withRegEx("(?s).*Server startup complete.*"))
+        RabbitMQContainer rabbitmq = new RabbitMQContainer("rabbitmq:" + AbstractRabbitMQTest.RABBIT_CONTAINER_VERSION).tap {
+            portBindings = ["$port:5672".toString()]
+        }
         ApplicationContext applicationContext = ApplicationContext.run(["rabbitmq.port": port, "spec.name": "TemporarilyDownConsumersSpec"], "test")
 
         when: "producer publishes a message when the server is down"
         applicationContext.getBean(MyProducer).send("hello")
 
         then: "a temporarily down exception is thrown"
-        var exception = thrown(RabbitClientException)
+        def exception = thrown(RabbitClientException)
         exception.message == 'Failed to publish a message with exchange: [] routing key [eventually-up] and mandatory flag [false]'
         exception.cause.message == 'Failed to retrieve a channel from the pool'
         exception.cause.cause.message == 'Connection is not ready yet'
 
         when: "the temporarily down connection is retrieved from the exception"
-        var connection = ((TemporarilyDownException) exception.cause.cause).connection
+        def connection = exception.cause.cause.connection
         then: "the connection is still down"
         connection.stillDown
 
@@ -240,7 +242,7 @@ class TemporarilyDownConsumersSpec extends Specification {
         then: "details can be retrieved from connection"
         connection.check()
         wasNotified
-        connection.stillDown == false
+        !connection.stillDown
         connection.address.hostName == 'localhost'
         connection.port == port
         connection.channelMax == 2047
@@ -278,7 +280,7 @@ class TemporarilyDownConsumersSpec extends Specification {
         when: "close connection"
         connection.close(-1, 'testing')
         then: "we can retrieve details"
-        connection.open == false
+        !connection.open
         connection.closeReason.reason.replyText == 'testing'
 
         when: "abort connection"
@@ -297,6 +299,7 @@ class TemporarilyDownConsumersSpec extends Specification {
     @Requires(property = "spec.name", value = "TemporarilyDownConsumersSpec")
     @RabbitClient
     static interface MyProducer {
+
         @Binding("eventually-up")
         void send(String message)
     }

diff --git a/rabbitmq/src/test/groovy/io/micronaut/rabbitmq/health/RabbitHealthIndicatorSpec.groovy b/rabbitmq/src/test/groovy/io/micronaut/rabbitmq/health/RabbitHealthIndicatorSpec.groovy
@@ -20,7 +20,7 @@ class RabbitHealthIndicatorSpec extends AbstractRabbitMQTest {
 
         then:
         result.status == UP
-        ((Map<String, Object>) result.details).version.toString().startsWith("3.7")
+        ((Map<String, Object>) result.details).version.toString() == RABBIT_CONTAINER_VERSION
     }
 
     void "test rabbitmq health indicator with 2 connections"() {
@@ -37,8 +37,8 @@ class RabbitHealthIndicatorSpec extends AbstractRabbitMQTest {
         then:
         result.status == UP
         Map<String, List> details = result.details
-        details.get("connections")[0].get("version").toString().startsWith("3.7")
-        details.get("connections")[1].get("version").toString().startsWith("3.7")
+        details.get("connections")[0].get("version").toString() == RABBIT_CONTAINER_VERSION
+        details.get("connections")[1].get("version").toString() == RABBIT_CONTAINER_VERSION
     }
 
     void "test rabbitmq health indicator shows down"() {

diff --git a/rabbitmq/src/test/groovy/io/micronaut/rabbitmq/listener/ConsumerRecoverySpec.groovy b/rabbitmq/src/test/groovy/io/micronaut/rabbitmq/listener/ConsumerRecoverySpec.groovy
@@ -24,13 +24,14 @@ import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.TimeoutException
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS
-import static org.hamcrest.MatcherAssert.assertThat
-import static org.hamcrest.Matchers.equalTo
 
 class ConsumerRecoverySpec extends AbstractRabbitMQClusterTest {
 
     private static final Logger log = LoggerFactory.getLogger(ConsumerRecoverySpec)
 
+    static final String EXCHANGE = "test-exchange"
+    static final String QUEUE = "test-durable-queue"
+
     @Shared
     private ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor()
 
@@ -42,7 +43,7 @@ class ConsumerRecoverySpec extends AbstractRabbitMQClusterTest {
 
     def setupSpec() {
         /*
-         * The current Micronaut publisher implementation has a flaw in detecting unroutable drop/return messages
+         * The current Micronaut publisher implementation has a flaw in detecting non-routable drop/return messages
          * in a Rabbit cluster setup. It considers the messages as published even if the broker did not enqueue it.
          * So for this test a simple custom publisher is used that detects unpublished messages.
          */
@@ -74,12 +75,17 @@ class ConsumerRecoverySpec extends AbstractRabbitMQClusterTest {
                     ch.basicPublish(EXCHANGE, "", true,
                             new AMQP.BasicProperties.Builder().deliveryMode(2).build(),
                             msg.bytes)
+                    // Since rabbit 3.9.x this does not seem to throw an exception in these tests
                     if (ch.waitForConfirms(1000)) {
-                        log.info("publish ack")
+                        log.info("publish ack {}", msg)
+                    } else {
+                        // If the message is not confirmed, it is considered not published
+                        publishedMessages.remove(msg)
+                        log.warn("publish nack {}", msg)
                     }
                 } catch (IOException | RuntimeException | InterruptedException | TimeoutException e) {
                     publishedMessages.remove(msg)
-                    log.error("failed to publish: {}", e.message)
+                    log.error("failed to publish {}: {}", msg, e.message)
                 }
             }, 500, 500, MILLISECONDS))
     }
@@ -208,8 +214,7 @@ class ConsumerRecoverySpec extends AbstractRabbitMQClusterTest {
         enablePublisher = false
 
         new PollingConditions(timeout: 60).eventually {
-            assertThat "all published messages must be consumed",
-                    publishedMessages, equalTo(consumer.consumedMessages)
+            assert publishedMessages == consumer.consumedMessages
         }
     }
 
@@ -219,7 +224,7 @@ class ConsumerRecoverySpec extends AbstractRabbitMQClusterTest {
         container.execInContainer("rabbitmqctl", "start_app")
 
         new PollingConditions(timeout: 60).eventually {
-            container.isHealthy()
+            assert container.isHealthy()
         }
         log.info("started container: {}", container.containerId)
     }
@@ -232,7 +237,7 @@ class ConsumerRecoverySpec extends AbstractRabbitMQClusterTest {
 
         RabbitListenerException lastException
 
-        @Queue(AbstractRabbitMQClusterTest.QUEUE)
+        @Queue(QUEUE)
         void handleMessage(@MessageBody String body) {
             consumedMessages << body
             log.info("{} received: {}", consumedMessages.size(), body)