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
committed by Artem Bilan
parent e6ae6ccb32
commit f04aa83d83
2 changed files with 24 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -52,6 +52,7 @@ import org.apache.logging.log4j.core.config.plugins.PluginElement;
import org.apache.logging.log4j.core.layout.PatternLayout;
import org.apache.logging.log4j.core.util.Integers;
import org.springframework.amqp.AmqpApplicationContextClosedException;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
@@ -71,6 +72,9 @@ import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
import org.springframework.amqp.utils.JavaUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.retry.RetryPolicy;
@@ -272,6 +276,9 @@ public class AmqpAppender extends AbstractAppender {
message = postProcessMessageBeforeSend(message, event);
this.rabbitTemplate.send(this.manager.exchangeName, routingKey, message);
}
catch (AmqpApplicationContextClosedException 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) {
@@ -298,7 +305,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;
}
/**
@@ -367,6 +374,8 @@ public class AmqpAppender extends AbstractAppender {
private static final int DEFAULT_MAX_SENDER_RETRIES = 30;
private final ApplicationContext context = new GenericApplicationContext();
/**
* True to send events on separate threads.
*/
@@ -574,6 +583,7 @@ public class AmqpAppender extends AbstractAppender {
.withNoConsoleNoAnsi(true)
.build();
this.connectionFactory = new CachingConnectionFactory(rabbitConnectionFactory);
this.connectionFactory.setApplicationContext(this.context);
if (StringUtils.hasText(this.connectionName)) {
this.connectionFactory.setConnectionNameStrategy(cf -> this.connectionName);
}
@@ -663,6 +673,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);
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 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.
@@ -32,6 +32,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.amqp.AmqpApplicationContextClosedException;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
@@ -51,6 +52,9 @@ import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
import org.springframework.amqp.utils.JavaUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.util.StringUtils;
@@ -120,6 +124,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.
*/
@@ -682,6 +688,7 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
this.locationLayout.setContext(getContext());
this.locationLayout.start();
this.connectionFactory = new CachingConnectionFactory(rabbitConnectionFactory);
this.connectionFactory.setApplicationContext(this.context);
if (StringUtils.hasText(this.connectionName)) {
this.connectionFactory.setConnectionNameStrategy(cf -> this.connectionName);
}
@@ -797,6 +804,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();
@@ -956,6 +964,9 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
message = postProcessMessageBeforeSend(message, event);
rabbitTemplate.send(AmqpAppender.this.exchangeName, routingKey, message);
}
catch (AmqpApplicationContextClosedException e) {
addError("Could not send log message " + logEvent.getMessage() + " appender is stopped");
}
catch (AmqpException e) {
int retries = event.incrementRetries();
if (retries < AmqpAppender.this.maxSenderRetries) {