fix

Author: Caideyipi

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

@@ -364,7 +364,8 @@ public void discardEventsOfPipe(final CommitterKey committerKey) {
     super.discardEventsOfPipe(committerKey);
     tsfileInsertEventDeque.removeIf(
         event -> {
-          if (event instanceof EnrichedEvent && isEventFromPipe((EnrichedEvent) event, committerKey)) {
+          if (event instanceof EnrichedEvent
+              && isEventFromPipe((EnrichedEvent) event, committerKey)) {
             if (((EnrichedEvent) event)
                 .clearReferenceCount(PipeRealtimePriorityBlockingQueue.class.getName())) {
               eventCounter.decreaseEventCount(event);

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

@@ -257,8 +257,7 @@ private static boolean isEventFromPipe(
     return committerKey.getPipeName().equals(event.getPipeName())
         && committerKey.getCreationTime() == event.getCreationTime()
         && committerKey.getRegionId() == event.getRegionId()
-        && (committerKey.getRestartTimes() < 0
-            || committerKey.equals(event.getCommitterKey()));
+        && (committerKey.getRestartTimes() < 0 || committerKey.equals(event.getCommitterKey()));
   }
 
   //////////////////////////// APIs provided for metric framework ////////////////////////////

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java

@@ -177,8 +177,7 @@ private static boolean isEventFromPipe(
     return committerKey.getPipeName().equals(event.getPipeName())
         && committerKey.getCreationTime() == event.getCreationTime()
         && committerKey.getRegionId() == event.getRegionId()
-        && (committerKey.getRestartTimes() < 0
-            || committerKey.equals(event.getCommitterKey()));
+        && (committerKey.getRestartTimes() < 0 || committerKey.equals(event.getCommitterKey()));
   }
 
   public synchronized void decreaseEventsReferenceCount(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java

@@ -880,8 +880,7 @@ private static boolean isDroppedPipe(final EnrichedEvent event, final CommitterK
     return committerKey.getPipeName().equals(event.getPipeName())
         && committerKey.getCreationTime() == event.getCreationTime()
         && committerKey.getRegionId() == event.getRegionId()
-        && (committerKey.getRestartTimes() < 0
-            || committerKey.equals(event.getCommitterKey()));
+        && (committerKey.getRestartTimes() < 0 || committerKey.equals(event.getCommitterKey()));
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketConnectorServer.java

@@ -529,8 +529,7 @@ private static boolean isEventFromPipe(
     return committerKey.getPipeName().equals(event.getPipeName())
         && committerKey.getCreationTime() == event.getCreationTime()
         && committerKey.getRegionId() == event.getRegionId()
-        && (committerKey.getRestartTimes() < 0
-            || committerKey.equals(event.getCommitterKey()));
+        && (committerKey.getRestartTimes() < 0 || committerKey.equals(event.getCommitterKey()));
   }
 
   private boolean isQueueAvailable(

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java

@@ -144,7 +144,8 @@ public void discardEventsOfPipe(final CommitterKey committerKey) {
     droppedPipeTaskKeys.add(committerKey);
     pendingQueue.removeIf(
         event -> {
-          if (event instanceof EnrichedEvent && isEventFromPipe((EnrichedEvent) event, committerKey)) {
+          if (event instanceof EnrichedEvent
+              && isEventFromPipe((EnrichedEvent) event, committerKey)) {
             if (((EnrichedEvent) event).clearReferenceCount(BlockingPendingQueue.class.getName())) {
               eventCounter.decreaseEventCount(event);
             }
@@ -197,8 +198,7 @@ protected static boolean isEventFromPipe(
     return committerKey.getPipeName().equals(event.getPipeName())
         && committerKey.getCreationTime() == event.getCreationTime()
         && committerKey.getRegionId() == event.getRegionId()
-        && (committerKey.getRestartTimes() < 0
-            || committerKey.equals(event.getCommitterKey()));
+        && (committerKey.getRestartTimes() < 0 || committerKey.equals(event.getCommitterKey()));
   }
 
   protected boolean isEventFromDroppedPipe(final E event) {