This is an automated email from the ASF dual-hosted git repository.
dakirily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-broker-j.git
The following commit(s) were added to refs/heads/main by this push:
new ef9735807b QPID-8771: [Broker-J] Isolate socket log MDC and capture
caller data (#454)
ef9735807b is described below
commit ef9735807b2de691ad3e99de8f8904b7bca8b1e2
Author: usugumo <[email protected]>
AuthorDate: Tue Oct 20 19:13:57 2026 +0200
QPID-8771: [Broker-J] Isolate socket log MDC and capture caller data (#454)
---
.../logback/BrokerLogbackSocketLoggerImpl.java | 37 +-
.../logback/MdcEnrichingSocketAppender.java | 72 ++++
.../server/logging/logback/MdcLoggingEvent.java | 165 ++++++++
.../logback/MdcEnrichingSocketAppenderTest.java | 436 +++++++++++++++++++++
.../logging/logback/MdcLoggingEventTest.java | 71 ++++
5 files changed, 748 insertions(+), 33 deletions(-)
diff --git
a/broker-plugins/logging-logback/src/main/java/org/apache/qpid/server/logging/logback/BrokerLogbackSocketLoggerImpl.java
b/broker-plugins/logging-logback/src/main/java/org/apache/qpid/server/logging/logback/BrokerLogbackSocketLoggerImpl.java
index ccaa7f8446..b4ced2569d 100644
---
a/broker-plugins/logging-logback/src/main/java/org/apache/qpid/server/logging/logback/BrokerLogbackSocketLoggerImpl.java
+++
b/broker-plugins/logging-logback/src/main/java/org/apache/qpid/server/logging/logback/BrokerLogbackSocketLoggerImpl.java
@@ -21,11 +21,9 @@
package org.apache.qpid.server.logging.logback;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import ch.qos.logback.classic.net.SocketAppender;
@@ -35,7 +33,6 @@ import ch.qos.logback.core.Context;
import ch.qos.logback.core.util.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.slf4j.MDC;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.ManagedAttributeField;
@@ -62,7 +59,7 @@ public class BrokerLogbackSocketLoggerImpl
private boolean _includeCallerData;
@ManagedAttributeField
- private Map<String,String> _mappedDiagnosticContext;
+ private volatile Map<String, String> _mappedDiagnosticContext;
@ManagedAttributeField
private Map<String, String> _contextProperties;
@@ -110,11 +107,11 @@ public class BrokerLogbackSocketLoggerImpl
}
@Override
- protected Appender<ILoggingEvent> createAppenderInstance(Context
loggerContext)
+ protected Appender<ILoggingEvent> createAppenderInstance(final Context
loggerContext)
{
if (_contextProperties != null && !_contextProperties.isEmpty())
{
- for (Map.Entry<String, String> property :
_contextProperties.entrySet())
+ for (final Map.Entry<String, String> property :
_contextProperties.entrySet())
{
final String key = property.getKey();
final String value = property.getValue();
@@ -132,33 +129,7 @@ public class BrokerLogbackSocketLoggerImpl
}
}
- SocketAppender socketAppender = new SocketAppender()
- {
- @Override
- protected void append(final
ILoggingEvent event)
- {
- Set<String> keys = new
HashSet<>();
- try
- {
- for (Map.Entry<String,
String> entry : _mappedDiagnosticContext.entrySet())
- {
-
MDC.put(entry.getKey(), entry.getValue());
-
keys.add(entry.getKey());
- }
-
- // Workaround for
suspected Logback defect LOGBACK-1088
-
event.prepareForDeferredProcessing();
- super.append(event);
- }
- finally
- {
- for (String key : keys)
- {
- MDC.remove(key);
- }
- }
- }
- };
+ final SocketAppender socketAppender = new
MdcEnrichingSocketAppender(this::getMappedDiagnosticContext);
socketAppender.setPort(_port);
socketAppender.setRemoteHost(_remoteHost);
socketAppender.setIncludeCallerData(_includeCallerData);
diff --git
a/broker-plugins/logging-logback/src/main/java/org/apache/qpid/server/logging/logback/MdcEnrichingSocketAppender.java
b/broker-plugins/logging-logback/src/main/java/org/apache/qpid/server/logging/logback/MdcEnrichingSocketAppender.java
new file mode 100644
index 0000000000..565a799e82
--- /dev/null
+++
b/broker-plugins/logging-logback/src/main/java/org/apache/qpid/server/logging/logback/MdcEnrichingSocketAppender.java
@@ -0,0 +1,72 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.server.logging.logback;
+
+import java.io.Serializable;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+import ch.qos.logback.classic.net.LoggingEventPreSerializationTransformer;
+import ch.qos.logback.classic.net.SocketAppender;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.classic.spi.LoggingEventVO;
+import ch.qos.logback.core.spi.PreSerializationTransformer;
+
+class MdcEnrichingSocketAppender extends SocketAppender
+{
+ private static final PreSerializationTransformer<ILoggingEvent>
EVENT_TRANSFORMER =
+ new LoggingEventPreSerializationTransformer()
+ {
+ @Override
+ public Serializable transform(final ILoggingEvent event)
+ {
+ return event instanceof MdcLoggingEvent ?
LoggingEventVO.build(event) : super.transform(event);
+ }
+ };
+
+ private final Supplier<Map<String, String>> _mdcSupplier;
+
+ MdcEnrichingSocketAppender(final Supplier<Map<String, String>> mdcSupplier)
+ {
+ _mdcSupplier = Objects.requireNonNull(mdcSupplier, "MDC supplier must
not be null");
+ }
+
+ @Override
+ protected void append(final ILoggingEvent event)
+ {
+ if (event == null)
+ {
+ return;
+ }
+
+ super.postProcessEvent(event);
+ final Map<String, String> additions = _mdcSupplier.get();
+ super.append(additions == null || additions.isEmpty() ? event : new
MdcLoggingEvent(event, additions));
+ }
+
+ @Override
+ public PreSerializationTransformer<ILoggingEvent> getPST()
+ {
+ return EVENT_TRANSFORMER;
+ }
+}
diff --git
a/broker-plugins/logging-logback/src/main/java/org/apache/qpid/server/logging/logback/MdcLoggingEvent.java
b/broker-plugins/logging-logback/src/main/java/org/apache/qpid/server/logging/logback/MdcLoggingEvent.java
new file mode 100644
index 0000000000..a2e8b96375
--- /dev/null
+++
b/broker-plugins/logging-logback/src/main/java/org/apache/qpid/server/logging/logback/MdcLoggingEvent.java
@@ -0,0 +1,165 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.server.logging.logback;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.classic.spi.IThrowableProxy;
+import ch.qos.logback.classic.spi.LoggerContextVO;
+import org.slf4j.Marker;
+import org.slf4j.event.KeyValuePair;
+
+final class MdcLoggingEvent implements ILoggingEvent
+{
+ private final ILoggingEvent _event;
+ private final Map<String, String> _mdc;
+
+ MdcLoggingEvent(final ILoggingEvent event, final Map<String, String>
additions)
+ {
+ _event = Objects.requireNonNull(event, "Event must not be null");
+ final Map<String, String> mdc = new
HashMap<>(event.getMDCPropertyMap());
+ mdc.putAll(Objects.requireNonNull(additions, "MDC additions must not
be null"));
+ _mdc = Collections.unmodifiableMap(mdc);
+ }
+
+ @Override
+ public String getThreadName()
+ {
+ return _event.getThreadName();
+ }
+
+ @Override
+ public Level getLevel()
+ {
+ return _event.getLevel();
+ }
+
+ @Override
+ public String getMessage()
+ {
+ return _event.getMessage();
+ }
+
+ @Override
+ public Object[] getArgumentArray()
+ {
+ return _event.getArgumentArray();
+ }
+
+ @Override
+ public String getFormattedMessage()
+ {
+ return _event.getFormattedMessage();
+ }
+
+ @Override
+ public String getLoggerName()
+ {
+ return _event.getLoggerName();
+ }
+
+ @Override
+ public LoggerContextVO getLoggerContextVO()
+ {
+ return _event.getLoggerContextVO();
+ }
+
+ @Override
+ public IThrowableProxy getThrowableProxy()
+ {
+ return _event.getThrowableProxy();
+ }
+
+ @Override
+ public StackTraceElement[] getCallerData()
+ {
+ return _event.getCallerData();
+ }
+
+ @Override
+ public boolean hasCallerData()
+ {
+ return _event.hasCallerData();
+ }
+
+ @Override
+ public List<Marker> getMarkerList()
+ {
+ return _event.getMarkerList();
+ }
+
+ @Override
+ public Map<String, String> getMDCPropertyMap()
+ {
+ return _mdc;
+ }
+
+ @Deprecated
+ @Override
+ public Map<String, String> getMdc()
+ {
+ return _mdc;
+ }
+
+ @Override
+ public long getTimeStamp()
+ {
+ return _event.getTimeStamp();
+ }
+
+ @Override
+ public Instant getInstant()
+ {
+ return _event.getInstant();
+ }
+
+ @Override
+ public int getNanoseconds()
+ {
+ return _event.getNanoseconds();
+ }
+
+ @Override
+ public long getSequenceNumber()
+ {
+ return _event.getSequenceNumber();
+ }
+
+ @Override
+ public List<KeyValuePair> getKeyValuePairs()
+ {
+ return _event.getKeyValuePairs();
+ }
+
+ @Override
+ public void prepareForDeferredProcessing()
+ {
+ _event.prepareForDeferredProcessing();
+ }
+}
diff --git
a/broker-plugins/logging-logback/src/test/java/org/apache/qpid/server/logging/logback/MdcEnrichingSocketAppenderTest.java
b/broker-plugins/logging-logback/src/test/java/org/apache/qpid/server/logging/logback/MdcEnrichingSocketAppenderTest.java
new file mode 100644
index 0000000000..05027e165b
--- /dev/null
+++
b/broker-plugins/logging-logback/src/test/java/org/apache/qpid/server/logging/logback/MdcEnrichingSocketAppenderTest.java
@@ -0,0 +1,436 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.server.logging.logback;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyNoInteractions;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.LoggerContext;
+import ch.qos.logback.classic.net.server.HardenedLoggingEventInputStream;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.classic.spi.LoggingEvent;
+import ch.qos.logback.classic.spi.LoggingEventVO;
+import ch.qos.logback.classic.util.LogbackMDCAdapter;
+import ch.qos.logback.core.filter.Filter;
+import ch.qos.logback.core.read.ListAppender;
+import ch.qos.logback.core.spi.FilterReply;
+import ch.qos.logback.core.spi.PreSerializationTransformer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.slf4j.MarkerFactory;
+import org.slf4j.event.KeyValuePair;
+
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class MdcEnrichingSocketAppenderTest extends UnitTestBase
+{
+ private static final int TIMEOUT_SECONDS = 5;
+ private static final Map<String, String> PRODUCER_MDC = Map.of("trace",
"request", "common", "producer");
+
+ private final AtomicReference<Map<String, String>> _additions = new
AtomicReference<>();
+ private final List<MdcEnrichingSocketAppender> _appenders = new
ArrayList<>();
+
+ private LoggerContext _loggerContext;
+ private Logger _logger;
+ private CountDownLatch _serializationAllowed;
+
+ @BeforeEach
+ void setUp()
+ {
+ _additions.set(Map.of());
+ _serializationAllowed = new CountDownLatch(1);
+ _loggerContext = new LoggerContext();
+ _loggerContext.setMDCAdapter(new LogbackMDCAdapter());
+ _loggerContext.start();
+ _logger = _loggerContext.getLogger(getTestClassName());
+ _logger.setLevel(Level.INFO);
+ _logger.setAdditive(false);
+ }
+
+ @AfterEach
+ void tearDown()
+ {
+ _serializationAllowed.countDown();
+ for (final MdcEnrichingSocketAppender appender : _appenders)
+ {
+ appender.stop();
+ }
+ _appenders.clear();
+ _loggerContext.stop();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void enrichmentPreservesOriginalEventAndProducerMdc(final boolean
prepareEarlier) throws Exception
+ {
+ final ListAppender<ILoggingEvent> earlier =
addRecordingAppender(prepareEarlier);
+ _additions.set(Map.of("common", "socket", "broker", "node"));
+ try (final SocketReceiver receiver = new SocketReceiver())
+ {
+ addSocketAppender(receiver, _additions::get, false);
+ final ListAppender<ILoggingEvent> later =
addRecordingAppender(false);
+ logOnProducer(() -> _logger.info("event"));
+
+ _loggerContext.getMDCAdapter().put("trace", "reader");
+ _serializationAllowed.countDown();
+ final LoggingEventVO received = receiver.readEvent();
+
+ assertEquals(Map.of("trace", "request", "common", "socket",
"broker", "node"),
+ received.getMDCPropertyMap());
+ assertEquals("producer-thread", received.getThreadName());
+ assertEquals(1, earlier.list.size());
+ assertEquals(1, later.list.size());
+ assertSame(earlier.list.get(0), later.list.get(0));
+ assertEquals(PRODUCER_MDC, later.list.get(0).getMDCPropertyMap());
+ }
+ }
+
+ @Test
+ void eachSocketHasIndependentMdc() throws Exception
+ {
+ try (final SocketReceiver first = new SocketReceiver();
+ final SocketReceiver second = new SocketReceiver())
+ {
+ addSocketAppender(first, () -> Map.of("common", "first",
"first-only", "value"), false);
+ addSocketAppender(second, () -> Map.of("common", "second"), false);
+ logOnProducer(() -> _logger.info("event"));
+ _serializationAllowed.countDown();
+
+ assertEquals(Map.of("trace", "request", "common", "first",
"first-only", "value"),
+ first.readEvent().getMDCPropertyMap());
+ assertEquals(Map.of("trace", "request", "common", "second"),
second.readEvent().getMDCPropertyMap());
+ }
+ }
+
+ @Test
+ void queuedEventsRetainTheirMdcAfterConfigurationChanges() throws Exception
+ {
+ final Map<String, String> additions = new HashMap<>(Map.of("common",
"first"));
+ _additions.set(additions);
+ try (final SocketReceiver receiver = new SocketReceiver())
+ {
+ addSocketAppender(receiver, _additions::get, false);
+ logOnProducer(() -> _logger.info("first"));
+ additions.put("common", "mutated");
+ _additions.set(Map.of("common", "second"));
+ logOnProducer(() -> _logger.info("second"));
+ _additions.set(Map.of());
+ logOnProducer(() -> _logger.info("third"));
+ _serializationAllowed.countDown();
+
+ final LoggingEventVO first = receiver.readEvent();
+ final LoggingEventVO second = receiver.readEvent();
+ final LoggingEventVO third = receiver.readEvent();
+ assertEquals("first", first.getMessage());
+ assertEquals("first", first.getMDCPropertyMap().get("common"));
+ assertEquals("second", second.getMessage());
+ assertEquals("second", second.getMDCPropertyMap().get("common"));
+ assertEquals("third", third.getMessage());
+ assertEquals(PRODUCER_MDC, third.getMDCPropertyMap());
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"false,false", "false,true", "true,false", "true,true"})
+ void callerCaptureOnProducer(final boolean includeCallerData, final
boolean enrich) throws Exception
+ {
+ _additions.set(enrich ? Map.of("broker", "node") : Map.of());
+ final ListAppender<ILoggingEvent> recording =
addRecordingAppender(false);
+ try (final SocketReceiver receiver = new SocketReceiver())
+ {
+ addSocketAppender(receiver, _additions::get, includeCallerData);
+ logOnProducer(this::logWithCaller);
+ assertEquals(includeCallerData,
recording.list.get(0).hasCallerData());
+ _serializationAllowed.countDown();
+ final LoggingEventVO received = receiver.readEvent();
+
+ assertEquals(includeCallerData, received.hasCallerData());
+ if (includeCallerData)
+ {
+ assertTrue(received.getCallerData().length > 0);
+ assertEquals(getClass().getName(),
received.getCallerData()[0].getClassName());
+ assertEquals("logWithCaller",
received.getCallerData()[0].getMethodName());
+ }
+ else
+ {
+ assertNull(received.getCallerData());
+ }
+ }
+ }
+
+ @Test
+ void nativeSerializationPreservesEventData() throws Exception
+ {
+ final Map<String, String> additions = new HashMap<>();
+ additions.put("broker", "node");
+ additions.put("nullable", null);
+ _additions.set(additions);
+ _loggerContext.putProperty("context-property", "context-value");
+ final IllegalStateException exception = new
IllegalStateException("outer");
+ final IllegalArgumentException cause = new
IllegalArgumentException("cause", exception);
+ exception.initCause(cause);
+ exception.addSuppressed(new
UnsupportedOperationException("suppressed"));
+ final LoggingEvent original = new LoggingEvent(Logger.class.getName(),
_logger, Level.WARN,
+ "value={}", exception, new Object[]{"argument"});
+ original.setInstant(Instant.parse("2026-09-11T12:34:56.123456789Z"));
+ original.setSequenceNumber(42);
+ original.addMarker(MarkerFactory.getMarker("first"));
+ original.addMarker(MarkerFactory.getMarker("second"));
+
+ try (final SocketReceiver receiver = new SocketReceiver())
+ {
+ final MdcEnrichingSocketAppender appender =
addSocketAppender(receiver, _additions::get, false);
+ logOnProducer(() -> appender.doAppend(original));
+ _serializationAllowed.countDown();
+ final LoggingEventVO received = receiver.readEvent();
+
+ assertEquals("producer-thread", received.getThreadName());
+ assertEquals(original.getLoggerName(), received.getLoggerName());
+ assertEquals(original.getLevel(), received.getLevel());
+ assertEquals(original.getMessage(), received.getMessage());
+ assertEquals(original.getFormattedMessage(),
received.getFormattedMessage());
+ assertArrayEquals(original.getArgumentArray(),
received.getArgumentArray());
+ assertEquals(original.getTimeStamp(), received.getTimeStamp());
+ assertEquals(original.getNanoseconds(), received.getNanoseconds());
+ assertEquals(original.getSequenceNumber(),
received.getSequenceNumber());
+ assertEquals(original.getMarkerList(), received.getMarkerList());
+ assertEquals("context-value",
received.getLoggerContextVO().getPropertyMap().get("context-property"));
+ assertEquals("node", received.getMDCPropertyMap().get("broker"));
+ assertTrue(received.getMDCPropertyMap().containsKey("nullable"));
+ assertNull(received.getMDCPropertyMap().get("nullable"));
+ assertEquals(IllegalStateException.class.getName(),
received.getThrowableProxy().getClassName());
+ assertEquals("outer", received.getThrowableProxy().getMessage());
+ assertEquals("cause",
received.getThrowableProxy().getCause().getMessage());
+
assertTrue(received.getThrowableProxy().getCause().getCause().isCyclic());
+ assertEquals("suppressed",
received.getThrowableProxy().getSuppressed()[0].getMessage());
+ }
+ }
+
+ @Test
+ void transformerPreservesStructuredArgumentsAndNativeValueObjects()
+ {
+ final LoggingEvent original = new LoggingEvent(Logger.class.getName(),
_logger, Level.INFO,
+ "value={}", null, new Object[]{"argument"});
+ original.setKeyValuePairs(List.of(new KeyValuePair("key", "value")));
+ original.prepareForDeferredProcessing();
+ final MdcEnrichingSocketAppender appender = new
MdcEnrichingSocketAppender(_additions::get);
+ final PreSerializationTransformer<ILoggingEvent> transformer =
appender.getPST();
+ final MdcLoggingEvent enriched = new MdcLoggingEvent(original,
Map.of("broker", "node"));
+ final LoggingEventVO transformed =
assertInstanceOf(LoggingEventVO.class, transformer.transform(enriched));
+
+ assertEquals(original.getKeyValuePairs(),
transformed.getKeyValuePairs());
+ assertEquals(original.getInstant(), enriched.getInstant());
+ assertEquals("node", transformed.getMDCPropertyMap().get("broker"));
+ final LoggingEventVO enrichedValueObject =
assertInstanceOf(LoggingEventVO.class,
+ transformer.transform(new MdcLoggingEvent(transformed,
Map.of("broker", "another-node"))));
+ assertEquals("another-node",
enrichedValueObject.getMDCPropertyMap().get("broker"));
+ assertEquals("node", transformed.getMDCPropertyMap().get("broker"));
+ assertEquals(original.getKeyValuePairs(),
enrichedValueObject.getKeyValuePairs());
+ assertSame(transformed, transformer.transform(transformed));
+ assertNull(transformer.transform(null));
+ }
+
+ @Test
+ void rejectedEventsAreNotEnriched() throws Exception
+ {
+ final Supplier<Map<String, String>> supplier = mock(Supplier.class);
+ final ILoggingEvent event = mock(ILoggingEvent.class);
+ try (final SocketReceiver receiver = new SocketReceiver())
+ {
+ final MdcEnrichingSocketAppender appender =
addSocketAppender(receiver, supplier, true);
+ appender.addFilter(new Filter<>()
+ {
+ @Override
+ public FilterReply decide(final ILoggingEvent loggingEvent)
+ {
+ return FilterReply.DENY;
+ }
+ });
+ appender.doAppend(event);
+ appender.stop();
+ appender.clearAllFilters();
+ appender.doAppend(event);
+
+ verifyNoInteractions(supplier, event);
+ }
+ }
+
+ private void logWithCaller()
+ {
+ _logger.info("caller");
+ }
+
+ private void logOnProducer(final Runnable log) throws Exception
+ {
+ final FutureTask<Void> task = new FutureTask<>(() ->
+ {
+ _loggerContext.getMDCAdapter().setContextMap(PRODUCER_MDC);
+ try
+ {
+ log.run();
+ assertEquals(PRODUCER_MDC,
_loggerContext.getMDCAdapter().getCopyOfContextMap());
+ }
+ finally
+ {
+ _loggerContext.getMDCAdapter().clear();
+ }
+ }, null);
+ final Thread producer = new Thread(task, "producer-thread");
+ producer.start();
+ try
+ {
+ task.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ }
+ finally
+ {
+ producer.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ }
+ assertFalse(producer.isAlive());
+ }
+
+ private ListAppender<ILoggingEvent> addRecordingAppender(final boolean
prepare)
+ {
+ final ListAppender<ILoggingEvent> appender = new ListAppender<>()
+ {
+ @Override
+ protected void append(final ILoggingEvent event)
+ {
+ if (prepare)
+ {
+ event.prepareForDeferredProcessing();
+ }
+ super.append(event);
+ }
+ };
+ appender.setContext(_loggerContext);
+ appender.start();
+ _logger.addAppender(appender);
+ return appender;
+ }
+
+ private MdcEnrichingSocketAppender addSocketAppender(final SocketReceiver
receiver,
+ final
Supplier<Map<String, String>> supplier,
+ final boolean
includeCallerData)
+ {
+ final MdcEnrichingSocketAppender appender = new
MdcEnrichingSocketAppender(supplier)
+ {
+ @Override
+ public PreSerializationTransformer<ILoggingEvent> getPST()
+ {
+ final PreSerializationTransformer<ILoggingEvent> transformer =
super.getPST();
+ return event ->
+ {
+ try
+ {
+ if (!_serializationAllowed.await(TIMEOUT_SECONDS,
TimeUnit.SECONDS))
+ {
+ throw new IllegalStateException("Serialization was
not released");
+ }
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Serialization
interrupted", e);
+ }
+ return transformer.transform(event);
+ };
+ }
+ };
+ appender.setContext(_loggerContext);
+ appender.setName("socket-" + _appenders.size());
+
appender.setRemoteHost(InetAddress.getLoopbackAddress().getHostAddress());
+ appender.setPort(receiver._serverSocket.getLocalPort());
+ appender.setIncludeCallerData(includeCallerData);
+ _appenders.add(appender);
+ appender.start();
+ assertTrue(appender.isStarted());
+ _logger.addAppender(appender);
+ return appender;
+ }
+
+ private class SocketReceiver implements AutoCloseable
+ {
+ private final ServerSocket _serverSocket = new ServerSocket(0, 1,
InetAddress.getLoopbackAddress());
+ private Socket _socket;
+ private HardenedLoggingEventInputStream _input;
+
+ private SocketReceiver() throws IOException
+ {
+ _serverSocket.setSoTimeout((int)
TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ }
+
+ private LoggingEventVO readEvent() throws IOException,
ClassNotFoundException
+ {
+ if (_socket == null)
+ {
+ _socket = _serverSocket.accept();
+ _socket.setSoTimeout((int)
TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ _input = new HardenedLoggingEventInputStream(_loggerContext,
_socket.getInputStream());
+ }
+ return assertInstanceOf(LoggingEventVO.class, _input.readObject());
+ }
+
+ @Override
+ public void close() throws IOException
+ {
+ try
+ {
+ if (_socket != null)
+ {
+ _socket.close();
+ }
+ }
+ finally
+ {
+ _serverSocket.close();
+ }
+ }
+ }
+}
diff --git
a/broker-plugins/logging-logback/src/test/java/org/apache/qpid/server/logging/logback/MdcLoggingEventTest.java
b/broker-plugins/logging-logback/src/test/java/org/apache/qpid/server/logging/logback/MdcLoggingEventTest.java
new file mode 100644
index 0000000000..78cff0c66e
--- /dev/null
+++
b/broker-plugins/logging-logback/src/test/java/org/apache/qpid/server/logging/logback/MdcLoggingEventTest.java
@@ -0,0 +1,71 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.server.logging.logback;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import ch.qos.logback.classic.spi.LoggingEvent;
+import org.junit.jupiter.api.Test;
+
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class MdcLoggingEventTest extends UnitTestBase
+{
+ @Test
+ @SuppressWarnings("deprecation")
+ void mdcIsAnUnmodifiableSnapshot()
+ {
+ final Map<String, String> originalMdc = new HashMap<>(Map.of("trace",
"request", "common", "original"));
+ final LoggingEvent original = new LoggingEvent();
+ original.setMDCPropertyMap(originalMdc);
+ final Map<String, String> additions = new HashMap<>(Map.of("common",
"socket", "broker", "node"));
+ final MdcLoggingEvent event = new MdcLoggingEvent(original, additions);
+ final Map<String, String> expected = Map.of("trace", "request",
"common", "socket", "broker", "node");
+
+ assertEquals(Map.of("trace", "request", "common", "original"),
originalMdc);
+ originalMdc.clear();
+ additions.clear();
+
+ assertEquals(expected, event.getMDCPropertyMap());
+ assertSame(event.getMDCPropertyMap(), event.getMdc());
+ assertThrows(UnsupportedOperationException.class, () ->
event.getMDCPropertyMap().put("common", "changed"));
+ assertThrows(UnsupportedOperationException.class, () ->
+
event.getMDCPropertyMap().entrySet().iterator().next().setValue("changed"));
+ }
+
+ @Test
+ void checkingForCallerDataDoesNotCaptureIt()
+ {
+ final LoggingEvent original = new LoggingEvent();
+ original.setMDCPropertyMap(Map.of());
+ final MdcLoggingEvent event = new MdcLoggingEvent(original,
Map.of("broker", "node"));
+
+ assertFalse(event.hasCallerData());
+ assertFalse(original.hasCallerData());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
