[core] Parse postpone write id from the writer marker

paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java

@@ -231,8 +231,11 @@ protected Function<WriterContainer<KeyValue>, Boolean> createWriterCleanChecker(
 
     public static int getWriteId(String fileName) {
         try {
-            String[] parts = fileName.split("-s-");
-            return Integer.parseInt(parts[1].substring(0, parts[1].indexOf('-')));
+            // the write id is the segment between the last "-s-" and the writer marker:
+            // the commit user is interpolated verbatim and may itself contain "-s-"
+            int writerMarker = fileName.lastIndexOf("-w-");
+            int sequenceMarker = fileName.lastIndexOf("-s-", writerMarker);
+            return Integer.parseInt(fileName.substring(sequenceMarker + 3, writerMarker));
         } catch (Exception e) {
             throw new RuntimeException(
                     "Data file name "

paimon-core/src/test/java/org/apache/paimon/postpone/PostponeBucketWriterTest.java

@@ -32,6 +32,7 @@
 
 import org.junit.jupiter.api.Test;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.Mockito.mock;
@@ -77,4 +78,20 @@ void testSkipBlobExternalizationWithoutExternalizer() throws Exception {
         verify(writerFactory, never()).externalizeBlob(any(), any());
         verify(rollingWriter).write(same(record));
     }
+
+    @Test
+    void testGetWriteIdWithSeparatorInCommitUser() {
+        // the commit user is interpolated verbatim into the writer prefix, so it may
+        // contain the "-s-" separator itself
+        assertThat(
+                        PostponeBucketFileStoreWrite.getWriteId(
+                                "a-s-123-b-u-my-s-user-s-42-w-data-uuid-0.orc"))
+                .isEqualTo(42);
+        assertThat(PostponeBucketFileStoreWrite.getWriteId("p-u-cu-s-7-w-data-uuid-0.orc"))
+                .isEqualTo(7);
+        // hostile commit user alone: the first split would hit the in-user separator and
+        // crash on the missing dash
+        assertThat(PostponeBucketFileStoreWrite.getWriteId("p-u-my-s-user-s-42-w-data-uuid-0.orc"))
+                .isEqualTo(42);
+    }
 }