Pipe: Fix epoch status metric (apache#16355)

* fix

* fix

Author: XNX02

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpoch.java

@@ -43,9 +43,16 @@ public TsFileEpoch(final TsFileResource resource) {
   }
 
   public TsFileEpoch.State getState(final PipeRealtimeDataRegionSource extractor) {
-    return dataRegionExtractor2State
-        .computeIfAbsent(extractor, o -> new AtomicReference<>(State.EMPTY))
-        .get();
+    AtomicReference<State> stateRef = dataRegionExtractor2State.get(extractor);
+
+    if (stateRef == null) {
+      dataRegionExtractor2State.putIfAbsent(
+          extractor, stateRef = new AtomicReference<>(State.EMPTY));
+      extractor.increaseExtractEpochSize();
+      setExtractorsRecentProcessedTsFileEpochState();
+    }
+
+    return stateRef.get();
   }
 
   public void migrateState(