[fix][io] Fix the Alluxio sink to write messages successfully after the first file rotation (#19247)

Author: sekikn

pulsar-io/alluxio/src/main/java/org/apache/pulsar/io/alluxio/sink/AlluxioSink.java

@@ -67,7 +67,7 @@ public class AlluxioSink implements Sink<GenericObject> {
 
     private FileSystem fileSystem;
     private FileOutStream fileOutStream;
-    private CreateFilePOptions.Builder optionsBuilder;
+
     private long recordsNum;
     private String tmpFilePath;
     private String fileDirPath;
@@ -114,8 +114,6 @@ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exc
             fileSystem.createDirectory(tmpAlluxioDirPath);
         }
 
-        optionsBuilder = FileSystemOptions.createFileDefaults(configuration).toBuilder();
-
         recordsNum = 0;
         recordsToAck = Lists.newArrayList();
         tmpFilePath = "";
@@ -206,6 +204,8 @@ private void writeToAlluxio(Record<GenericObject> record) throws AlluxioExceptio
     }
 
     private void createTmpFile() throws AlluxioException, IOException {
+        CreateFilePOptions.Builder optionsBuilder =
+                FileSystemOptions.createFileDefaults(configuration).toBuilder();
         UUID id = UUID.randomUUID();
         String fileExtension = alluxioSinkConfig.getFileExtension();
         tmpFilePath = tmpFileDirPath + "/" + id.toString() + "_tmp" + fileExtension;

pulsar-io/alluxio/src/test/java/org/apache/pulsar/io/alluxio/sink/AlluxioSinkTest.java

@@ -40,6 +40,7 @@
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
@@ -125,6 +126,13 @@ public Object getNativeObject() {
         when(mockRecord.getSchema()).thenAnswer((Answer<Schema<KeyValue<String, Foobar>>>) invocation -> kvSchema);
     }
 
+    @AfterMethod
+    public void tearDown() throws Exception {
+        if (cluster != null) {
+            cluster.stop();
+        }
+    }
+
     @Test
     public void openTest() throws Exception {
         map.put("filePrefix", "TopicA");
@@ -147,7 +155,6 @@ public void openTest() throws Exception {
         Assert.assertTrue(client.exists(alluxioTmpURI));
 
         sink.close();
-        cluster.stop();
     }
 
     @Test
@@ -186,11 +193,36 @@ public Object getNativeObject() {
         Assert.assertTrue(client.exists(alluxioTmpURI));
 
         List<URIStatus> listAlluxioDirStatus = client.listStatus(alluxioURI);
-
         List<String> pathList = listAlluxioDirStatus.stream().map(URIStatus::getPath).collect(Collectors.toList());
-
         Assert.assertEquals(pathList.size(), 2);
 
+        for (String path : pathList) {
+            if (path.contains("tmp")) {
+                // Ensure that the temporary file is rotated and the directory is empty
+                Assert.assertEquals(path, "/pulsar/tmp");
+            } else {
+                // Ensure that all rotated files conform the naming convention
+                Assert.assertTrue(path.startsWith("/pulsar/TopicA-"));
+            }
+        }
+
+        // Ensure the subsequent writes are also successful
+        sink.write(() -> new GenericObject() {
+            @Override
+            public SchemaType getSchemaType() {
+                return SchemaType.KEY_VALUE;
+            }
+
+            @Override
+            public Object getNativeObject() {
+                return new KeyValue<>((String) fooBar.getField("address"), fooBar);
+            }
+        });
+
+        listAlluxioDirStatus = client.listStatus(alluxioURI);
+        pathList = listAlluxioDirStatus.stream().map(URIStatus::getPath).collect(Collectors.toList());
+        Assert.assertEquals(pathList.size(), 3);
+
         for (String path : pathList) {
             if (path.contains("tmp")) {
                 Assert.assertEquals(path, "/pulsar/tmp");
@@ -200,7 +232,6 @@ public Object getNativeObject() {
         }
 
         sink.close();
-        cluster.stop();
     }
 
     private LocalAlluxioCluster setupSingleMasterCluster() throws Exception {