W-16789548: New chains for validation: all are not disposed generatin…

…g a memory leak (#14011) (#14036)

* New chains for validation: all are not disposed generating a memory leak

* add TC

core-tests/src/test/java/org/mule/runtime/core/privileged/processor/chain/AbstractMessageProcessorChainTestCase.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2023 Salesforce, Inc. All rights reserved.
+ * The software in this package is published under the terms of the CPAL v1.0
+ * license, a copy of which has been included with this distribution in the
+ * LICENSE.txt file.
+ */
+package org.mule.runtime.core.privileged.processor.chain;
+
+import static org.mule.runtime.core.api.lifecycle.LifecycleUtils.startIfNeeded;
+import static org.mule.runtime.core.api.lifecycle.LifecycleUtils.stopIfNeeded;
+import static org.mule.runtime.core.privileged.processor.MessageProcessors.newChain;
+
+import static java.util.Optional.empty;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.sameInstance;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.MockitoAnnotations.openMocks;
+
+import org.mule.runtime.core.api.context.notification.MuleContextListener;
+import org.mule.runtime.core.internal.context.DefaultMuleContext;
+import org.mule.tck.junit4.AbstractMuleTestCase;
+import org.mule.tck.testmodels.mule.TestMessageProcessor;
+
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+
+
+public class AbstractMessageProcessorChainTestCase extends AbstractMuleTestCase {
+
+  @InjectMocks
+  MessageProcessorChain messageChain = newChain(empty(), new TestMessageProcessor("test"));
+
+  @Mock
+  DefaultMuleContext muleContext;
+
+  public AbstractMessageProcessorChainTestCase() {
+    openMocks(this);
+  }
+
+  @Test
+  public void testRemoveChain() throws Exception {
+    startIfNeeded(messageChain);
+    stopIfNeeded(messageChain);
+
+    ArgumentCaptor<MuleContextListener> addListenerCaptor = ArgumentCaptor.forClass(MuleContextListener.class);
+    ArgumentCaptor<MuleContextListener> removeListenerCaptor = ArgumentCaptor.forClass(MuleContextListener.class);
+
+    verify(muleContext, times(1)).addListener(addListenerCaptor.capture());
+    verify(muleContext, times(1)).removeListener(removeListenerCaptor.capture());
+
+    assertThat(addListenerCaptor.getValue(), is(sameInstance(removeListenerCaptor.getValue())));
+  }
+}

core/src/main/java/org/mule/runtime/core/privileged/processor/chain/AbstractMessageProcessorChain.java

@@ -197,6 +197,8 @@ abstract class AbstractMessageProcessorChain extends AbstractExecutableComponent
 
   private ComponentTracer<CoreEvent> chainComponentTracer;
 
+  private MuleContextListener stopListener;
+
   // This is used to verify if a span has to be ended in case of error handling.
   // In case an exception is raised before the chain begins to execute, there is no current span set for the chain.
   // This can happen, for example, if an exception is raised because of too many child context created
@@ -592,7 +594,7 @@ private static String getProcessorPath(Processor processor) {
 
   private void registerStopListener() {
     if (muleContext instanceof DefaultMuleContext) {
-      MuleContextListener listener = new MuleContextListener() {
+      stopListener = new MuleContextListener() {
 
         @Override
         public void onCreation(MuleContext context) {
@@ -615,7 +617,7 @@ public void onStop(MuleContext context, Registry registry) {
           ((DefaultMuleContext) muleContext).removeListener(this);
         }
       };
-      ((DefaultMuleContext) muleContext).addListener(listener);
+      ((DefaultMuleContext) muleContext).addListener(stopListener);
     }
   }
 
@@ -779,6 +781,10 @@ public void start() throws MuleException {
   public void stop() throws MuleException {
     canProcessMessage = false;
     stopIfNeeded(getMessageProcessorsForLifecycle());
+
+    if (stopListener != null) {
+      ((DefaultMuleContext) muleContext).removeListener(stopListener);
+    }
   }
 
   @Override