Cxf 9251_Deregister_LoggingCallback

core/src/main/java/org/apache/cxf/io/CachedOutputStream.java

@@ -164,6 +164,13 @@ public List<CachedOutputStreamCallback> getCallbacks() {
         return callbacks == null ? null : Collections.unmodifiableList(callbacks);
     }
 
+    private List<CachedOutputStreamCallback> getCallbacksSnapshot() {
+        if (callbacks == null || callbacks.isEmpty()) {
+            return Collections.emptyList();
+        }
+        return new ArrayList<>(callbacks);
+    }
+
     /**
      * Perform any actions required on stream flush (freeze headers, reset
      * output stream ... etc.)
@@ -208,7 +215,8 @@ public void lockOutputStream() throws IOException {
         currentStream.flush();
         outputLocked = true;
         if (null != callbacks) {
-            for (CachedOutputStreamCallback cb : callbacks) {
+            // Using a snapshot so the cb may unregister itself from the cos
+            for (CachedOutputStreamCallback cb : getCallbacksSnapshot()) {
                 cb.onClose(this);
             }
         }
@@ -221,7 +229,8 @@ public void close() throws IOException {
         currentStream.flush();
         outputLocked = true;
         if (null != callbacks) {
-            for (CachedOutputStreamCallback cb : callbacks) {
+            // Using a snapshot so the cb may unregister itself from the cos
+            for (CachedOutputStreamCallback cb : getCallbacksSnapshot()) {
                 try {
                     cb.onClose(this);
                 } catch (final RuntimeException ex) {

rt/features/logging/src/main/java/org/apache/cxf/ext/logging/LoggingOutInterceptor.java

@@ -193,11 +193,28 @@ public LoggingCallback(final LogEventSender sender, final Message msg, final Out
             this.lim = limit == -1 ? Integer.MAX_VALUE : limit;
         }
 
+        private void deregisterFrom(CachedOutputStream cos){
+            // -------------
+            // CXF-9251
+            // If cos has a tmp file (so 'threshold' was triggered), after the introduction of DelayedCachedOutputStreamCleaner, a reference of LoggingOutputStream
+            // is held in a queue list ( DelayQueue<DelayedCloseable> queue )
+            // If something goes wrong while closing the LoggingOutputStream, this can be recall and log twice (or more) when trying to delete the orphan tmp file.
+            // Furthermore, the LoggingOutputStream that registered this callback holds a reference to this Object, so it remains in memory avoiding GC until the DelayedCachedOutputStreamCleaner does his job (default 30 minutes)
+            // -------------
+            // Doing, instead, this we should be ok :)
+            cos.deregisterCallback(this);
+        }
+
         public void onFlush(CachedOutputStream cos) {
 
         }
 
         public void onClose(CachedOutputStream cos) {
+
+            // The callback has been called, so we release it.
+            // Avoid logging twice in certain scenarios.
+            deregisterFrom(cos);
+
             final LogEvent event = eventMapper.map(message, sensitiveProtocolHeaderNames);
             if (shouldLogContent(event)) {
                 copyPayload(cos, event);

rt/features/logging/src/test/java/org/apache/cxf/ext/logging/AttributeMaskSensitiveHelperTest.java

@@ -181,6 +181,9 @@ public void shouldReplaceSensitiveDataOutWithAdd() throws IOException {
         LogEvent event = logEventSender.getLogEvent();
         assertNotNull(event);
         assertEquals(maskedContent, event.getPayload());
+
+        // Assert that the LoggingCallback has been deregistered from callbacks
+        assertEquals(0, ((LoggingOutputStream) out).getCallbacks().size());
     }
 
     @Test
@@ -203,6 +206,9 @@ public void shouldReplaceSensitiveDataOutWithSet() throws IOException {
         LogEvent event = logEventSender.getLogEvent();
         assertNotNull(event);
         assertEquals(maskedContent, event.getPayload());
+
+        // Assert that the LoggingCallback has been deregistered from callbacks
+        assertEquals(0, ((LoggingOutputStream) out).getCallbacks().size());
     }
 
     @Test
@@ -222,6 +228,9 @@ public void shouldNotReplaceSensitiveDataEmptyExpression() throws IOException {
         LogEvent event = logEventSender.getLogEvent();
         assertNotNull(event);
         assertEquals(loggingContent, event.getPayload());
+
+        // Assert that the LoggingCallback has been deregistered from callbacks
+        assertEquals(0, ((LoggingOutputStream) out).getCallbacks().size());
     }
 
     private Message prepareInMessage() {

rt/features/logging/src/test/java/org/apache/cxf/ext/logging/FaultTest.java

@@ -65,5 +65,8 @@ public void logOnceForFaultsOccurringAfterLoggingOutPhase() throws IOException {
 
         assertEquals(1, logEventSender.getLogEvents().size());
         assertEquals("TestMessage", logEventSender.getLogEvents().get(0).getPayload());
+
+        // Assert that the LoggingCallback has been deregistered from callbacks
+        assertEquals(0, ((LoggingOutputStream) postFaultOut).getCallbacks().size());
     }
 }

rt/features/logging/src/test/java/org/apache/cxf/ext/logging/MaskSensitiveHelperTest.java

@@ -203,6 +203,9 @@ public void shouldReplaceSensitiveDataOutWithAdd() throws IOException {
         LogEvent event = logEventSender.getLogEvent();
         assertNotNull(event);
         assertEquals(maskedContent, event.getPayload());
+
+        // Assert that the LoggingCallback has been deregistered from callbacks
+        assertEquals(0, ((LoggingOutputStream) out).getCallbacks().size());
     }
 
     @Test
@@ -224,6 +227,9 @@ public void shouldReplaceSensitiveDataOutWithSet() throws IOException {
         LogEvent event = logEventSender.getLogEvent();
         assertNotNull(event);
         assertEquals(maskedContent, event.getPayload());
+
+        // Assert that the LoggingCallback has been deregistered from callbacks
+        assertEquals(0, ((LoggingOutputStream) out).getCallbacks().size());
     }
 
     @Test
@@ -243,6 +249,9 @@ public void shouldNotReplaceSensitiveDataEmptyExpression() throws IOException {
         LogEvent event = logEventSender.getLogEvent();
         assertNotNull(event);
         assertEquals(loggingContent, event.getPayload());
+
+        // Assert that the LoggingCallback has been deregistered from callbacks
+        assertEquals(0, ((LoggingOutputStream) out).getCallbacks().size());
     }
 
     private Message prepareInMessage() {

rt/features/logging/src/test/java/org/apache/cxf/ext/logging/TransformTest.java

@@ -92,6 +92,9 @@ public void transformOutboundInterceptorOutputStream() throws IOException {
         LogEvent event = logEventSender.getLogEvent();
         assertNotNull(event);
         assertEquals(TRANSFORMED_LOGGING_CONTENT, event.getPayload());
+
+        // Assert that the LoggingCallback has been deregistered from callbacks
+        assertEquals(0, ((LoggingOutputStream) out).getCallbacks().size());
     }
 
     @Test

rt/features/logging/src/test/java/org/apache/cxf/ext/logging/TruncatedTest.java

@@ -68,6 +68,9 @@ public void truncatedOutboundInterceptorOutputStream() throws IOException {
         assertNotNull(event);
         assertEquals("T", event.getPayload()); // only the first byte is read!
         assertTrue(event.isTruncated());
+
+        // Assert that the LoggingCallback has been deregistered from callbacks
+        assertEquals(0, ((LoggingOutputStream) out).getCallbacks().size());
     }
 
     @Test