[fix][io] Fix Alluxio sink to respect the alluxioMasterHost property (#…

…19172)
apache · Jan 11, 2023 · dda1437 · dda1437
1 parent 82b1357
commit dda1437
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 8 deletions.
diff --git a/pulsar-io/alluxio/src/main/java/org/apache/pulsar/io/alluxio/sink/AlluxioSink.java b/pulsar-io/alluxio/src/main/java/org/apache/pulsar/io/alluxio/sink/AlluxioSink.java
@@ -92,7 +92,7 @@ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exc
         // initialize FileSystem
         String alluxioMasterHost = alluxioSinkConfig.getAlluxioMasterHost();
         int alluxioMasterPort = alluxioSinkConfig.getAlluxioMasterPort();
-        InstancedConfiguration.defaults().set(PropertyKey.MASTER_HOSTNAME, alluxioMasterHost);
+        configuration.set(PropertyKey.MASTER_HOSTNAME, alluxioMasterHost);
         configuration.set(PropertyKey.MASTER_RPC_PORT, alluxioMasterPort);
         if (alluxioSinkConfig.getSecurityLoginUser() != null) {
             configuration.set(PropertyKey.SECURITY_LOGIN_USERNAME, alluxioSinkConfig.getSecurityLoginUser());

diff --git a/pulsar-io/alluxio/src/test/java/org/apache/pulsar/io/alluxio/sink/AlluxioSinkTest.java b/pulsar-io/alluxio/src/test/java/org/apache/pulsar/io/alluxio/sink/AlluxioSinkTest.java
@@ -64,6 +64,7 @@ public class AlluxioSinkTest {
 
     protected Map<String, Object> map;
     protected AlluxioSink sink;
+    protected LocalAlluxioCluster cluster;
 
     @Mock
     protected Record<GenericObject> mockRecord;
@@ -86,10 +87,16 @@ public static void init() {
     }
 
     @BeforeMethod
-    public final void setUp() {
+    public final void setUp() throws Exception {
+        cluster = setupSingleMasterCluster();
+
         map = new HashMap<>();
-        map.put("alluxioMasterHost", "localhost");
-        map.put("alluxioMasterPort", "19998");
+        // alluxioMasterHost should be set via LocalAlluxioCluster#getHostname
+        // instead of using a fixed value "localhost", since it seems that
+        // LocalAlluxioCluster may bind other address than localhost
+        // when the node has multiple network interfaces.
+        map.put("alluxioMasterHost", cluster.getHostname());
+        map.put("alluxioMasterPort", cluster.getMasterRpcPort());
         map.put("alluxioDir", "/pulsar");
         map.put("filePrefix", "prefix");
         map.put("schemaEnable", "true");
@@ -127,8 +134,6 @@ public void openTest() throws Exception {
 
         String alluxioDir = "/pulsar";
 
-        LocalAlluxioCluster cluster = setupSingleMasterCluster();
-
         sink = new AlluxioSink();
         sink.open(map, mockSinkContext);
 
@@ -156,8 +161,6 @@ public void writeAndCloseTest() throws Exception {
 
         String alluxioDir = "/pulsar";
 
-        LocalAlluxioCluster cluster = setupSingleMasterCluster();
-
         sink = new AlluxioSink();
         sink.open(map, mockSinkContext);