From 9bc5af5be2d010358b205534009369703ba82231 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E9=82=A3=E5=87=A0?= <2078634206@qq.com> Date: Wed, 30 Sep 2026 19:06:13 +0800 Subject: [PATCH 1/2] [ISSUE #11284] fix(store): do not report plugin init failures as class-not-found MessageStoreFactory.build wrapped Class.forName, getConstructor and newInstance in one catch and always reported "Initialize plugin's class: X not found!", even when the class existed but its constructor threw, so the top-level message pointed operators at classpath problems instead of the real failure. Use a neutral "Failed to initialize message store plugin" message, keep the original cause, and add unit tests covering the missing-class, throwing-constructor, chaining and no-plugin paths. --- .../store/plugin/MessageStoreFactory.java | 2 +- .../store/plugin/MessageStoreFactoryTest.java | 185 ++++++++++++++++++ 2 files changed, 186 insertions(+), 1 deletion(-) create mode 100755 store/src/test/java/org/apache/rocketmq/store/plugin/MessageStoreFactoryTest.java diff --git a/store/src/main/java/org/apache/rocketmq/store/plugin/MessageStoreFactory.java b/store/src/main/java/org/apache/rocketmq/store/plugin/MessageStoreFactory.java index 8d929ea5651..e1bc9f8e4ed 100644 --- a/store/src/main/java/org/apache/rocketmq/store/plugin/MessageStoreFactory.java +++ b/store/src/main/java/org/apache/rocketmq/store/plugin/MessageStoreFactory.java @@ -36,7 +36,7 @@ public static MessageStore build(MessageStorePluginContext context, AbstractPluginMessageStore pluginMessageStore = construct.newInstance(context, messageStore); messageStore = pluginMessageStore; } catch (Throwable e) { - throw new RuntimeException("Initialize plugin's class: " + pluginClass + " not found!", e); + throw new RuntimeException("Failed to initialize message store plugin: " + pluginClass, e); } } } diff --git a/store/src/test/java/org/apache/rocketmq/store/plugin/MessageStoreFactoryTest.java b/store/src/test/java/org/apache/rocketmq/store/plugin/MessageStoreFactoryTest.java new file mode 100755 index 00000000000..2395cfd6651 --- /dev/null +++ b/store/src/test/java/org/apache/rocketmq/store/plugin/MessageStoreFactoryTest.java @@ -0,0 +1,185 @@ +/* + * 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.rocketmq.store.plugin; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.fail; + +import java.lang.reflect.InvocationTargetException; +import java.util.concurrent.CompletableFuture; +import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.store.MessageStore; +import org.apache.rocketmq.store.QueryMessageResult; +import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage; +import org.apache.rocketmq.store.timer.rocksdb.TimerMessageRocksDBStore; +import org.apache.rocketmq.store.transaction.TransMessageRocksDBStore; +import org.junit.Test; + +public class MessageStoreFactoryTest { + + private static final String MISSING_PLUGIN_CLASS = + "org.apache.rocketmq.store.plugin.NoSuchPluginClassSurelyMissing"; + private static final String THROWING_PLUGIN = + MessageStoreFactoryTest.class.getName() + "$ThrowingPlugin"; + private static final String OUTER_PLUGIN = + MessageStoreFactoryTest.class.getName() + "$OuterPlugin"; + private static final String INNER_PLUGIN = + MessageStoreFactoryTest.class.getName() + "$InnerPlugin"; + + private MessageStorePluginContext buildContext(String plugin) { + BrokerConfig brokerConfig = new BrokerConfig(); + brokerConfig.setMessageStorePlugIn(plugin); + return new MessageStorePluginContext(new MessageStoreConfig(), null, null, brokerConfig, null); + } + + @Test + public void testPluginClassNotExist() throws Exception { + MessageStorePluginContext context = buildContext(MISSING_PLUGIN_CLASS); + try { + MessageStoreFactory.build(context, null); + fail("build should fail when the plugin class does not exist"); + } catch (RuntimeException e) { + assertThat(e.getMessage()) + .contains("Failed to initialize message store plugin") + .contains("NoSuchPluginClassSurelyMissing"); + assertThat(e.getCause()).isInstanceOf(ClassNotFoundException.class); + } + } + + @Test + public void testPluginConstructorThrows() throws Exception { + MessageStorePluginContext context = buildContext(THROWING_PLUGIN); + try { + MessageStoreFactory.build(context, null); + fail("build should fail when the plugin constructor throws"); + } catch (RuntimeException e) { + assertThat(e.getMessage()) + .contains("Failed to initialize message store plugin") + .contains(THROWING_PLUGIN) + .doesNotContain("not found"); + assertThat(e.getCause()).isInstanceOf(InvocationTargetException.class); + Throwable rootCause = e.getCause().getCause(); + assertThat(rootCause).isInstanceOf(IllegalStateException.class); + assertThat(rootCause).hasMessage("boom"); + } + } + + @Test + public void testBuildSinglePlugin() throws Exception { + MessageStorePluginContext context = buildContext(INNER_PLUGIN); + MessageStore store = MessageStoreFactory.build(context, null); + assertThat(store).isInstanceOf(InnerPlugin.class); + assertThat(((InnerPlugin) store).getNext()).isNull(); + assertThat(((InnerPlugin) store).context).isSameAs(context); + } + + @Test + public void testBuildChainsPluginsInConfigOrder() throws Exception { + // no space after the comma: build splits the config on "," without trimming entries + MessageStorePluginContext context = buildContext(OUTER_PLUGIN + "," + INNER_PLUGIN); + MessageStore store = MessageStoreFactory.build(context, null); + assertThat(store).isInstanceOf(OuterPlugin.class); + AbstractPluginMessageStore outer = (AbstractPluginMessageStore) store; + assertThat(outer.getNext()).isInstanceOf(InnerPlugin.class); + AbstractPluginMessageStore inner = (AbstractPluginMessageStore) outer.getNext(); + assertThat(inner.getNext()).isNull(); + assertThat(outer.context).isSameAs(context); + assertThat(inner.context).isSameAs(context); + } + + @Test + public void testNullPluginConfigReturnsOriginalStore() throws Exception { + MessageStorePluginContext context = buildContext(null); + MessageStore delegate = new InnerPlugin(context, null); + assertThat(MessageStoreFactory.build(context, delegate)).isSameAs(delegate); + } + + @Test + public void testBlankPluginConfigReturnsOriginalStore() throws Exception { + MessageStorePluginContext context = buildContext(" "); + MessageStore delegate = new InnerPlugin(context, null); + assertThat(MessageStoreFactory.build(context, delegate)).isSameAs(delegate); + } + + /** + * Test-local plugin base. AbstractPluginMessageStore leaves a few MessageStore methods + * unimplemented in this branch, so concrete plugins used by the tests fill the gap here. + */ + private static class TestPlugin extends AbstractPluginMessageStore { + TestPlugin(MessageStorePluginContext context, MessageStore next) { + super(context, next); + } + + @Override + public QueryMessageResult queryMessage(String topic, String key, int maxNum, long begin, long end, + String indexType, String lastKey) { + return next.queryMessage(topic, key, maxNum, begin, end, indexType, lastKey); + } + + @Override + public CompletableFuture queryMessageAsync(String topic, String key, int maxNum, + long begin, long end, String indexType, String lastKey) { + return next.queryMessageAsync(topic, key, maxNum, begin, end, indexType, lastKey); + } + + @Override + public MessageRocksDBStorage getMessageRocksDBStorage() { + return next == null ? null : next.getMessageRocksDBStorage(); + } + + @Override + public TimerMessageRocksDBStore getTimerMessageRocksDBStore() { + return next == null ? null : next.getTimerMessageRocksDBStore(); + } + + @Override + public TransMessageRocksDBStore getTransMessageRocksDBStore() { + return next == null ? null : next.getTransMessageRocksDBStore(); + } + + @Override + public void setTimerMessageRocksDBStore(TimerMessageRocksDBStore timerMessageRocksDBStore) { + next.setTimerMessageRocksDBStore(timerMessageRocksDBStore); + } + + @Override + public void setTransMessageRocksDBStore(TransMessageRocksDBStore transMessageRocksDBStore) { + next.setTransMessageRocksDBStore(transMessageRocksDBStore); + } + } + + public static class OuterPlugin extends TestPlugin { + public OuterPlugin(MessageStorePluginContext context, MessageStore next) { + super(context, next); + } + } + + public static class InnerPlugin extends TestPlugin { + public InnerPlugin(MessageStorePluginContext context, MessageStore next) { + super(context, next); + } + } + + public static class ThrowingPlugin extends TestPlugin { + public ThrowingPlugin(MessageStorePluginContext context, MessageStore next) { + super(context, next); + throw new IllegalStateException("boom"); + } + } +} From 327a2b614b97c223f80138bd9ee6bd6becab4ee5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E9=82=A3=E5=87=A0?= <2078634206@qq.com> Date: Thu, 1 Oct 2026 03:53:23 +0800 Subject: [PATCH 2/2] [ISSUE #11284] Address review feedback on plugin init diagnostics - commit MessageStoreFactoryTest with file mode 100644 like the rest of store tests - replace try/fail/catch with assertThatThrownBy chains in the two failure tests - rephrase the TestPlugin javadoc ("this branch" read like the PR branch) - log plugin init failure at ERROR (store logger) before rethrowing, so the cause chain reaches the broker log even if the RuntimeException is re-wrapped --- .../store/plugin/MessageStoreFactory.java | 6 +++ .../store/plugin/MessageStoreFactoryTest.java | 41 ++++++++----------- 2 files changed, 22 insertions(+), 25 deletions(-) mode change 100755 => 100644 store/src/test/java/org/apache/rocketmq/store/plugin/MessageStoreFactoryTest.java diff --git a/store/src/main/java/org/apache/rocketmq/store/plugin/MessageStoreFactory.java b/store/src/main/java/org/apache/rocketmq/store/plugin/MessageStoreFactory.java index e1bc9f8e4ed..46188023c5f 100644 --- a/store/src/main/java/org/apache/rocketmq/store/plugin/MessageStoreFactory.java +++ b/store/src/main/java/org/apache/rocketmq/store/plugin/MessageStoreFactory.java @@ -19,9 +19,14 @@ import java.io.IOException; import java.lang.reflect.Constructor; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.logging.org.slf4j.Logger; +import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; import org.apache.rocketmq.store.MessageStore; public final class MessageStoreFactory { + private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); + public static MessageStore build(MessageStorePluginContext context, MessageStore messageStore) throws IOException { String plugin = context.getBrokerConfig().getMessageStorePlugIn(); @@ -36,6 +41,7 @@ public static MessageStore build(MessageStorePluginContext context, AbstractPluginMessageStore pluginMessageStore = construct.newInstance(context, messageStore); messageStore = pluginMessageStore; } catch (Throwable e) { + LOGGER.error("Failed to initialize message store plugin: {}", pluginClass, e); throw new RuntimeException("Failed to initialize message store plugin: " + pluginClass, e); } } diff --git a/store/src/test/java/org/apache/rocketmq/store/plugin/MessageStoreFactoryTest.java b/store/src/test/java/org/apache/rocketmq/store/plugin/MessageStoreFactoryTest.java old mode 100755 new mode 100644 index 2395cfd6651..aab73f11b41 --- a/store/src/test/java/org/apache/rocketmq/store/plugin/MessageStoreFactoryTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/plugin/MessageStoreFactoryTest.java @@ -18,7 +18,7 @@ package org.apache.rocketmq.store.plugin; import static org.assertj.core.api.Assertions.assertThat; -import static org.junit.Assert.fail; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.lang.reflect.InvocationTargetException; import java.util.concurrent.CompletableFuture; @@ -51,33 +51,24 @@ private MessageStorePluginContext buildContext(String plugin) { @Test public void testPluginClassNotExist() throws Exception { MessageStorePluginContext context = buildContext(MISSING_PLUGIN_CLASS); - try { - MessageStoreFactory.build(context, null); - fail("build should fail when the plugin class does not exist"); - } catch (RuntimeException e) { - assertThat(e.getMessage()) - .contains("Failed to initialize message store plugin") - .contains("NoSuchPluginClassSurelyMissing"); - assertThat(e.getCause()).isInstanceOf(ClassNotFoundException.class); - } + assertThatThrownBy(() -> MessageStoreFactory.build(context, null)) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Failed to initialize message store plugin") + .hasMessageContaining("NoSuchPluginClassSurelyMissing") + .hasCauseInstanceOf(ClassNotFoundException.class); } @Test public void testPluginConstructorThrows() throws Exception { MessageStorePluginContext context = buildContext(THROWING_PLUGIN); - try { - MessageStoreFactory.build(context, null); - fail("build should fail when the plugin constructor throws"); - } catch (RuntimeException e) { - assertThat(e.getMessage()) - .contains("Failed to initialize message store plugin") - .contains(THROWING_PLUGIN) - .doesNotContain("not found"); - assertThat(e.getCause()).isInstanceOf(InvocationTargetException.class); - Throwable rootCause = e.getCause().getCause(); - assertThat(rootCause).isInstanceOf(IllegalStateException.class); - assertThat(rootCause).hasMessage("boom"); - } + assertThatThrownBy(() -> MessageStoreFactory.build(context, null)) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Failed to initialize message store plugin") + .hasMessageContaining(THROWING_PLUGIN) + .hasMessageNotContaining("not found") + .hasCauseInstanceOf(InvocationTargetException.class) + .hasRootCauseInstanceOf(IllegalStateException.class) + .hasRootCauseMessage("boom"); } @Test @@ -118,8 +109,8 @@ public void testBlankPluginConfigReturnsOriginalStore() throws Exception { } /** - * Test-local plugin base. AbstractPluginMessageStore leaves a few MessageStore methods - * unimplemented in this branch, so concrete plugins used by the tests fill the gap here. + * Test-local plugin base. AbstractPluginMessageStore does not delegate these methods, + * so concrete plugins used by the tests fill the gap here. */ private static class TestPlugin extends AbstractPluginMessageStore { TestPlugin(MessageStorePluginContext context, MessageStore next) {