Escape last segment path as whole

sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Lineage.java

@@ -73,13 +73,14 @@ public class Lineage {
       }
       if (lastSegmentSep != null) {
         List<String> subSegments =
-            Splitter.onPattern(lastSegmentSep).splitToList(segments.get(segments.size() - 1));
+            Splitter.onPattern(lastSegmentSep)
+                .splitToList(wrapSegment(segments.get(segments.size() - 1)));
         for (int i = 0; i < subSegments.size() - 1; i++) {
           parts.add(subSegments.get(i) + lastSegmentSep);
         }
-        parts.add(wrapSegment(subSegments.get(subSegments.size() - 1)));
+        parts.add(subSegments.get(subSegments.size() - 1));
       } else {
-        parts.add(wrapSegment(segments.get(segments.size() - 1)));
+        parts.add(segments.get(segments.size() - 1));
       }
     }
     return parts.iterator();

sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/LineageTest.java

@@ -85,6 +85,10 @@ public class LineageTest {
     Iterator<String> pathFQN =
         Lineage.getFQNParts("system", null, ImmutableList.of("bucket", "dir1/dir2/file"), "/");
     assertEquals("system:bucket.dir1/dir2/file", getFqnPartsString(pathFQN));
+
+    Iterator<String> pathFQNReserved =
+        Lineage.getFQNParts("system", null, ImmutableList.of("bucket", "dir1/dir.2/file"), "/");
+    assertEquals("system:bucket.`dir1/dir.2/file`", getFqnPartsString(pathFQNReserved));
   }
 
   private static String getFqnPartsString(Iterator<String> fqnPartsIterator) {