Cleaned multiple potential problems in pipe module (#17396)

* fix

* fix

* fix

* fix

* gras-shop

* fix

* spls

* fix

* pipe-dn

* logger-bug

* fix

Author: Caideyipi

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeat.java

@@ -55,7 +55,7 @@ public PipeHeartbeat(
       // the final results and namely these dataNodes are omitted in calculation.
       remainingEventCountMap.put(
           pipeMeta.getStaticMeta(),
-          Objects.nonNull(pipeCompletedListFromAgent)
+          Objects.nonNull(pipeRemainingEventCountListFromAgent)
               ? pipeRemainingEventCountListFromAgent.get(i)
               : 0L);
       remainingTimeMap.put(

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/IoTDBConfigRegionSource.java

@@ -153,8 +153,10 @@ protected void triggerSnapshot() {
   @Override
   public synchronized EnrichedEvent supply() throws Exception {
     final EnrichedEvent event = super.supply();
-    PipeEventCommitManager.getInstance()
-        .enrichWithCommitterKeyAndCommitId(event, creationTime, regionId);
+    if (Objects.nonNull(event)) {
+      PipeEventCommitManager.getInstance()
+          .enrichWithCommitterKeyAndCommitId(event, creationTime, regionId);
+    }
     return event;
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncher.java

@@ -78,7 +78,7 @@ public static synchronized void launchPipePluginAgent(
         curList.add(uninstalledOrConflictedPipePluginMetaList.get(index + offset));
         offset++;
       }
-      index += (offset + 1);
+      index += offset;
       fetchAndSavePipePluginJars(curList);
     }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java

@@ -706,7 +706,7 @@ protected void calculateMemoryUsage(
               MESSAGE_PIPE_NOT_ENOUGH_MEMORY,
               needMemory,
               freeMemorySizeInBytes,
-              freeMemorySizeInBytes,
+              reservedMemorySizeInBytes,
               PipeDataNodeResourceManager.memory().getTotalMemorySizeInBytes());
       LOGGER.warn(message);
       throw new PipeException(message);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java

@@ -238,7 +238,7 @@ private void collectEvent(final Event event) {
       ((PipeHeartbeatEvent) event).recordConnectorQueueSize(pendingQueue);
     }
 
-    pendingQueue.directOffer(event);
+    pendingQueue.offer(event);
     collectInvocationCount.incrementAndGet();
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java

@@ -72,7 +72,7 @@ public PipeRealtimePriorityBlockingQueue() {
   }
 
   @Override
-  public boolean directOffer(final Event event) {
+  public boolean offer(final Event event) {
     checkBeforeOffer(event);
 
     if (event instanceof TsFileInsertionEvent) {
@@ -85,18 +85,13 @@ public boolean directOffer(final Event event) {
       ((EnrichedEvent) event).decreaseReferenceCount(PipeEventCollector.class.getName(), false);
       return false;
     } else {
-      return super.directOffer(event);
+      return super.offer(event);
     }
   }
 
-  @Override
-  public boolean waitedOffer(final Event event) {
-    return directOffer(event);
-  }
-
   @Override
   public boolean put(final Event event) {
-    directOffer(event);
+    offer(event);
     return true;
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/deletion/PipeDeleteDataNodeEvent.java

@@ -31,7 +31,6 @@
 import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResource;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractDeleteDataNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode;
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
 import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry;
@@ -193,7 +192,7 @@ public ByteBuffer serializeToByteBuffer() {
   @Override
   public void deserializeFromByteBuffer(final ByteBuffer buffer) {
     isGeneratedByPipe = ReadWriteIOUtils.readBool(buffer);
-    deleteDataNode = (DeleteDataNode) PlanNodeType.deserialize(buffer);
+    deleteDataNode = (AbstractDeleteDataNode) PlanNodeType.deserialize(buffer);
     progressIndex = deleteDataNode.getProgressIndex();
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java

@@ -138,7 +138,7 @@ public void unbindFrom(final AbstractMetricService metricService) {
     // phantom reference count
     metricService.remove(MetricType.AUTO_GAUGE, Metric.PIPE_PHANTOM_REFERENCE_COUNT.toString());
 
-    metricService.remove(MetricType.RATE, Metric.PIPE_TSFILE_SEND_DISK_IO.toString());
+    metricService.remove(MetricType.COUNTER, Metric.PIPE_TSFILE_SEND_DISK_IO.toString());
   }
 
   public void recordDiskIO(final long bytes) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/receiver/PipeDataNodeReceiverMetrics.java

@@ -338,14 +338,14 @@ private void unbind(final AbstractMetricService metricService) {
         Tag.NAME.toString(),
         RECEIVER,
         Tag.TYPE.toString(),
-        "handshakeDatanodeV1");
+        "handshakeDataNodeV1");
     metricService.remove(
         MetricType.TIMER,
         Metric.PIPE_DATANODE_RECEIVER.toString(),
         Tag.NAME.toString(),
         RECEIVER,
         Tag.TYPE.toString(),
-        "handshakeDatanodeV2");
+        "handshakeDataNodeV2");
     metricService.remove(
         MetricType.TIMER,
         Metric.PIPE_DATANODE_RECEIVER.toString(),

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java

@@ -744,7 +744,7 @@ public void collectWindowOutputs(
               throw new UnsupportedOperationException(
                   String.format(
                       "The output tablet does not support column type %s",
-                      valueColumnTypes[rowIndex]));
+                      valueColumnTypes[columnIndex]));
           }
         } else {
           bitMaps[columnIndex].mark(rowIndex);
@@ -758,7 +758,7 @@ public void collectWindowOutputs(
     int filteredCount = 0;
     for (int i = 0; i < columnNameStringList.length; ++i) {
       if (!bitMaps[i].isAllMarked()) {
-        originColumnIndex2FilteredColumnIndexMapperList[i] = ++filteredCount;
+        originColumnIndex2FilteredColumnIndexMapperList[i] = filteredCount++;
       }
     }