Pipe: Reduce the amount of log information printed after a batch fails (#14172)

Co-authored-by: Steve Yurong Su <[email protected]>

Author: luoluoyuyu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/handler/PipeConsensusTabletBatchEventHandler.java

@@ -117,15 +117,15 @@ public void onComplete(final TPipeConsensusBatchTransferResp response) {
   @Override
   public void onError(final Exception exception) {
     LOGGER.warn(
-        "PipeConsensus: Failed to transfer TabletInsertionEvent batch {} (request commit ids={}).",
+        "PipeConsensus: Failed to transfer TabletInsertionEvent batch. Total failed events: {}, related pipe names: {}",
+        events.size(),
         events.stream()
             .map(
                 event ->
                     event instanceof EnrichedEvent
-                        ? ((EnrichedEvent) event).coreReportMessage()
-                        : event.toString())
-            .collect(Collectors.toList()),
-        requestCommitIds,
+                        ? ((EnrichedEvent) event).getPipeName()
+                        : "UNKNOWN")
+            .collect(Collectors.toSet()),
         exception);
 
     connector.addFailureEventsToRetryQueue(events);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java

@@ -121,9 +121,9 @@ public void onComplete(final TPipeTransferResp response) {
   public void onError(final Exception exception) {
     try {
       LOGGER.warn(
-          "Failed to transfer TabletInsertionEvent batch {} (request commit ids={}).",
-          events.stream().map(EnrichedEvent::coreReportMessage).collect(Collectors.toList()),
-          events.stream().map(EnrichedEvent::getCommitId).collect(Collectors.toList()),
+          "Failed to transfer TabletInsertionEvent batch. Total failed events: {}, related pipe names: {}",
+          events.size(),
+          events.stream().map(EnrichedEvent::getPipeName).collect(Collectors.toSet()),
           exception);
     } finally {
       connector.addFailureEventsToRetryQueue(events);