Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1225: Fix Log4j2 Appender Termination

Browse Source 
Replaces https://github.com/spring-projects/spring-amqp/pull/1225

`manager.stop()` was never called to destroy the connection factory, preventing
JVM exit.

Also protect for re-connecting after stop (both appenders).

Tested with a Spring Boot application.

**cherry-pick to 2.2.x, 2.1.x, 1.7.x**
This commit is contained in:
Gary Russell
2020-07-22 15:42:15 -04:00
parent 11c4367c84
commit 028b28ae17
2 changed files with 24 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -55,13 +55,15 @@ import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean;
import org.springframework.amqp.rabbit.core.DeclareExchangeConnectionListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.LogAppenderUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
@@ -308,6 +310,9 @@ public class AmqpAppender extends AbstractAppender {
message = postProcessMessageBeforeSend(message, event);
this.rabbitTemplate.send(this.manager.exchangeName, routingKey, message);
}
catch (IllegalStateException e) {
getHandler().error("Could not send log message " + logEvent.getMessage() + " appender is stopped");
}
catch (AmqpException e) {
int retries = event.incrementRetries();
if (this.manager.async && retries < this.manager.maxSenderRetries) {
@@ -334,7 +339,7 @@ public class AmqpAppender extends AbstractAppender {
@Override
protected boolean stop(long timeout, TimeUnit timeUnit, boolean changeLifeCycleState) {
boolean stopped = super.stop(timeout, timeUnit, changeLifeCycleState);
return stopped & this.manager.stop(timeout, timeUnit);
return this.manager.stop(timeout, timeUnit) || stopped;
}
/**
@@ -397,6 +402,8 @@ public class AmqpAppender extends AbstractAppender {
protected static class AmqpManager extends AbstractManager {
private final ApplicationContext context = new GenericApplicationContext();
/**
* True to send events on separate threads.
*/
@@ -440,7 +447,7 @@ public class AmqpAppender extends AbstractAppender {
/**
* RabbitMQ ConnectionFactory.
*/
private AbstractConnectionFactory connectionFactory;
private CachingConnectionFactory connectionFactory;
/**
* RabbitMQ host to connect to.
@@ -537,6 +544,7 @@ public class AmqpAppender extends AbstractAppender {
.replaceAll("%X\\{applicationId\\}", this.applicationId),
null, null, null, Charset.forName(this.charset), false, true, null, null);
this.connectionFactory = new CachingConnectionFactory(createRabbitConnectionFactory());
this.connectionFactory.setApplicationContext(this.context);
if (this.addresses != null) {
this.connectionFactory.setAddresses(this.addresses);
}
@@ -588,6 +596,7 @@ public class AmqpAppender extends AbstractAppender {
this.retryTimer.cancel();
this.senderPool.shutdownNow();
this.connectionFactory.destroy();
this.connectionFactory.onApplicationEvent(new ContextClosedEvent(this.context));
try {
return this.senderPool.awaitTermination(timeout, timeUnit);
}

15
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2017 the original author or authors.
* Copyright 2014-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.
@@ -38,12 +38,14 @@ import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.DeclareExchangeConnectionListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.LogAppenderUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.support.GenericApplicationContext;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.PatternLayout;
@@ -104,6 +106,8 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
*/
public static final String THREAD_NAME = "thread";
private final ApplicationContext context = new GenericApplicationContext();
/**
* Name of the exchange to publish log events to.
*/
@@ -162,7 +166,7 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
/**
* RabbitMQ ConnectionFactory.
*/
private AbstractConnectionFactory connectionFactory;
private CachingConnectionFactory connectionFactory;
/**
* Additional client connection properties added to the rabbit connection, with the form
@@ -472,6 +476,7 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
this.connectionFactory.setUsername(this.username);
this.connectionFactory.setPassword(this.password);
this.connectionFactory.setVirtualHost(this.virtualHost);
this.connectionFactory.setApplicationContext(this.context);
LogAppenderUtils.updateClientConnectionProperties(this.connectionFactory, this.clientConnectionProperties);
updateConnectionClientProperties(this.connectionFactory.getRabbitConnectionFactory().getClientProperties());
setUpExchangeDeclaration();
@@ -502,6 +507,7 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
}
if (null != this.connectionFactory) {
this.connectionFactory.destroy();
this.connectionFactory.onApplicationEvent(new ContextClosedEvent(this.context));
}
this.retryTimer.cancel();
this.routingKeyLayout.stop();
@@ -640,6 +646,9 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
message = postProcessMessageBeforeSend(message, event);
rabbitTemplate.send(AmqpAppender.this.exchangeName, routingKey, message);
}
catch (IllegalStateException e) {
addError("Could not send log message " + logEvent.getMessage() + " appender is stopped");
}
catch (AmqpException e) {
int retries = event.incrementRetries();
if (retries < AmqpAppender.this.maxSenderRetries) {