Merge upstream/release_v4.8.2 to pick up tronprotocol#6757 (--es fix)

Author: warku123

common/src/main/java/org/tron/core/config/args/EventConfig.java

@@ -43,7 +43,7 @@ public class EventConfig {
   @Getter
   @Setter
   public static class NativeConfig {
-    private boolean useNativeQueue = true;
+    private boolean useNativeQueue = false;
     private int bindport = 5555;
     private int sendqueuelength = 1000;
   }

common/src/main/resources/reference.conf

@@ -758,7 +758,7 @@ event.subscribe = {
   enable = false
 
   native = {
-    useNativeQueue = true
+    useNativeQueue = false
     bindport = 5555
     sendqueuelength = 1000
   }

framework/src/main/java/org/tron/common/logsfilter/EventPluginLoader.java

@@ -3,6 +3,7 @@
 import com.beust.jcommander.internal.Sets;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Strings;
 import java.io.File;
 import java.util.HashSet;
 import java.util.List;
@@ -14,8 +15,10 @@
 import org.bouncycastle.util.encoders.Hex;
 import org.pf4j.CompoundPluginDescriptorFinder;
 import org.pf4j.DefaultPluginManager;
+import org.pf4j.DefaultVersionManager;
 import org.pf4j.ManifestPluginDescriptorFinder;
 import org.pf4j.PluginManager;
+import org.pf4j.VersionManager;
 import org.springframework.util.StringUtils;
 import org.tron.common.logsfilter.nativequeue.NativeMessageQueue;
 import org.tron.common.logsfilter.trigger.BlockLogTrigger;
@@ -29,6 +32,16 @@
 @Slf4j
 public class EventPluginLoader {
 
+  /**
+   * Minimum event-plugin Plugin-Version compatible with this node. Bumped to 3.0.0 to
+   * reject pre-fastjson-removal builds whose worker threads would fail with
+   * NoClassDefFoundError on com.alibaba.fastjson at runtime. The previous event-plugin
+   * release is 2.2.0, so 3.0.0 is the first version that ships the Jackson replacement.
+   */
+  static final String MIN_PLUGIN_VERSION = "3.0.0";
+
+  private static final VersionManager VERSION_MANAGER = new DefaultVersionManager();
+
   private static EventPluginLoader instance;
 
   private long MAX_PENDING_SIZE = 50000;
@@ -457,6 +470,10 @@ protected CompoundPluginDescriptorFinder createPluginDescriptorFinder() {
       return false;
     }
 
+    if (!isPluginVersionSupported(pluginManager, pluginId)) {
+      return false;
+    }
+
     pluginManager.startPlugins();
 
     eventListeners = pluginManager.getExtensions(IPluginEventListener.class);
@@ -471,6 +488,21 @@ protected CompoundPluginDescriptorFinder createPluginDescriptorFinder() {
     return true;
   }
 
+  static boolean isPluginVersionSupported(PluginManager pm, String pluginId) {
+    String pluginVersion = pm.getPlugin(pluginId).getDescriptor().getVersion();
+    if (Strings.isNullOrEmpty(pluginVersion)) {
+      return false;
+    }
+    boolean isSupported = VERSION_MANAGER.compareVersions(pluginVersion, MIN_PLUGIN_VERSION) >= 0;
+
+    if (!isSupported) {
+      logger.error(
+          "event-plugin '{}' version {} is older than required {}, please upgrade event-plugin",
+          pluginId, pluginVersion, MIN_PLUGIN_VERSION);
+    }
+    return isSupported;
+  }
+
   public void stopPlugin() {
     if (Objects.nonNull(pluginManager)) {
       pluginManager.stopPlugins();

framework/src/main/java/org/tron/core/config/args/Args.java

@@ -167,16 +167,19 @@ public static void setParam(final String[] args, final String confFileName) {
         ? cmd.shellConfFileName : confFileName;
     Config config = Configuration.getByFileName(configFilePath);
 
-    // 2. Config overrides defaults
+    // 2. Config overrides defaults (event config bean is read here but not yet applied)
     applyConfigParams(config);
 
-    // 3. CLI overrides Config (highest priority)
+    // 3. CLI overrides Config (highest priority, including --es → eventSubscribe)
     applyCLIParams(cmd, jc);
 
-    // 4. Apply platform constraints (e.g. ARM64 forces RocksDB)
+    // 4. Apply event config after CLI
+    applyEventConfig(eventConfig);
+
+    // 5. Apply platform constraints (e.g. ARM64 forces RocksDB)
     applyPlatformConstraints();
 
-    // 5. Init witness (depends on CLI witness flag)
+    // 6. Init witness (depends on CLI witness flag)
     initLocalWitnesses(config, cmd);
   }
 
@@ -217,7 +220,7 @@ private static void applyStorageConfig(StorageConfig sc) {
     PARAMETER.storage.setIndexSwitch(
         org.apache.commons.lang3.StringUtils.isNotEmpty(indexSwitch) ? indexSwitch : "on");
     PARAMETER.storage.setTransactionHistorySwitch(sc.getTransHistory().getSwitch());
-    // contractParse is set in applyEventConfig — it belongs to event.subscribe domain
+    // contractParse is set in applyConfigParams alongside event config, not here
     PARAMETER.storage.setCheckpointVersion(sc.getCheckpoint().getVersion());
     PARAMETER.storage.setCheckpointSync(sc.getCheckpoint().isSync());
 
@@ -343,21 +346,21 @@ private static void applyRateLimiterConfig(RateLimiterConfig rl) {
     PARAMETER.rateLimiterInitialization = initialization;
   }
 
+  /**
+   * Package-private entry point only for tests
+   */
+  static void applyEventConfig() {
+    applyEventConfig(eventConfig);
+  }
+
   /**
    * Bridge EventConfig bean values to CommonParameter fields.
    * Converts EventConfig (raw bean) into EventPluginConfig and FilterQuery (business objects).
    */
   private static void applyEventConfig(EventConfig ec) {
-    PARAMETER.eventSubscribe = ec.isEnable();
-    // contractParse belongs to event.subscribe but Storage object holds it
-    PARAMETER.storage.setContractParseSwitch(ec.isContractParse());
-
-    // PARAMETER.eventPluginConfig and PARAMETER.eventFilter are only consumed by
-    // Manager.startEventSubscribing(), which itself is gated by isEventSubscribe()
-    // (= ec.isEnable()) at Manager.java:564. When subscribe is disabled, building
-    // these objects has no observable effect — skip both early so PARAMETER stays
-    // consistent with the runtime intent.
-    if (!ec.isEnable()) {
+    // cmd parameter has higher priority
+    PARAMETER.eventSubscribe = PARAMETER.eventSubscribe || ec.isEnable();
+    if (!PARAMETER.eventSubscribe) {
       return;
     }
 
@@ -770,9 +773,12 @@ public static void applyConfigParams(
 
     // node.shutdown — handled in applyNodeConfig
 
-    // Event config: bind from config.conf "event.subscribe" section
+    // Event config: read bean here; applyEventConfig() is called once in setParam()
+    // after applyCLIParams() so that --es is already reflected in eventSubscribe.
     eventConfig = EventConfig.fromConfig(config);
-    applyEventConfig(eventConfig);
+    // contractParse is event-domain but must be set from config before CLI can
+    // override it with --contract-parse-enable (which runs in applyCLIParams).
+    PARAMETER.storage.setContractParseSwitch(eventConfig.isContractParse());
 
     logConfig();
   }

framework/src/main/java/org/tron/core/net/peer/PeerConnection.java

@@ -170,7 +170,8 @@ public class PeerConnection {
 
   public void setChannel(Channel channel) {
     this.channel = channel;
-    if (relayNodes.stream().anyMatch(n -> n.getAddress().equals(channel.getInetAddress()))) {
+    if (relayNodes != null
+        && relayNodes.stream().anyMatch(n -> n.getAddress().equals(channel.getInetAddress()))) {
       this.isRelayPeer = true;
     }
     this.nodeStatistics = TronStatsManager.getNodeStatistics(channel.getInetAddress());

framework/src/test/java/org/tron/common/logsfilter/EventLoaderTest.java

@@ -3,11 +3,16 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import java.util.ArrayList;
 import java.util.List;
 import org.junit.Assert;
 import org.junit.Test;
+import org.pf4j.PluginDescriptor;
+import org.pf4j.PluginManager;
+import org.pf4j.PluginWrapper;
 import org.tron.common.logsfilter.trigger.BlockLogTrigger;
 import org.tron.common.logsfilter.trigger.TransactionLogTrigger;
 
@@ -48,6 +53,32 @@ public void launchNativeQueue() {
     EventPluginLoader.getInstance().stopPlugin();
   }
 
+  @Test
+  public void testIsPluginVersionSupported() {
+    assertEquals("3.0.0", EventPluginLoader.MIN_PLUGIN_VERSION);
+    // last releases before fastjson removal — must be rejected
+    assertFalse(checkVersion("1.0.0"));
+    assertFalse(checkVersion("2.2.0"));
+    assertFalse(checkVersion("2.9.9"));
+    // 3.0.0 onward — must be accepted
+    assertTrue(checkVersion("3.0.0"));
+    assertTrue(checkVersion("3.1.5"));
+    assertTrue(checkVersion("10.0.0"));
+    // empty/null version — reject
+    assertFalse(checkVersion(""));
+    assertFalse(checkVersion(null));
+  }
+
+  private static boolean checkVersion(String version) {
+    PluginManager pm = mock(PluginManager.class);
+    PluginWrapper wrapper = mock(PluginWrapper.class);
+    PluginDescriptor desc = mock(PluginDescriptor.class);
+    when(pm.getPlugin("test")).thenReturn(wrapper);
+    when(wrapper.getDescriptor()).thenReturn(desc);
+    when(desc.getVersion()).thenReturn(version);
+    return EventPluginLoader.isPluginVersionSupported(pm, "test");
+  }
+
   @Test
   public void testBlockLogTrigger() {
     BlockLogTrigger blt = new BlockLogTrigger();

framework/src/test/java/org/tron/core/config/args/ArgsTest.java

@@ -344,6 +344,21 @@ public void testCliEsOverridesConfig() {
     Args.clearParam();
   }
 
+  /**
+   * Regression: when --es is the sole source of event.subscribe.enable=true
+   * (config has it disabled), eventPluginConfig must be built.
+   * Previously applyEventConfig() ran before applyCLIParams() and returned
+   * early (both flags false), leaving eventPluginConfig=null; Manager then
+   * called EventPluginLoader.start(null) and threw "Failed to load eventPlugin."
+   */
+  @Test
+  public void testCliEsBuildsEventPluginConfig() {
+    Args.setParam(new String[] {"--es"}, TestConstants.TEST_CONF);
+    Assert.assertTrue(Args.getInstance().isEventSubscribe());
+    Assert.assertNotNull(Args.getInstance().getEventPluginConfig());
+    Args.clearParam();
+  }
+
   /**
    * Verify that config file storage values are applied when no CLI override is present.
    *
@@ -454,6 +469,7 @@ public void testEventConfigDisabledSkipsEpcAndFilter() {
     Config config = ConfigFactory.parseMap(override)
         .withFallback(ConfigFactory.defaultReference());
     Args.applyConfigParams(config);
+    Args.applyEventConfig();
     Assert.assertNull(Args.getInstance().getEventPluginConfig());
     Assert.assertNull(Args.getInstance().getEventFilter());
     Args.clearParam();
@@ -467,6 +483,7 @@ public void testEventConfigEnabledBuildsEpcAndFilter() {
     Config config = ConfigFactory.parseMap(override)
         .withFallback(ConfigFactory.defaultReference());
     Args.applyConfigParams(config);
+    Args.applyEventConfig();
     Assert.assertNotNull(Args.getInstance().getEventPluginConfig());
     Assert.assertNotNull(Args.getInstance().getEventFilter());
     Args.clearParam();
@@ -481,6 +498,7 @@ public void testEventConfigEnabledWithInvalidFromBlockLeavesFilterNull() {
     Config config = ConfigFactory.parseMap(override)
         .withFallback(ConfigFactory.defaultReference());
     Args.applyConfigParams(config);
+    Args.applyEventConfig();
     // epc still built; filter rejected
     Assert.assertNotNull(Args.getInstance().getEventPluginConfig());
     Assert.assertNull(Args.getInstance().getEventFilter());

framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java

@@ -3,18 +3,32 @@
 import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
+import org.junit.AfterClass;
 import org.junit.Assert;
+import org.junit.BeforeClass;
 import org.junit.Test;
+import org.tron.common.TestConstants;
 import org.tron.common.utils.Pair;
 import org.tron.core.capsule.BlockCapsule.BlockId;
 import org.tron.core.config.Parameter.NetConstants;
+import org.tron.core.config.args.Args;
 import org.tron.core.exception.P2pException;
 import org.tron.core.net.message.keepalive.PingMessage;
 import org.tron.core.net.message.sync.ChainInventoryMessage;
 import org.tron.core.net.peer.PeerConnection;
 
 public class ChainInventoryMsgHandlerTest {
 
+  @BeforeClass
+  public static void init() {
+    Args.setParam(new String[]{}, TestConstants.TEST_CONF);
+  }
+
+  @AfterClass
+  public static void destroy() {
+    Args.clearParam();
+  }
+
   private ChainInventoryMsgHandler handler = new ChainInventoryMsgHandler();
   private PeerConnection peer = new PeerConnection();
   private ChainInventoryMessage msg = new ChainInventoryMessage(new ArrayList<>(), 0L);