This is an automated email from the ASF dual-hosted git repository. penghui pushed a commit to branch branch-2.8 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 076fd93c9c94a8db75ce45e51bf3066016947ccb Author: Jason918 <[email protected]> AuthorDate: Mon Jul 12 22:26:34 2021 +0800 Change ContextClassLoader to NarClassLoader in ProtocolHandler (#11276) Co-authored-by: Jiang Haiting <[email protected]> (cherry picked from commit 68870054351f32e3992f4c6ea6034cd79ace5193) --- .../protocol/ProtocolHandlerWithClassLoader.java | 47 +++++++++++-- .../ProtocolHandlerWithClassLoaderTest.java | 81 ++++++++++++++++++++++ 2 files changed, 121 insertions(+), 7 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/protocol/ProtocolHandlerWithClassLoader.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/protocol/ProtocolHandlerWithClassLoader.java index 5adb852..223cf81 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/protocol/ProtocolHandlerWithClassLoader.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/protocol/ProtocolHandlerWithClassLoader.java @@ -43,41 +43,74 @@ class ProtocolHandlerWithClassLoader implements ProtocolHandler { @Override public String protocolName() { - return handler.protocolName(); + try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) { + return handler.protocolName(); + } } @Override public boolean accept(String protocol) { - return handler.accept(protocol); + try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) { + return handler.accept(protocol); + } } @Override public void initialize(ServiceConfiguration conf) throws Exception { - handler.initialize(conf); + try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) { + handler.initialize(conf); + } } @Override public String getProtocolDataToAdvertise() { - return handler.getProtocolDataToAdvertise(); + try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) { + return handler.getProtocolDataToAdvertise(); + } } @Override public void start(BrokerService service) { - handler.start(service); + try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) { + handler.start(service); + } } @Override public Map<InetSocketAddress, ChannelInitializer<SocketChannel>> newChannelInitializers() { - return handler.newChannelInitializers(); + try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) { + return handler.newChannelInitializers(); + } } @Override public void close() { - handler.close(); + try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) { + handler.close(); + } + try { classLoader.close(); } catch (IOException e) { log.warn("Failed to close the protocol handler class loader", e); } } + + /** + * Help to switch the class loader of current thread to the NarClassLoader, and change it back when it's done. + * With the help of try-with-resources statement, the code would be cleaner than using try finally every time. + */ + private static class ClassLoaderSwitcher implements AutoCloseable { + private final ClassLoader prevClassLoader; + + ClassLoaderSwitcher(ClassLoader classLoader) { + prevClassLoader = Thread.currentThread().getContextClassLoader(); + Thread.currentThread().setContextClassLoader(classLoader); + } + + @Override + public void close() { + Thread.currentThread().setContextClassLoader(prevClassLoader); + } + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/protocol/ProtocolHandlerWithClassLoaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/protocol/ProtocolHandlerWithClassLoaderTest.java index 0704d63..42d26d5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/protocol/ProtocolHandlerWithClassLoaderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/protocol/ProtocolHandlerWithClassLoaderTest.java @@ -25,8 +25,14 @@ import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNull; import static org.testng.Assert.assertTrue; +import static org.testng.Assert.expectThrows; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.socket.SocketChannel; +import java.net.InetSocketAddress; +import java.util.Map; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.common.nar.NarClassLoader; @@ -68,4 +74,79 @@ public class ProtocolHandlerWithClassLoaderTest { verify(h, times(1)).getProtocolDataToAdvertise(); } + public void testClassLoaderSwitcher() throws Exception { + NarClassLoader loader = mock(NarClassLoader.class); + + String protocol = "test-protocol"; + + ProtocolHandler h = new ProtocolHandler() { + @Override + public String protocolName() { + assertEquals(Thread.currentThread().getContextClassLoader(), loader); + return protocol; + } + + @Override + public boolean accept(String protocol) { + assertEquals(Thread.currentThread().getContextClassLoader(), loader); + return true; + } + + @Override + public void initialize(ServiceConfiguration conf) throws Exception { + assertEquals(Thread.currentThread().getContextClassLoader(), loader); + throw new Exception("test exception"); + } + + @Override + public String getProtocolDataToAdvertise() { + assertEquals(Thread.currentThread().getContextClassLoader(), loader); + return "test-protocol-data"; + } + + @Override + public void start(BrokerService service) { + assertEquals(Thread.currentThread().getContextClassLoader(), loader); + } + + @Override + public Map<InetSocketAddress, ChannelInitializer<SocketChannel>> newChannelInitializers() { + assertEquals(Thread.currentThread().getContextClassLoader(), loader); + return null; + } + + @Override + public void close() { + assertEquals(Thread.currentThread().getContextClassLoader(), loader); + } + }; + ProtocolHandlerWithClassLoader wrapper = new ProtocolHandlerWithClassLoader(h, loader); + + ClassLoader curClassLoader = Thread.currentThread().getContextClassLoader(); + + assertEquals(wrapper.protocolName(), protocol); + assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader); + + assertTrue(wrapper.accept(protocol)); + assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader); + + + ServiceConfiguration conf = new ServiceConfiguration(); + expectThrows(Exception.class, () -> wrapper.initialize(conf)); + assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader); + + assertEquals(wrapper.getProtocolDataToAdvertise(), "test-protocol-data"); + assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader); + + BrokerService service = mock(BrokerService.class); + wrapper.start(service); + assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader); + + + assertNull(wrapper.newChannelInitializers()); + assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader); + + wrapper.close(); + assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader); + } }
