Sending RDD in pySpark to a Scala function calling in python side.

my_python_kafka_producer.py

from pyspark.serializers import PickleSerializer, AutoBatchedSerializer


reserialized_rdd = rdd._reserialize(AutoBatchedSerializer(PickleSerializer()))
rdd_java = rdd.ctx._jvm.SerDe.pythonToJava(rdd._jrdd, True)

_jvm = sc._jvm
_jvm.myclass.apps.mine\
                 .pyKafka\
                 .sendMessageByRDD("host:6667", 
                                   "SASL_PLAINTEXT", 
                                   rdd_java, 
                                   'my_topic')
                                     

pyKafka.scala

import org.apache.spark.api.java.JavaRDD

def sendMessageByRDD(broker :String, protocol :String, rdd :JavaRDD[Any], topic :String) = {
    val props = new Properties() 
    
    props.put("security.protocol", protocol)
    props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer")
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("bootstrap.servers", broker)
    ???
}