Fix flaky datanode UTs under parallel surefire forks

Two intermittent failures observed in ut-datanode (forkCount>1):

1. DataRegionTest.tearDown:165 "failed to close some ports"
   EnvironmentUtils.examinePorts() checks 6667/5555/31999/9091 machine-wide
   via connect(). DataNodeInternalRPCServiceImplTest legitimately binds 6667
   through DataRegionConsensusImpl.start() -> IoTConsensusRPCService. A
   sibling fork's tearDown then sees 6667 open and fails -- a cross-fork
   false positive, not a real leak. This was missed by #17698's premise
   ("zero ServerSocket / bind() / TServer.serve() calls in tests") because
   the binding happens indirectly through IConsensus.start().

2. AlignedSeriesScanOperatorTest.batchTest2:423 NPE on null TsBlock
   FullOuterTimeJoinOperator.next() (and AbstractSeriesScanOperator.next())
   legitimately return null or an empty TsBlock when prepareInput() yields
   on its 500ms maxRunTime slice. The test dereferenced tsBlock immediately.

Fixes:

- Introduce EnvironmentUtils.FORK_PORT_OFFSET = ((surefire.forkNumber - 1)
  % 30 + 1) * 1000. Each fork shifts its bound ports AND the corresponding
  examinePorts() check by this offset, so forks live in disjoint port
  namespaces. examinePorts() in fork A only sees ports bound by fork A
  (real leak detection preserved); fork A no longer sees fork B's
  bindings (cross-fork false positive removed). Modulo 30 caps the offset
  so the highest checked port (31999 + 30*1000 = 61999) stays valid.

- DataNodeInternalRPCServiceImplTest binds 6667 + FORK_PORT_OFFSET via the
  same constant -- binding and check stay in sync.

- AlignedSeriesScanOperatorTest.batchTest1/2/3: add
  `if (tsBlock == null || tsBlock.isEmpty()) continue;` after each
  operator.next(), matching the pattern in TreeTopKOperatorTest,
  MergeTreeSortOperatorTest, TreeSortOperatorTest, etc.

Audited iotdb-core/datanode UTs for other monitored port bindings --
none beyond DataNodeInternalRPCServiceImplTest.

Author: JackieTien97

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/AlignedSeriesScanOperatorTest.java

@@ -140,6 +140,9 @@ public void batchTest1() throws Exception {
       int count = 0;
       while (seriesScanOperator.hasNext()) {
         TsBlock tsBlock = seriesScanOperator.next();
+        if (tsBlock == null || tsBlock.isEmpty()) {
+          continue;
+        }
         assertEquals(6, tsBlock.getValueColumnCount());
         assertTrue(tsBlock.getColumn(0) instanceof BooleanColumn);
         assertTrue(tsBlock.getColumn(1) instanceof IntColumn);
@@ -420,6 +423,9 @@ public void batchTest2() throws Exception {
       int count = 0;
       while (timeJoinOperator.isBlocked().isDone() && timeJoinOperator.hasNext()) {
         TsBlock tsBlock = timeJoinOperator.next();
+        if (tsBlock == null || tsBlock.isEmpty()) {
+          continue;
+        }
         assertEquals(18, tsBlock.getValueColumnCount());
         assertTrue(tsBlock.getColumn(0) instanceof BooleanColumn);
         assertTrue(tsBlock.getColumn(1) instanceof IntColumn);
@@ -727,6 +733,9 @@ public void batchTest3() throws Exception {
       int count = 499;
       while (timeJoinOperator.isBlocked().isDone() && timeJoinOperator.hasNext()) {
         TsBlock tsBlock = timeJoinOperator.next();
+        if (tsBlock == null || tsBlock.isEmpty()) {
+          continue;
+        }
         assertEquals(18, tsBlock.getValueColumnCount());
         assertTrue(tsBlock.getColumn(0) instanceof BooleanColumn);
         assertTrue(tsBlock.getColumn(1) instanceof IntColumn);

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java

@@ -93,6 +93,12 @@ public class DataNodeInternalRPCServiceImplTest {
   private static final File storageDir = new File("target" + java.io.File.separator + "impl");
   private static DataRegion dataRegion;
 
+  // Each parallel surefire fork binds its own consensus port. Reuses
+  // EnvironmentUtils.FORK_PORT_OFFSET so this binding and examinePorts() are
+  // guaranteed to look at the same port — fork-local leaks still fail, sibling
+  // forks no longer cause cross-fork false positives.
+  private static final int CONSENSUS_PORT = 6667 + EnvironmentUtils.FORK_PORT_OFFSET;
+
   @BeforeClass
   public static void setUpBeforeClass() throws IOException, MetadataException, ConsensusException {
     // In standalone mode, we need to set dataNodeId to 0 for RaftPeerId in RatisConsensus
@@ -109,7 +115,7 @@ public static void setUpBeforeClass() throws IOException, MetadataException, Con
                 ConsensusFactory.IOT_CONSENSUS,
                 ConsensusConfig.newBuilder()
                     .setThisNodeId(1)
-                    .setThisNode(new TEndPoint("0.0.0.0", 6667))
+                    .setThisNode(new TEndPoint("0.0.0.0", CONSENSUS_PORT))
                     .setStorageDir(storageDir.getAbsolutePath())
                     .setConsensusGroupType(TConsensusGroupType.DataRegion)
                     .build(),
@@ -124,7 +130,8 @@ public static void setUpBeforeClass() throws IOException, MetadataException, Con
         ((IoTConsensus) DataRegionConsensusImpl.getInstance()).getImpl(new DataRegionId(1)))) {
       DataRegionConsensusImpl.getInstance()
           .createLocalPeer(
-              id, Collections.singletonList(new Peer(id, 1, new TEndPoint("0.0.0.0", 6667))));
+              id,
+              Collections.singletonList(new Peer(id, 1, new TEndPoint("0.0.0.0", CONSENSUS_PORT))));
     }
     DataRegionConsensusImpl.getInstance().start();
     SchemaRegionConsensusImpl.getInstance().start();

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java

@@ -96,6 +96,17 @@ public class EnvironmentUtils {
   public static boolean examinePorts =
       Boolean.parseBoolean(System.getProperty("test.port.closed", "false"));
 
+  // Per-fork port namespace. Each surefire fork shifts every test-bound port
+  // (and the corresponding examinePorts() check) by FORK_PORT_OFFSET so that
+  // parallel forks live in disjoint port spaces. This keeps the leak-detection
+  // power of examinePorts() (a fork still sees its own un-closed ports) while
+  // removing the cross-fork false positive (one fork's bound port no longer
+  // looks like another fork's leak). surefire.forkNumber starts at 1; defaults
+  // to 1 outside surefire (IDE). Modulo 30 caps the offset so the highest
+  // checked port (31999 + 30*1000 = 61999) stays within range.
+  public static final int FORK_PORT_OFFSET =
+      ((Integer.parseInt(System.getProperty("surefire.forkNumber", "1")) - 1) % 30 + 1) * 1000;
+
   public static void cleanEnv() throws IOException, StorageEngineException {
     // wait all compaction finished
     CompactionTaskManager.getInstance().waitAllCompactionFinish();
@@ -174,23 +185,31 @@ public static void cleanEnv() throws IOException, StorageEngineException {
   }
 
   private static boolean examinePorts() {
-    TTransport transport = TSocketWrapper.wrap(tConfiguration, "127.0.0.1", 6667, 100);
+    // All checks use this fork's port namespace (base + FORK_PORT_OFFSET) so
+    // that we only ever see ports bound by THIS JVM — sibling forks live in
+    // disjoint namespaces.
+    int rpcPort = 6667 + FORK_PORT_OFFSET;
+    int syncPort = 5555 + FORK_PORT_OFFSET;
+    int jmxPort = 31999 + FORK_PORT_OFFSET;
+    int metricPort = 9091 + FORK_PORT_OFFSET;
+
+    TTransport transport = TSocketWrapper.wrap(tConfiguration, "127.0.0.1", rpcPort, 100);
     if (transport != null && !transport.isOpen()) {
       try {
         transport.open();
-        logger.error("stop daemon failed. 6667 can be connected now.");
+        logger.error("stop daemon failed. {} can be connected now.", rpcPort);
         transport.close();
         return false;
       } catch (TTransportException e) {
         // do nothing
       }
     }
     // try sync service
-    transport = TSocketWrapper.wrap(tConfiguration, "127.0.0.1", 5555, 100);
+    transport = TSocketWrapper.wrap(tConfiguration, "127.0.0.1", syncPort, 100);
     if (transport != null && !transport.isOpen()) {
       try {
         transport.open();
-        logger.error("stop Sync daemon failed. 5555 can be connected now.");
+        logger.error("stop Sync daemon failed. {} can be connected now.", syncPort);
         transport.close();
         return false;
       } catch (TTransportException e) {
@@ -199,18 +218,19 @@ private static boolean examinePorts() {
     }
     // try jmx connection
     try {
-      JMXServiceURL url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://localhost:31999/jmxrmi");
+      JMXServiceURL url =
+          new JMXServiceURL("service:jmx:rmi:///jndi/rmi://localhost:" + jmxPort + "/jmxrmi");
       JMXConnector jmxConnector = JMXConnectorFactory.connect(url);
-      logger.error("stop JMX failed. 31999 can be connected now.");
+      logger.error("stop JMX failed. {} can be connected now.", jmxPort);
       jmxConnector.close();
       return false;
     } catch (IOException e) {
       // do nothing
     }
     // try MetricService
     try (Socket socket = new Socket()) {
-      socket.connect(new InetSocketAddress("127.0.0.1", 9091), 100);
-      logger.error("stop MetricService failed. 9091 can be connected now.");
+      socket.connect(new InetSocketAddress("127.0.0.1", metricPort), 100);
+      logger.error("stop MetricService failed. {} can be connected now.", metricPort);
       return false;
     } catch (Exception e) {
       // do nothing