diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
index cb6994ed44fe..28f29c2cb943 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
@@ -100,7 +100,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
   private final PipeMetaKeeper pipeMetaKeeper;
 
   // Tracks runtime exception changes that have not yet been persisted through consensus.
-  private volatile boolean pipeMetaExceptionPersistencePending;
+  private final AtomicBoolean pipeMetaExceptionPersistencePending = new AtomicBoolean(false);
 
   // Pure in-memory object, not involved in snapshot serialization and deserialization.
   private final PipeTaskInfoVersion pipeTaskInfoVersion;
@@ -1171,7 +1171,9 @@ public boolean recordDataNodePushPipeMetaExceptions(
     try {
       final boolean hasException =
           recordDataNodePushPipeMetaExceptionsInternal(respMap, fallbackPipeStaticMeta);
-      pipeMetaExceptionPersistencePending |= hasException;
+      if (hasException) {
+        pipeMetaExceptionPersistencePending.set(true);
+      }
       return hasException;
     } finally {
       releaseWriteLock();
@@ -1179,11 +1181,11 @@ public boolean recordDataNodePushPipeMetaExceptions(
   }
 
   public boolean hasPendingPipeMetaExceptionPersistence() {
-    return pipeMetaExceptionPersistencePending;
+    return pipeMetaExceptionPersistencePending.get();
   }
 
   public void markPipeMetaExceptionsPersisted() {
-    pipeMetaExceptionPersistencePending = false;
+    pipeMetaExceptionPersistencePending.set(false);
   }
 
   private boolean recordDataNodePushPipeMetaExceptionsInternal(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java
index 5c8f13cdd0b3..1ac7dd1d8940 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java
@@ -37,6 +37,7 @@
 import java.util.EnumMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 public class PipeResourceMetrics implements IMetricSet {
 
@@ -65,8 +66,9 @@ public enum TsFileParserAdmissionRejectReason {
     }
   }
 
-  private volatile Map<TsFileParserAdmissionRejectReason, Counter>
-      tsFileParserAdmissionRejectCounters = Collections.emptyMap();
+  // Replaced as a whole on bind/unbind; the published map is never modified.
+  private final AtomicReference<Map<TsFileParserAdmissionRejectReason, Counter>>
+      tsFileParserAdmissionRejectCounters = new AtomicReference<>(Collections.emptyMap());
   private volatile Counter tsFileParserAdmissionYieldCounter =
       DoNothingMetricManager.DO_NOTHING_COUNTER;
   private volatile Timer tsFileParserAdmissionWaitTimer = DoNothingMetricManager.DO_NOTHING_TIMER;
@@ -144,7 +146,7 @@ public void bindTo(final AbstractMetricService metricService) {
               Tag.REASON.toString(),
               reason.tag));
     }
-    tsFileParserAdmissionRejectCounters = rejectCounters;
+    tsFileParserAdmissionRejectCounters.set(Collections.unmodifiableMap(rejectCounters));
     tsFileParserAdmissionYieldCounter =
         metricService.getOrCreateCounter(
             Metric.PIPE_TSFILE_PARSER_ADMISSION_YIELD.toString(), MetricLevel.IMPORTANT);
@@ -210,7 +212,7 @@ public void unbindFrom(final AbstractMetricService metricService) {
     metricService.remove(MetricType.TIMER, Metric.PIPE_TSFILE_PARSER_ADMISSION_WAIT.toString());
     metricService.remove(MetricType.AUTO_GAUGE, Metric.PIPE_TSFILE_PARSER_IN_FLIGHT.toString());
     metricService.remove(MetricType.AUTO_GAUGE, Metric.PIPE_TSFILE_PARSER_WAITING.toString());
-    tsFileParserAdmissionRejectCounters = Collections.emptyMap();
+    tsFileParserAdmissionRejectCounters.set(Collections.emptyMap());
     tsFileParserAdmissionRejectCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
     tsFileParserAdmissionYieldCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
     tsFileParserAdmissionWaitTimer = DoNothingMetricManager.DO_NOTHING_TIMER;
@@ -223,6 +225,7 @@ public void recordDiskIO(final long bytes) {
   public void recordTsFileParserAdmissionReject(final TsFileParserAdmissionRejectReason reason) {
     tsFileParserAdmissionRejectCounter.inc();
     tsFileParserAdmissionRejectCounters
+        .get()
         .getOrDefault(reason, DoNothingMetricManager.DO_NOTHING_COUNTER)
         .inc();
   }