Merge remote-tracking branch 'upstream/develop' into feat/opt-api

Author: bladehan1

chainbase/src/main/java/org/tron/core/db/TronDatabase.java

@@ -76,13 +76,25 @@ public void reset() {
   @Override
   public void close() {
     logger.info("******** Begin to close {}. ********", getName());
+    doClose();
+    logger.info("******** End to close {}. ********", getName());
+  }
+
+  /**
+   * Releases writeOptions and dbSource (best-effort, exceptions logged at WARN).
+   * Subclasses with extra resources should override {@link #close()} and call
+   * {@code doClose()} directly — not {@code super.close()} — to avoid duplicated logs.
+   */
+  protected void doClose() {
     try {
       writeOptions.close();
+    } catch (Exception e) {
+      logger.warn("Failed to close writeOptions in {}.", getName(), e);
+    }
+    try {
       dbSource.closeDB();
     } catch (Exception e) {
-      logger.warn("Failed to close {}.", getName(), e);
-    } finally {
-      logger.info("******** End to close {}. ********", getName());
+      logger.warn("Failed to close dbSource in {}.", getName(), e);
     }
   }
 

chainbase/src/main/java/org/tron/core/store/CheckPointV2Store.java

@@ -62,20 +62,16 @@ public void updateByBatch(Map<byte[], byte[]> rows) {
     this.dbSource.updateByBatch(rows, writeOptions);
   }
 
-  /**
-   * close the database.
-   */
   @Override
   public void close() {
     logger.debug("******** Begin to close {}. ********", getName());
     try {
       writeOptions.close();
-      dbSource.closeDB();
     } catch (Exception e) {
-      logger.warn("Failed to close {}.", getName(), e);
-    } finally {
-      logger.debug("******** End to close {}. ********", getName());
+      logger.warn("Failed to close writeOptions in {}.", getName(), e);
     }
+    doClose();
+    logger.debug("******** End to close {}. ********", getName());
   }
 
 }

framework/src/main/java/org/tron/program/SolidityNode.java

@@ -2,7 +2,10 @@
 
 import static org.tron.core.config.Parameter.ChainConstant.BLOCK_PRODUCED_INTERVAL;
 
+import com.google.common.annotations.VisibleForTesting;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.support.DefaultListableBeanFactory;
@@ -11,6 +14,7 @@
 import org.tron.common.application.ApplicationFactory;
 import org.tron.common.application.TronApplicationContext;
 import org.tron.common.client.DatabaseGrpcClient;
+import org.tron.common.es.ExecutorServiceManager;
 import org.tron.common.parameter.CommonParameter;
 import org.tron.common.prometheus.Metrics;
 import org.tron.core.ChainBaseManager;
@@ -39,6 +43,9 @@ public class SolidityNode {
 
   private volatile boolean flag = true;
 
+  private ExecutorService getBlockEs;
+  private ExecutorService processBlockEs;
+
   public SolidityNode(Manager dbManager) {
     this.dbManager = dbManager;
     this.chainBaseManager = dbManager.getChainBaseManager();
@@ -72,13 +79,26 @@ public static void start() {
     appT.startup();
     SolidityNode node = new SolidityNode(appT.getDbManager());
     node.run();
-    appT.blockUntilShutdown();
+    awaitShutdown(appT, node);
+  }
+
+  @VisibleForTesting
+  static void awaitShutdown(Application appT, SolidityNode node) {
+    try {
+      appT.blockUntilShutdown();
+    } finally {
+      // SolidityNode is created manually rather than managed by Spring/Application,
+      // so its executors must be shut down explicitly on exit.
+      node.shutdown();
+    }
   }
 
   private void run() {
     try {
-      new Thread(this::getBlock).start();
-      new Thread(this::processBlock).start();
+      getBlockEs = ExecutorServiceManager.newSingleThreadExecutor("solid-get-block");
+      processBlockEs = ExecutorServiceManager.newSingleThreadExecutor("solid-process-block");
+      getBlockEs.execute(this::getBlock);
+      processBlockEs.execute(this::processBlock);
       logger.info("Success to start solid node, ID: {}, remoteBlockNum: {}.", ID.get(),
           remoteBlockNum);
     } catch (Exception e) {
@@ -88,6 +108,15 @@ private void run() {
     }
   }
 
+  public void shutdown() {
+    flag = false;
+    // Signal both pools before awaiting either so they drain concurrently
+    getBlockEs.shutdown();
+    processBlockEs.shutdown();
+    ExecutorServiceManager.shutdownAndAwaitTermination(getBlockEs, "solid-get-block");
+    ExecutorServiceManager.shutdownAndAwaitTermination(processBlockEs, "solid-process-block");
+  }
+
   private void getBlock() {
     long blockNum = ID.incrementAndGet();
     while (flag) {
@@ -137,7 +166,7 @@ private void loopProcessBlock(Block block) {
   }
 
   private Block getBlockByNum(long blockNum) {
-    while (true) {
+    while (flag) {
       try {
         long time = System.currentTimeMillis();
         Block block = databaseGrpcClient.getBlock(blockNum);
@@ -155,10 +184,11 @@ private Block getBlockByNum(long blockNum) {
         sleep(exceptionSleepTime);
       }
     }
+    return null;
   }
 
   private long getLastSolidityBlockNum() {
-    while (true) {
+    while (flag) {
       try {
         long time = System.currentTimeMillis();
         long blockNum = databaseGrpcClient.getDynamicProperties().getLastSolidityBlockNum();
@@ -171,6 +201,7 @@ private long getLastSolidityBlockNum() {
         sleep(exceptionSleepTime);
       }
     }
+    return 0;
   }
 
   public void sleep(long time) {
@@ -193,4 +224,4 @@ private void resolveCompatibilityIssueIfUsingFullNodeDatabase() {
       chainBaseManager.getDynamicPropertiesStore().saveLatestSolidifiedBlockNum(headBlockNum);
     }
   }
-}
+}

framework/src/test/java/org/tron/common/ClassLevelAppContextFixture.java

@@ -0,0 +1,62 @@
+package org.tron.common;
+
+import io.grpc.ManagedChannel;
+import java.util.concurrent.TimeUnit;
+import org.tron.common.application.ApplicationFactory;
+import org.tron.common.application.TronApplicationContext;
+import org.tron.core.config.DefaultConfig;
+
+/**
+ * Shared class-level fixture for tests that manually manage a TronApplicationContext.
+ */
+public class ClassLevelAppContextFixture {
+
+  private TronApplicationContext context;
+
+  public TronApplicationContext createContext() {
+    context = new TronApplicationContext(DefaultConfig.class);
+    return context;
+  }
+
+  public TronApplicationContext createAndStart() {
+    createContext();
+    startApp();
+    return context;
+  }
+
+  public void startApp() {
+    ApplicationFactory.create(context).startup();
+  }
+
+  public TronApplicationContext getContext() {
+    return context;
+  }
+
+  public void close() {
+    if (context != null) {
+      context.close();
+      context = null;
+    }
+  }
+
+  public static void shutdownChannel(ManagedChannel channel) {
+    if (channel == null) {
+      return;
+    }
+    try {
+      channel.shutdown();
+      if (!channel.awaitTermination(5, TimeUnit.SECONDS)) {
+        channel.shutdownNow();
+      }
+    } catch (InterruptedException e) {
+      channel.shutdownNow();
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  public static void shutdownChannels(ManagedChannel... channels) {
+    for (ManagedChannel channel : channels) {
+      shutdownChannel(channel);
+    }
+  }
+}

framework/src/test/java/org/tron/common/backup/BackupServerTest.java

@@ -46,7 +46,7 @@ public void tearDown() {
   @Test(timeout = 60_000)
   public void test() throws InterruptedException {
     backupServer.initServer();
-    // wait for the server to start
+    // wait for the server to start so channel is assigned before close() is called
     Thread.sleep(1000);
   }
 }

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

@@ -65,7 +65,6 @@ public synchronized void testEventParser() {
 
     for (int i = 0; i < entryArr.size(); i++) {
       JSONObject e = entryArr.getJSONObject(i);
-      System.out.println(e.getString("name"));
       if (e.getString("name") != null) {
         if (e.getString("name").equalsIgnoreCase("eventBytesL")) {
           entry = e;

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

@@ -68,7 +68,6 @@ public synchronized void testEventParser() {
 
     ABI.Entry entry = null;
     for (ABI.Entry e : abi.getEntrysList()) {
-      System.out.println(e.getName());
       if (e.getName().equalsIgnoreCase("eventBytesL")) {
         entry = e;
         break;

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

@@ -17,6 +17,7 @@
 import java.util.List;
 import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
+import org.junit.After;
 import org.junit.Assert;
 import org.junit.Test;
 import org.tron.common.logsfilter.capsule.ContractEventTriggerCapsule;
@@ -28,6 +29,11 @@
 @Slf4j
 public class FilterQueryTest {
 
+  @After
+  public void tearDown() {
+    EventPluginLoader.getInstance().setFilterQuery(null);
+  }
+
   @Test
   public synchronized void testParseFilterQueryBlockNumber() {
     assertEquals(LATEST_BLOCK_NUM, parseToBlockNumber(EMPTY));

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

@@ -1,7 +1,10 @@
 package org.tron.common.logsfilter;
 
+import java.util.concurrent.ExecutorService;
+import org.junit.After;
 import org.junit.Assert;
 import org.junit.Test;
+import org.tron.common.es.ExecutorServiceManager;
 import org.tron.common.logsfilter.nativequeue.NativeMessageQueue;
 import org.zeromq.SocketType;
 import org.zeromq.ZContext;
@@ -13,6 +16,15 @@ public class NativeMessageQueueTest {
   public String dataToSend = "################";
   public String topic = "testTopic";
 
+  private ExecutorService subscriberExecutor;
+  private final String zmqSubscriber = "zmq-subscriber";
+
+  @After
+  public void tearDown() {
+    ExecutorServiceManager.shutdownAndAwaitTermination(subscriberExecutor, zmqSubscriber);
+    subscriberExecutor = null;
+  }
+
   @Test
   public void invalidBindPort() {
     boolean bRet = NativeMessageQueue.getInstance().start(-1111, 0);
@@ -39,22 +51,23 @@ public void publishTrigger() {
     try {
       Thread.sleep(1000);
     } catch (InterruptedException e) {
-      e.printStackTrace();
+      Thread.currentThread().interrupt();
     }
 
     NativeMessageQueue.getInstance().publishTrigger(dataToSend, topic);
 
     try {
       Thread.sleep(1000);
     } catch (InterruptedException e) {
-      e.printStackTrace();
+      Thread.currentThread().interrupt();
     }
 
     NativeMessageQueue.getInstance().stop();
   }
 
   public void startSubscribeThread() {
-    Thread thread = new Thread(() -> {
+    subscriberExecutor = ExecutorServiceManager.newSingleThreadExecutor(zmqSubscriber);
+    subscriberExecutor.execute(() -> {
       try (ZContext context = new ZContext()) {
         ZMQ.Socket subscriber = context.createSocket(SocketType.SUB);
 
@@ -70,6 +83,5 @@ public void startSubscribeThread() {
         // ZMQ.Socket will be automatically closed when ZContext is closed
       }
     });
-    thread.start();
   }
 }

framework/src/test/java/org/tron/common/logsfilter/capsule/BlockFilterCapsuleTest.java

@@ -22,7 +22,6 @@ public void setUp() {
   public void testSetAndGetBlockHash() {
     blockFilterCapsule
         .setBlockHash("e58f33f9baf9305dc6f82b9f1934ea8f0ade2defb951258d50167028c780351f");
-    System.out.println(blockFilterCapsule);
     Assert.assertEquals("e58f33f9baf9305dc6f82b9f1934ea8f0ade2defb951258d50167028c780351f",
         blockFilterCapsule.getBlockHash());
   }