changes

src/csharp/Microsoft.Spark/Interop/Ipc/IJvmBridgeFactory.cs

@@ -2,10 +2,14 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+using System.Net;
+
 namespace Microsoft.Spark.Interop.Ipc
 {
     internal interface IJvmBridgeFactory
     {
         IJvmBridge Create(int portNumber);
+
+        IJvmBridge Create(IPAddress ip, int portNumber);
     }
 }

src/csharp/Microsoft.Spark/Interop/Ipc/JvmBridge.cs

@@ -41,17 +41,22 @@ internal sealed class JvmBridge : IJvmBridge
             new ConcurrentQueue<ISocketWrapper>();
         private readonly ILoggerService _logger =
             LoggerServiceFactory.GetLogger(typeof(JvmBridge));
+        private readonly IPAddress _ipAddress;
         private readonly int _portNumber;
         private readonly JvmThreadPoolGC _jvmThreadPoolGC;
         private readonly bool _isRunningRepl;
 
-        internal JvmBridge(int portNumber)
+        internal JvmBridge(int portNumber): this(IPAddress.Loopback, portNumber)
+        {
+        }
+
+        internal JvmBridge(IPAddress ipAddress, int portNumber)
         {
             if (portNumber == 0)
             {
                 throw new Exception("Port number is not set.");
             }
-
+            _ipAddress = ipAddress;
             _portNumber = portNumber;
             _logger.LogInfo($"JvMBridge port is {portNumber}");
 
@@ -85,7 +90,7 @@ private ISocketWrapper GetConnection()
             {
                 IPEndPoint dotnetBackendIPEndpoint = SparkEnvironment.ConfigurationService.GetBackendIPEndpoint();
                 socket = SocketFactory.CreateSocket();
-                socket.Connect(dotnetBackendIPEndpoint.Address, dotnetBackendIPEndpoint.Port);
+                socket.Connect(_ipAddress, _portNumber);
             }
 
             return socket;

src/csharp/Microsoft.Spark/Interop/SparkEnvironment.cs

@@ -3,6 +3,7 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Net;
 using Microsoft.Spark.Interop.Ipc;
 using Microsoft.Spark.Services;
 
@@ -70,8 +71,9 @@ public static IJvmBridge JvmBridge
         {
             get
             {
+                IPEndPoint jvmBackendEndPoint = ConfigurationService.GetBackendIPEndpoint();
                 return s_jvmBridge ??=
-                    JvmBridgeFactory.Create(ConfigurationService.GetBackendPortNumber());
+                    JvmBridgeFactory.Create(jvmBackendEndPoint.Address, jvmBackendEndPoint.Port);
             }
             set
             {

src/csharp/Microsoft.Spark/Network/DefaultSocketWrapper.cs

@@ -27,10 +27,10 @@ internal sealed class DefaultSocketWrapper : ISocketWrapper
         /// 
         /// This socket is bound to provided IP address with port 0.
         /// </summary>
-        public DefaultSocketWrapper() :
+        public DefaultSocketWrapper(IPAddress ipAddress) :
             this(new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
         {
-            _innerSocket.Bind(new IPEndPoint(IPAddress.Any, 0));
+            _innerSocket.Bind(new IPEndPoint(ipAddress, 0));
         }
 
         /// <summary>

src/csharp/Microsoft.Spark/Network/SocketFactory.cs

@@ -2,6 +2,8 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+using System.Net;
+
 namespace Microsoft.Spark.Network
 {
     /// <summary>
@@ -19,5 +21,10 @@ public static ISocketWrapper CreateSocket()
         {
             return new DefaultSocketWrapper();
         }
+
+        public static ISocketWrapper CreateSocket(IPAddress ip)
+        {
+            return new DefaultSocketWrapper(ip);
+        }
     }
 }

src/csharp/Microsoft.Spark/Services/ConfigurationService.cs

@@ -26,6 +26,7 @@ internal sealed class ConfigurationService : IConfigurationService
 
         private const string DotnetBackendPortEnvVarName = "DOTNETBACKEND_PORT";
         private const string DotnetBackendIPAddressEnvVarName = "DOTNET_SPARK_BACKEND_IP_ADDRESS";
+        private const string DotnetCallbackServerIPAddressEnvVarName = "DOTNET_SPARK_CALLBACK_SERVER_IP_ADDRESS";
         private const int DotnetBackendDebugPort = 5567;
 
         private const string DotnetNumBackendThreadsEnvVarName = "DOTNET_SPARK_NUM_BACKEND_THREADS";
@@ -123,24 +124,6 @@ public IPEndPoint GetBackendIPEndpoint()
             return new IPEndPoint(IPAddress.Parse(ipAddress), portNumber);
         }
 
-        /// <summary>
-        /// Returns the port number for socket communication between JVM and CLR.
-        /// </summary>
-        public int GetBackendPortNumber()
-        {
-            if (!int.TryParse(
-                GetEnvironmentVariable(DotnetBackendPortEnvVarName),
-                out int portNumber))
-            {
-                _logger.LogInfo($"'{DotnetBackendPortEnvVarName}' environment variable is not set.");
-                portNumber = DotnetBackendDebugPort;
-            }
-
-            _logger.LogInfo($"Using port {portNumber} for connection.");
-
-            return portNumber;
-        }
-
         /// <summary>
         /// Returns the max number of threads for socket communication between JVM and CLR.
         /// </summary>
@@ -157,14 +140,19 @@ public int GetNumBackendThreads()
         }
 
         /// <summary>
-        /// Returns the IP address for socket communication between JVM and CLR.
+        /// Returns the IP address for socket communication between JVM and CallBack Server.
         /// </summary>
-        public string GetBackendIPAddress()
+        public IPAddress GetCallbackServerIPAddress()
         {
-            string ipAddress = Environment.GetEnvironmentVariable(DotnetBackendIPAddressEnvVarName);
-            _logger.LogInfo($"Using IP address {ipAddress} for connection.");
+            string ipAddress = Environment.GetEnvironmentVariable(DotnetCallbackServerIPAddressEnvVarName);
+            if (ipAddress == null)
+            {
+                _logger.LogInfo($"'{DotnetCallbackServerIPAddressEnvVarName}' environment variable is not set.");
+                ipAddress = "127.0.0.1";
+            }
+            _logger.LogInfo($"Using IP address {ipAddress} for connection with Callback Server.");
 
-            return ipAddress;
+            return IPAddress.Parse(ipAddress);
         }
 
         /// <summary>

src/csharp/Microsoft.Spark/Services/IConfigurationService.cs

@@ -17,11 +17,6 @@ internal interface IConfigurationService
         /// </summary>
         TimeSpan JvmThreadGCInterval { get; }
 
-        /// <summary>
-        /// The port number used for communicating with the .NET backend process.
-        /// </summary>
-        int GetBackendPortNumber();
-
         /// <summary>
         /// Returns the max number of threads for socket communication between JVM and CLR.
         /// </summary>
@@ -33,9 +28,9 @@ internal interface IConfigurationService
         IPEndPoint GetBackendIPEndpoint();
 
         /// <summary>
-        /// The IP address used for communicating with the .NET backend process.
+        /// The IP address used for communicating with CallBack server.
         /// </summary>
-        string GetBackendIPAddress();
+        IPAddress GetCallbackServerIPAddress();
 
         /// <summary>
         /// The full path to the .NET worker executable.

src/csharp/Microsoft.Spark/Sql/DataFrame.cs

@@ -849,7 +849,8 @@ public IEnumerable<Row> ToLocalIterator(bool prefetchPartitions)
                     Reference.Invoke("toPythonIterator", prefetchPartitions),
                     true);
             using ISocketWrapper socket = SocketFactory.CreateSocket();
-            socket.Connect(IPAddress.Loopback, port, secret);
+            IPEndPoint dotnetBackendIPEndpoint = SparkEnvironment.ConfigurationService.GetBackendIPEndpoint();
+            socket.Connect(dotnetBackendIPEndpoint.Address, port, secret);
             foreach (Row row in new RowCollector().Collect(socket, server))
             {
                 yield return row;

src/scala/microsoft-spark-2-4/src/main/scala/org/apache/spark/deploy/dotnet/DotnetRunner.scala

@@ -64,7 +64,7 @@ object DotnetRunner extends Logging {
     // In debug mode this runner will not launch a .NET process.
     val runInDebugMode = settings._1
     @volatile var dotnetBackendPortNumber = settings._2
-    val dotnetBackendIPAddress = sys.env.getOrElse("DOTNET_SPARK_BACKEND_IP_ADDRESS", "127.0.0.1").toInt
+    val dotnetBackendIPAddress = sys.env.getOrElse("DOTNET_SPARK_BACKEND_IP_ADDRESS", "127.0.0.1")
     var dotnetExecutable = ""
     var otherArgs: Array[String] = null
 
@@ -112,8 +112,9 @@ object DotnetRunner extends Logging {
       override def run() {
         // need to get back dotnetBackendPortNumber because if the value passed to init is 0
         // the port number is dynamically assigned in the backend
-        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendPortNumber)
-        logInfo(s"Port number used by DotnetBackend is $dotnetBackendPortNumber")
+        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendIPAddress, dotnetBackendPortNumber)
+        logInfo(s"IP address used by DotnetBackend is $dotnetBackendIPAddress and " +
+          s"Port number used is $dotnetBackendPortNumber")
         initialized.release()
         dotnetBackend.run()
       }

src/scala/microsoft-spark-3-0/src/main/scala/org/apache/spark/api/dotnet/DotnetBackend.scala

@@ -34,7 +34,7 @@ class DotnetBackend extends Logging {
   @volatile
   private[dotnet] var callbackClient: Option[CallbackClient] = None
 
-  def init(portNumber: Int): Int = {
+  def init(ipAddress: String, portNumber: Int): Int = {
     val conf = Option(SparkEnv.get).map(_.conf).getOrElse(new SparkConf())
     val numBackendThreads = conf.get(DOTNET_NUM_BACKEND_THREADS)
     logInfo(s"The number of DotnetBackend threads is set to $numBackendThreads.")

src/scala/microsoft-spark-3-0/src/main/scala/org/apache/spark/deploy/dotnet/DotnetRunner.scala

@@ -63,6 +63,7 @@ object DotnetRunner extends Logging {
     // In debug mode this runner will not launch a .NET process.
     val runInDebugMode = settings._1
     @volatile var dotnetBackendPortNumber = settings._2
+    val dotnetBackendIPAddress = sys.env.getOrElse("DOTNET_SPARK_BACKEND_IP_ADDRESS", "127.0.0.1")
     var dotnetExecutable = ""
     var otherArgs: Array[String] = null
 
@@ -101,7 +102,6 @@ object DotnetRunner extends Logging {
 
     // Time to wait for DotnetBackend to initialize in seconds.
     val backendTimeout = sys.env.getOrElse("DOTNETBACKEND_TIMEOUT", "120").toInt
-    val dotnetBackendIPAddress = sys.env.getOrElse("DOTNET_SPARK_BACKEND_IP_ADDRESS", "0.0.0.0")
 
     // Launch a DotnetBackend server for the .NET process to connect to; this will let it see our
     // Java system properties etc.
@@ -112,8 +112,8 @@ object DotnetRunner extends Logging {
         // need to get back dotnetBackendPortNumber because if the value passed to init is 0
         // the port number is dynamically assigned in the backend
         dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendIPAddress, dotnetBackendPortNumber)
-        logInfo(s"Port number used by DotnetBackend is $dotnetBackendPortNumber on IP address " +
-          s"$dotnetBackendIPAddress")
+        logInfo(s"IP address used by DotnetBackend is $dotnetBackendIPAddress and " +
+          s"Port number used is $dotnetBackendPortNumber")
         initialized.release()
         dotnetBackend.run()
       }

src/scala/microsoft-spark-3-1/src/main/scala/org/apache/spark/api/dotnet/DotnetBackend.scala

@@ -34,7 +34,7 @@ class DotnetBackend extends Logging {
   @volatile
   private[dotnet] var callbackClient: Option[CallbackClient] = None
 
-  def init(portNumber: Int): Int = {
+  def init(ipAddress: String, portNumber: Int): Int = {
     val conf = Option(SparkEnv.get).map(_.conf).getOrElse(new SparkConf())
     val numBackendThreads = conf.get(DOTNET_NUM_BACKEND_THREADS)
     logInfo(s"The number of DotnetBackend threads is set to $numBackendThreads.")

src/scala/microsoft-spark-3-1/src/main/scala/org/apache/spark/deploy/dotnet/DotnetRunner.scala

@@ -64,7 +64,7 @@ object DotnetRunner extends Logging {
     // In debug mode this runner will not launch a .NET process.
     val runInDebugMode = settings._1
     @volatile var dotnetBackendPortNumber = settings._2
-    val dotnetBackendIPAddress = settings._3
+    val dotnetBackendIPAddress = sys.env.getOrElse("DOTNET_SPARK_BACKEND_IP_ADDRESS", "127.0.0.1")
     var dotnetExecutable = ""
     var otherArgs: Array[String] = null
 
@@ -112,9 +112,9 @@ object DotnetRunner extends Logging {
       override def run() {
         // need to get back dotnetBackendPortNumber because if the value passed to init is 0
         // the port number is dynamically assigned in the backend
-        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendPortNumber, dotnetBackendIPAddress)
-        logInfo(s"Port number used by DotnetBackend is $dotnetBackendPortNumber on IP address " +
-          s"$dotnetBackendIPAddress")
+        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendIPAddress, dotnetBackendPortNumber)
+        logInfo(s"IP address used by DotnetBackend is $dotnetBackendIPAddress and " +
+          s"Port number used is $dotnetBackendPortNumber")
         initialized.release()
         dotnetBackend.run()
       }

src/scala/microsoft-spark-3-2/src/main/scala/org/apache/spark/api/dotnet/DotnetBackend.scala

@@ -34,7 +34,7 @@ class DotnetBackend extends Logging {
   @volatile
   private[dotnet] var callbackClient: Option[CallbackClient] = None
 
-  def init(portNumber: Int): Int = {
+  def init(ipAddress: String, portNumber: Int): Int = {
     val conf = Option(SparkEnv.get).map(_.conf).getOrElse(new SparkConf())
     val numBackendThreads = conf.get(DOTNET_NUM_BACKEND_THREADS)
     logInfo(s"The number of DotnetBackend threads is set to $numBackendThreads.")

src/scala/microsoft-spark-3-2/src/main/scala/org/apache/spark/deploy/dotnet/DotnetRunner.scala

@@ -64,7 +64,7 @@ object DotnetRunner extends Logging {
     // In debug mode this runner will not launch a .NET process.
     val runInDebugMode = settings._1
     @volatile var dotnetBackendPortNumber = settings._2
-    val dotnetBackendIPAddress = settings._3
+    val dotnetBackendIPAddress = sys.env.getOrElse("DOTNET_SPARK_BACKEND_IP_ADDRESS", "127.0.0.1")
     var dotnetExecutable = ""
     var otherArgs: Array[String] = null
 
@@ -112,9 +112,9 @@ object DotnetRunner extends Logging {
       override def run() {
         // need to get back dotnetBackendPortNumber because if the value passed to init is 0
         // the port number is dynamically assigned in the backend
-        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendPortNumber, dotnetBackendIPAddress)
-        logInfo(s"Port number used by DotnetBackend is $dotnetBackendPortNumber on IP address " +
-          s"$dotnetBackendIPAddress")
+        dotnetBackendPortNumber = dotnetBackend.init(dotnetBackendIPAddress, dotnetBackendPortNumber)
+        logInfo(s"IP address used by DotnetBackend is $dotnetBackendIPAddress and " +
+          s"Port number used is $dotnetBackendPortNumber")
         initialized.release()
         dotnetBackend.run()
       }