# Ray streaming with Kafka **URL:** \ **Category:** Kubernetes\ **Created:** [August 19, 2021, 12:47am UTC](https://discuss.ray.io/t/ray-streaming-with-kafka/3284 "2021-08-19T00:47:00Z")\ **Posts on this page:** 4\ **Page:** 1
**Author:** ![blublinsky](https://avatars.discourse-cdn.com/v4/letter/b/f05b48/32.png) [@blublinsky](https://discuss.ray.io/u/blublinsky)\ **Post date:** [August 19, 2021, 12:47am UTC](https://discuss.ray.io/t/ray-streaming-with-kafka/3284/1 "2021-08-19T00:47:00Z")
There is this excellent blog post [Anyscale - Serverless Kafka Stream Processing with Ray](https://www.anyscale.com/blog/serverless-kafka-stream-processing-with-ray) on running Kafka with Ray, which at the end references [ray/streaming at master · ray-project/ray · GitHub](https://github.com/ray-project/ray/tree/master/streaming) library. Are there any examples on integration this library with kafka? The issue is that you can’t run StreamingContext.Builder().build() inside kafka receiver actor, and if I run it outside, then its not clear, how to specify ```auto ctx.source(getmessages()) ``` for the case when I have many listener’s actors. Do I create a context per listener? Can I merge them together at the end? ---
**Author:** ![blublinsky](https://avatars.discourse-cdn.com/v4/letter/b/f05b48/32.png) [@blublinsky](https://discuss.ray.io/u/blublinsky)\ **Post date:** [August 19, 2021, 1:13pm UTC](https://discuss.ray.io/t/ray-streaming-with-kafka/3284/2 "2021-08-19T13:13:03Z")
Did a bit more digging and seeing that `StreamingContext.Builder().build()` fails because of the error: ```auto ValueError: Cross language feature needs --load-code-from-local to be set. ``` Quick search suggest setting ```auto ray.init(_load_code_from_local=True) ``` But my version ```auto __commit__ = "ee291ae762325f8e70bc3fbd5652b563744d22b2" __version__ = "2.0.0.dev0" ``` Does not seem to support it ---
**Author:** ![blublinsky](https://avatars.discourse-cdn.com/v4/letter/b/f05b48/32.png) [@blublinsky](https://discuss.ray.io/u/blublinsky)\ **Post date:** [August 29, 2021, 12:46am UTC](https://discuss.ray.io/t/ray-streaming-with-kafka/3284/3 "2021-08-29T00:46:25Z")
I finally found a solution here - [ray/wordcount.py at master · ray-project/ray · GitHub](https://github.com/ray-project/ray/blob/master/streaming/python/examples/wordcount.py). It fixed ```auto ValueError: Cross language feature needs --load-code-from-local to be set. ``` Unfortunately the sample itself does not work. The error I am getting when running it is: ```auto 2021-08-28 19:40:54,055 INFO services.py:1265 -- View the Ray dashboard at http://127.0.0.1:8265 (raylet) WARNING: An illegal reflective access operation has occurred (raylet) WARNING: Illegal reflective access by org.nustaq.serialization.FSTClazzInfo (file:/usr/local/lib/python3.7/site-packages/ray/jars/ray_dist.jar) to field java.lang.String.value (raylet) WARNING: Please consider reporting this to the maintainers of org.nustaq.serialization.FSTClazzInfo (raylet) WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations (raylet) WARNING: All illegal access operations will be denied in a future release Traceback (most recent call last): File "/Users/boris/Projects/scalingpythonml/streaming/ray_streaming/wordcount.py ", line 73, in .option(Config.CHANNEL_TYPE, Config.NATIVE_CHANNEL) \ File "/usr/local/lib/python3.7/site-packages/ray/streaming/context.py", line 48, in build ctx = StreamingContext() File "/usr/local/lib/python3.7/site-packages/ray/streaming/context.py", line 54, in __init__ self._j_ctx = self._gateway_client.create_streaming_context() File "/usr/local/lib/python3.7/site-packages/ray/streaming/runtime/gateway_client.py", line 21, in create_streaming_context return deserialize(ray.get(call)) File "/usr/local/lib/python3.7/site-packages/ray/_private/client_mode_hook.py", line 89, in wrapper return func(*args, **kwargs) File "/usr/local/lib/python3.7/site-packages/ray/worker.py", line 1631, in get raise value ray.exceptions.CrossLanguageError: An exception raised from JAVA: io.ray.runtime.exception.RayActorException: (pid=5454, ip=192.168.0.2) The actor died because of it's creation task failed at io.ray.runtime.task.TaskExecutor.execute(TaskExecutor.java:216) at io.ray.runtime.RayNativeRuntime.nativeRunTaskExecutor(Native Method) at io.ray.runtime.RayNativeRuntime.run(RayNativeRuntime.java:225) at io.ray.runtime.runner.worker.DefaultWorker.main(DefaultWorker.java:15) Caused by: java.lang.RuntimeException: Failed to load functions from class io.ray.streaming.runtime.python.PythonGateway at io.ray.runtime.functionmanager.FunctionManager$JobFunctionTable.loadFunctionsForClass(FunctionManager.java:240) at io.ray.runtime.functionmanager.FunctionManager$JobFunctionTable.getFunction(FunctionManager.java:173) at io.ray.runtime.functionmanager.FunctionManager.getFunction(FunctionManager.java:108) at io.ray.runtime.task.TaskExecutor.getRayFunction(TaskExecutor.java:69) at io.ray.runtime.task.TaskExecutor.execute(TaskExecutor.java:121) ... 3 more Caused by: java.lang.ClassNotFoundException: io.ray.streaming.runtime.python.PythonGateway at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:435) at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:589) at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:522) at java.base/java.lang.Class.forName0(Native Method) at java.base/java.lang.Class.forName(Class.java:468) at io.ray.runtime.functionmanager.FunctionManager$JobFunctionTable.loadFunctionsForClass(FunctionManager.java:199) ... 7 more ``` ---
**Author:** ![blublinsky](https://avatars.discourse-cdn.com/v4/letter/b/f05b48/32.png) [@blublinsky](https://discuss.ray.io/u/blublinsky)\ **Post date:** [August 29, 2021, 5:39pm UTC](https://discuss.ray.io/t/ray-streaming-with-kafka/3284/4 "2021-08-29T17:39:03Z")
More investigations here. It looks like usage of ```auto ray.init(job_config=ray.job_config.JobConfig(code_search_path=sys.path)) ``` breaks any application that is using actors. Is it a bug?