GH-1148: Log4jAppender - fix initialization cycle

Resolves #1148

Defer creation of all Spring static loggers from appender until the
appender is initialized.

The first logger (e.g. Spring Boot) initializes the static `LoggerContext`
field in `LogAdapter.Log4jLog`; this, in turn initializes the appender.

However, the appender references Spring objects that have static loggers.
This causes the `Log4jLog` CTOR to be called before the field is initialized,
causing an NPE.

Defer creation of all static loggers until the appender is fully initialized.

Tested with the log4j2 sample app.

Also add a test for async sends.

**cherry-pick to 2.1.x**

* NOSONAR - inconsistent sync

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java

@@ -123,7 +123,7 @@ public class AmqpAppender extends AbstractAppender {
 	/**
 	 * The template.
 	 */
-	private final RabbitTemplate rabbitTemplate = new RabbitTemplate();
+	private RabbitTemplate rabbitTemplate;
 
 	/**
 	 * Where LoggingEvents are queued to send.
@@ -256,18 +256,12 @@ public static AmqpAppender createAppender(// NOSONAR NCSS line count
 	 * Submit the required number of senders into the pool.
 	 */
 	private void startSenders() {
-		this.rabbitTemplate.setConnectionFactory(this.manager.connectionFactory);
 		if (this.manager.async) {
+			this.manager.senderPool = Executors.newCachedThreadPool();
 			for (int i = 0; i < this.manager.senderPoolSize; i++) {
 				this.manager.senderPool.submit(new EventSender());
 			}
 		}
-		else if (this.manager.maxSenderRetries > 0) {
-			RetryTemplate retryTemplate = new RetryTemplate();
-			RetryPolicy retryPolicy = new SimpleRetryPolicy(this.manager.maxSenderRetries);
-			retryTemplate.setRetryPolicy(retryPolicy);
-			this.rabbitTemplate.setRetryTemplate(retryTemplate);
-		}
 	}
 
 	@Override
@@ -293,6 +287,22 @@ protected Message postProcessMessageBeforeSend(Message message, Event event) {
 	}
 
 	protected void sendEvent(Event event, Map<?, ?> properties) {
+		synchronized (this) {
+			if (this.rabbitTemplate == null) {
+				if (this.manager.activateOptions()) {
+					this.rabbitTemplate = new RabbitTemplate(this.manager.connectionFactory);
+					if (!this.manager.async && this.manager.maxSenderRetries > 0) {
+						RetryTemplate retryTemplate = new RetryTemplate();
+						RetryPolicy retryPolicy = new SimpleRetryPolicy(this.manager.maxSenderRetries);
+						retryTemplate.setRetryPolicy(retryPolicy);
+						this.rabbitTemplate.setRetryTemplate(retryTemplate);
+					}
+				}
+				else {
+					throw new AmqpException("Cannot create template");
+				}
+			}
+		}
 		LogEvent logEvent = event.getEvent();
 		String name = logEvent.getLoggerName();
 		Level level = logEvent.getLevel();
@@ -358,7 +368,7 @@ protected void doSend(Event event, LogEvent logEvent, MessageProperties amqpProp
 				message = new Message(msgBody.toString().getBytes(), amqpProps); //NOSONAR (default charset)
 			}
 			message = postProcessMessageBeforeSend(message, event);
-			this.rabbitTemplate.send(this.manager.exchangeName, routingKey, message);
+			this.rabbitTemplate.send(this.manager.exchangeName, routingKey, message); // NOSONAR (sync)
 		}
 		catch (AmqpException e) {
 			int retries = event.incrementRetries();
@@ -651,7 +661,7 @@ protected AmqpManager(LoggerContext loggerContext, String name) {
 			super(loggerContext, name);
 		}
 
-		private boolean activateOptions() {
+		boolean activateOptions() {
 			ConnectionFactory rabbitConnectionFactory = createRabbitConnectionFactory();
 			if (rabbitConnectionFactory != null) {
 				Assert.state(this.applicationId != null, "applicationId is required");
@@ -673,7 +683,6 @@ private boolean activateOptions() {
 							this.clientConnectionProperties);
 				}
 				setUpExchangeDeclaration();
-				this.senderPool = Executors.newCachedThreadPool();
 				return true;
 			}
 			return false;

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j2/AmqpAppenderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
 
 package org.springframework.amqp.rabbit.log4j2;
 
+import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.startsWith;
@@ -37,6 +38,7 @@
 import java.net.URI;
 import java.util.Map;
 import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 
 import org.apache.logging.log4j.LogManager;
@@ -54,6 +56,7 @@
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean;
+import org.springframework.amqp.rabbit.connection.RabbitUtils;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.junit.BrokerRunning;
@@ -178,55 +181,58 @@ public void testProperties() {
 		assertEquals(ArrayBlockingQueue.class, events.getClass());
 	}
 
-	@Test
-	public void testAmqpAppenderEventQueueTypeDefaultsToLinkedBlockingQueue() throws Exception {
-		Logger logger = LogManager.getLogger("default_queue_logger");
-		AmqpAppender appender = (AmqpAppender) TestUtils.getPropertyValue(logger, "context.configuration.appenders",
-				Map.class).get("rabbitmq_default_queue");
-
-		Object events = TestUtils.getPropertyValue(appender, "events");
-
-		Object manager = TestUtils.getPropertyValue(appender, "manager");
-		assertTrue(TestUtils.getPropertyValue(manager, "addMdcAsHeaders", Boolean.class));
-
-		assertThat(events, instanceOf(LinkedBlockingQueue.class));
-	}
-
 	@Test
 	public void testSaslConfig() {
 		Logger logger = LogManager.getLogger("sasl");
 		AmqpAppender appender = (AmqpAppender) TestUtils.getPropertyValue(logger, "context.configuration.appenders",
 				Map.class).get("sasl1");
-		SaslConfig saslConfig =
-				TestUtils.getPropertyValue(appender, "manager.connectionFactory.rabbitConnectionFactory",
-						ConnectionFactory.class)
-						.getSaslConfig();
+		SaslConfig saslConfig = RabbitUtils.stringToSaslConfig(
+				TestUtils.getPropertyValue(appender, "manager.saslConfig", String.class),
+				mock(ConnectionFactory.class));
 		assertThat(saslConfig, instanceOf(DefaultSaslConfig.class));
 		assertEquals("PLAIN", TestUtils.getPropertyValue(saslConfig, "mechanism"));
 		appender = (AmqpAppender) TestUtils.getPropertyValue(logger, "context.configuration.appenders",
 				Map.class).get("sasl2");
-		saslConfig =
-				TestUtils.getPropertyValue(appender, "manager.connectionFactory.rabbitConnectionFactory",
-						ConnectionFactory.class)
-						.getSaslConfig();
+		saslConfig = RabbitUtils.stringToSaslConfig(
+				TestUtils.getPropertyValue(appender, "manager.saslConfig", String.class),
+				mock(ConnectionFactory.class));
 		assertThat(saslConfig, instanceOf(DefaultSaslConfig.class));
 		assertEquals("EXTERNAL", TestUtils.getPropertyValue(saslConfig, "mechanism"));
 		appender = (AmqpAppender) TestUtils.getPropertyValue(logger, "context.configuration.appenders",
 				Map.class).get("sasl3");
-		saslConfig =
-				TestUtils.getPropertyValue(appender, "manager.connectionFactory.rabbitConnectionFactory",
-						ConnectionFactory.class)
-						.getSaslConfig();
+		saslConfig = RabbitUtils.stringToSaslConfig(
+				TestUtils.getPropertyValue(appender, "manager.saslConfig", String.class),
+				mock(ConnectionFactory.class));
 		assertThat(saslConfig, instanceOf(JDKSaslConfig.class));
 		appender = (AmqpAppender) TestUtils.getPropertyValue(logger, "context.configuration.appenders",
 				Map.class).get("sasl4");
-		saslConfig =
-				TestUtils.getPropertyValue(appender, "manager.connectionFactory.rabbitConnectionFactory",
-						ConnectionFactory.class)
-						.getSaslConfig();
+		saslConfig = RabbitUtils.stringToSaslConfig(
+				TestUtils.getPropertyValue(appender, "manager.saslConfig", String.class),
+				mock(ConnectionFactory.class));
 		assertThat(saslConfig, instanceOf(CRDemoMechanism.CRDemoSaslConfig.class));
 	}
 
+	@Test
+	public void testAmqpAppenderEventQueueTypeDefaultsToLinkedBlockingQueue() throws InterruptedException {
+		Logger logger = LogManager.getLogger("default_queue_logger");
+		logger.info("test");
+		AmqpAppender appender = (AmqpAppender) TestUtils.getPropertyValue(logger, "context.configuration.appenders",
+				Map.class).get("rabbitmq_default_queue");
+
+		Object events = TestUtils.getPropertyValue(appender, "events");
+
+		Object manager = TestUtils.getPropertyValue(appender, "manager");
+		assertTrue(TestUtils.getPropertyValue(manager, "addMdcAsHeaders", Boolean.class));
+
+		assertThat(events.getClass(), equalTo(LinkedBlockingQueue.class));
+		BlockingQueue<?> queue = (BlockingQueue<?>) events;
+		int n = 0;
+		while (n++ < 100 && queue.size() > 0) {
+			Thread.sleep(100);
+		}
+		assertThat(queue.size(), equalTo(0));
+	}
+
 	@Test
 	public void testUriProperties() {
 		Logger logger = LogManager.getLogger("bar");

spring-rabbit/src/test/resources/log4j2-amqp-appender.xml

@@ -14,7 +14,8 @@
 				  connectionName="log4j2Appender"
 				  clientConnectionProperties="foo:bar,baz:qux"
 				  async="false"
-				  senderPoolSize="3" maxSenderRetries="5"
+				  senderPoolSize="3"
+				  maxSenderRetries="5"
 				  bufferSize="10">
 			<ArrayBlockingQueue/>
 		</RabbitMQ>
@@ -23,6 +24,8 @@
 				  host="localhost" port="5672" user="guest" password="guest" applicationId="testAppId" charset="UTF-8"
 				  routingKeyPattern="%X{applicationId}.%c.%p"
 				  exchange="log4j2Test_default_queue" deliveryMode="NON_PERSISTENT"
+				  async="true"
+				  senderPoolSize="2"
 				  addMdcAsHeaders="true">
 		</RabbitMQ>
 		<RabbitMQ name="rabbitmq_uri"