Fix for using Broadcast variables in Databricks

src/csharp/Microsoft.Spark.Worker/DaemonWorker.cs

@@ -128,7 +128,7 @@ private void StartServer(ISocketWrapper listener)
                         SerDe.Write(socket.OutputStream, taskRunnerId);
                         socket.OutputStream.Flush();
 
-                        if (Utils.SettingUtils.IsDatabricks)
+                        if (ConfigurationService.IsDatabricks)
                         {
                             SerDe.ReadString(socket.InputStream);
                         }

src/csharp/Microsoft.Spark.Worker/Processor/PayloadProcessor.cs

@@ -7,6 +7,7 @@
 using System.Collections.Generic;
 using System.IO;
 using Microsoft.Spark.Interop.Ipc;
+using Microsoft.Spark.Services;
 using Microsoft.Spark.Worker.Utils;
 
 namespace Microsoft.Spark.Worker.Processor
@@ -68,7 +69,7 @@ internal Payload Process(Stream stream)
             // our current context.
             AssemblyLoaderHelper.RegisterAssemblyHandler();
 
-            if (SettingUtils.IsDatabricks)
+            if (ConfigurationService.IsDatabricks)
             {
                 SerDe.ReadString(stream);
                 SerDe.ReadString(stream);

src/csharp/Microsoft.Spark.Worker/Utils/SettingUtils.cs

@@ -12,9 +12,6 @@ namespace Microsoft.Spark.Worker.Utils
     /// </summary>
     internal static class SettingUtils
     {
-        internal static bool IsDatabricks { get; } =
-            !string.IsNullOrEmpty(GetEnvironmentVariable("DATABRICKS_RUNTIME_VERSION"));
-
         internal static string GetWorkerFactorySecret(Version version)
         {
             return (version >= new Version(Versions.V2_3_1)) ?

src/csharp/Microsoft.Spark/Broadcast.cs

@@ -171,11 +171,24 @@ private JvmObjectReference CreateBroadcast_V2_3_2_AndAbove(
             // Spark versions.
             bool encryptionEnabled = bool.Parse(
                 sc.GetConf().Get("spark.io.encryption.enabled", "false"));
+            JvmObjectReference _pythonBroadcast;
 
-            var _pythonBroadcast = (JvmObjectReference)javaSparkContext.Jvm.CallStaticJavaMethod(
-                "org.apache.spark.api.python.PythonRDD",
-                "setupBroadcast",
-                _path);
+            // Databricks has changed the signature of setupBroadcast in its Spark
+            if (ConfigurationService.IsDatabricks)
+            {
+                _pythonBroadcast = (JvmObjectReference)javaSparkContext.Jvm.CallStaticJavaMethod(
+                    "org.apache.spark.api.python.PythonRDD",
+                    "setupBroadcast",
+                    javaSparkContext,
+                    _path);
+            }
+            else
+            {
+                _pythonBroadcast = (JvmObjectReference)javaSparkContext.Jvm.CallStaticJavaMethod(
+                    "org.apache.spark.api.python.PythonRDD",
+                    "setupBroadcast",
+                    _path);
+            }
 
             if (encryptionEnabled)
             {

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

@@ -5,6 +5,7 @@
 using System;
 using System.IO;
 using System.Runtime.InteropServices;
+using static System.Environment;
 using Microsoft.Spark.Utils;
 
 namespace Microsoft.Spark.Services
@@ -49,6 +50,9 @@ public TimeSpan JvmThreadGCInterval
             }
         }
 
+        internal static bool IsDatabricks { get; } =
+            !string.IsNullOrEmpty(GetEnvironmentVariable("DATABRICKS_RUNTIME_VERSION"));
+
         /// <summary>
         /// Returns the port number for socket communication between JVM and CLR.
         /// </summary>