Updated test to assert on log when publish to error topic is failed

pubsub-plus-connector/src/test/java/io/quarkiverse/solace/SolaceConsumerTest.java

@@ -26,10 +26,17 @@
 
 import io.quarkiverse.solace.base.SolaceContainer;
 import io.quarkiverse.solace.base.WeldTestBase;
+import io.quarkiverse.solace.logging.SolaceTestAppender;
 import io.smallrye.reactive.messaging.test.common.config.MapBasedConfig;
 
 @TestMethodOrder(MethodOrderer.OrderAnnotation.class)
 public class SolaceConsumerTest extends WeldTestBase {
+    private org.apache.log4j.Logger rootLogger = org.apache.log4j.Logger.getLogger("io.quarkiverse.solace");
+    private SolaceTestAppender solaceTestAppender = new SolaceTestAppender();
+
+    private SolaceConsumerTest() {
+        rootLogger.addAppender(solaceTestAppender);
+    }
 
     @Test
     @Order(1)
@@ -180,28 +187,6 @@ void consumerFailedProcessingMoveToDMQ() {
 
     @Test
     @Order(6)
-    void consumerCreateMissingResourceAddSubscriptionPermissionException() {
-        MapBasedConfig config = new MapBasedConfig()
-                .with("mp.messaging.incoming.in.connector", "quarkus-solace")
-                .with("mp.messaging.incoming.in.consumer.queue.add-additional-subscriptions", "true")
-                .with("mp.messaging.incoming.in.consumer.queue.missing-resource-creation-strategy", "create-on-start")
-                .with("mp.messaging.incoming.in.consumer.queue.name", SolaceContainer.INTEGRATION_TEST_QUEUE_NAME)
-                .with("mp.messaging.incoming.in.consumer.queue.type", "durable-exclusive")
-                .with("mp.messaging.incoming.in.consumer.queue.subscriptions", topic);
-
-        Exception exception = assertThrows(Exception.class, () -> {
-            // Run app that consumes messages
-            MyConsumer app = runApplication(config, MyConsumer.class);
-        });
-
-        // Assert on published messages
-        await().untilAsserted(() -> assertThat(exception.getMessage())
-                .contains("com.solacesystems.jcsmp.AccessDeniedException: Permission Not Allowed - Queue '"
-                        + SolaceContainer.INTEGRATION_TEST_QUEUE_NAME + "' - Topic '" + topic));
-    }
-
-    @Test
-    @Order(7)
     void consumerPublishToErrorTopicPermissionException() {
         MapBasedConfig config = new MapBasedConfig()
                 .with("mp.messaging.incoming.in.connector", "quarkus-solace")
@@ -210,6 +195,7 @@ void consumerPublishToErrorTopicPermissionException() {
                 .with("mp.messaging.incoming.in.consumer.queue.failure-strategy", "error_topic")
                 .with("mp.messaging.incoming.in.consumer.queue.error.topic",
                         "publish/deny")
+                .with("mp.messaging.incoming.in.consumer.queue.error.message.max-delivery-attempts", 0)
                 .with("mp.messaging.incoming.error-in.connector", "quarkus-solace")
                 .with("mp.messaging.incoming.error-in.consumer.queue.name", SolaceContainer.INTEGRATION_TEST_ERROR_QUEUE_NAME)
                 .with("mp.messaging.incoming.error-in.consumer.queue.type", "durable-exclusive");
@@ -226,6 +212,32 @@ void consumerPublishToErrorTopicPermissionException() {
         publisher.publish(outboundMessage, tp);
 
         await().untilAsserted(() -> assertThat(app.getReceivedFailedMessages().size()).isEqualTo(0));
+        //        await().untilAsserted(() -> assertThat(inMemoryLogHandler.getRecords().stream().filter(record -> record.getMessage().contains("A exception occurred when publishing to topic")).count()).isEqualTo(4));
+        await().untilAsserted(() -> assertThat(solaceTestAppender.getLog().stream()
+                .anyMatch(record -> record.getMessage().toString().contains("Publishing error message to topic")))
+                .isEqualTo(true));
+    }
+
+    @Test
+    @Order(7)
+    void consumerCreateMissingResourceAddSubscriptionPermissionException() {
+        MapBasedConfig config = new MapBasedConfig()
+                .with("mp.messaging.incoming.in.connector", "quarkus-solace")
+                .with("mp.messaging.incoming.in.consumer.queue.add-additional-subscriptions", "true")
+                .with("mp.messaging.incoming.in.consumer.queue.missing-resource-creation-strategy", "create-on-start")
+                .with("mp.messaging.incoming.in.consumer.queue.name", SolaceContainer.INTEGRATION_TEST_QUEUE_NAME)
+                .with("mp.messaging.incoming.in.consumer.queue.type", "durable-exclusive")
+                .with("mp.messaging.incoming.in.consumer.queue.subscriptions", topic);
+
+        Exception exception = assertThrows(Exception.class, () -> {
+            // Run app that consumes messages
+            MyConsumer app = runApplication(config, MyConsumer.class);
+        });
+
+        // Assert on published messages
+        await().untilAsserted(() -> assertThat(exception.getMessage())
+                .contains("com.solacesystems.jcsmp.AccessDeniedException: Permission Not Allowed - Queue '"
+                        + SolaceContainer.INTEGRATION_TEST_QUEUE_NAME + "' - Topic '" + topic));
     }
 
     @ApplicationScoped

pubsub-plus-connector/src/test/java/io/quarkiverse/solace/logging/SolaceTestAppender.java

@@ -0,0 +1,30 @@
+package io.quarkiverse.solace.logging;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.log4j.AppenderSkeleton;
+import org.apache.log4j.spi.LoggingEvent;
+
+public class SolaceTestAppender extends AppenderSkeleton {
+    private List<LoggingEvent> log = new ArrayList<>();
+
+    @Override
+    protected void append(LoggingEvent loggingEvent) {
+        log.add(loggingEvent);
+    }
+
+    @Override
+    public void close() {
+
+    }
+
+    @Override
+    public boolean requiresLayout() {
+        return false;
+    }
+
+    public List<LoggingEvent> getLog() {
+        return new ArrayList<LoggingEvent>(log);
+    }
+}