Unable to connect to the event hub emulator using PySpark

[victorfuzaro]

Describe the bug
I am getting connection aborted/refused when trying to connect to the eventhub emulator through pyspark

To Reproduce
Steps to reproduce the behavior:

Pre-requisite is to install pyspark library version: 3.5.0 or 3.5.1 from pypi with pip install pyspark==<version_number>

A basic spark streaming application:

from pyspark.sql import SparkSession

spark = (SparkSession.builder.appName("spark-app")
         .config("spark.jars.packages",
                 "com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.22")
         .getOrCreate())

eventhub_connection_string = "Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;EntityPath=eh1"

eventhub_config = {
    "eventhubs.connectionString": spark.sparkContext._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(
        eventhub_connection_string
    )
}

df = spark.readStream.format("eventhubs").options(**eventhub_config).load()

df.writeStream.format("console").outputMode("append").start().awaitTermination()

Expected behavior
The pyspark application should start consuming data from the eventhub emulator streaming source continuously as it happens with a real Azure EventHub

Desktop:

• OS: MacOS Sonoma 14.5 w/ M2 processor

Image Platform-Architecture Used:

• Linux-arm64, docker image: latest

Docker Version:

• Docker Version: 26.1.0-rd, build cca8e72

Arguments && Environment variables to start Emulator:

• ACCEPT_EULA: "Y"
• CONFIG_PATH: "eventhubConfig.json"

Emulator Launch Method:

• Direct Docker compose, exactly as per the official documentation. I also tried mapping local ports 5671 and 5672 to container's 5671/5672 ports

Additional context:
I can send/consume events from the emulator with a python script using the standard python EventHubConsumerClient library, but this is not my use case as I do need to consume with pyspark, so I guess something may be wrong somehow with the connectivity or maybe even with the Azure Eventhub maven project that somehow may not be compatible with the event hub emulator

Log stack:

Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
24/07/03 11:10:24 WARN ResolveWriteToStream: Temporary checkpoint location created which is deleted normally when the query didn't fail: /private/var/folders/xl/1f8_mm81143b8xqvtpfqzv7w0000gn/T/temporary-9dbef554-4b26-4836-863d-a07f4b7e8228. If it's required to delete it under any circumstances, please set spark.sql.streaming.forceDeleteTempCheckpointLocation to true. Important to know deleting temp checkpoint folder is best effort.
24/07/03 11:10:24 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.
24/07/03 11:10:24 WARN ConnectionHandler: onTransportError hostname[localhost:5671], connectionId[MF_2b6422_1720001424588], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_2b6422_1720001424588], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_2b6422_1720001424588], hostname[localhost], open hasn't complete, stopping the reactor
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_2b6422_1720001424588], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_2b6422_1720001424588], hostname[localhost], closing current connection
24/07/03 11:10:24 WARN MessagingFactory: messagingFactory[MF_2b6422_1720001424588], hostName[localhost], message[stopping the reactor because thread was interrupted or the reactor has no more events to process.]
24/07/03 11:10:24 WARN ConnectionHandler: onTransportError hostname[localhost:5671], connectionId[MF_39118f_1720001424652], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_39118f_1720001424652], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_39118f_1720001424652], hostname[localhost], open hasn't complete, stopping the reactor
24/07/03 11:10:24 WARN MessagingFactory: messagingFactory[MF_39118f_1720001424652], hostName[localhost], message[stopping the reactor because thread was interrupted or the reactor has no more events to process.]
24/07/03 11:10:24 WARN ConnectionHandler: onTransportError hostname[localhost:5671], connectionId[MF_c187ce_1720001424669], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_c187ce_1720001424669], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_c187ce_1720001424669], hostname[localhost], open hasn't complete, stopping the reactor
24/07/03 11:10:24 WARN MessagingFactory: messagingFactory[MF_c187ce_1720001424669], hostName[localhost], message[stopping the reactor because thread was interrupted or the reactor has no more events to process.]
24/07/03 11:10:24 WARN ConnectionHandler: onTransportError hostname[localhost:5671], connectionId[MF_c128fa_1720001424685], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_c128fa_1720001424685], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_c128fa_1720001424685], hostname[localhost], open hasn't complete, stopping the reactor
24/07/03 11:10:24 WARN MessagingFactory: messagingFactory[MF_c128fa_1720001424685], hostName[localhost], message[stopping the reactor because thread was interrupted or the reactor has no more events to process.]
24/07/03 11:10:24 WARN ConnectionHandler: onTransportError hostname[localhost:5671], connectionId[MF_4a2afc_1720001424701], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_4a2afc_1720001424701], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_4a2afc_1720001424701], hostname[localhost], open hasn't complete, stopping the reactor
24/07/03 11:10:24 WARN MessagingFactory: messagingFactory[MF_4a2afc_1720001424701], hostName[localhost], message[stopping the reactor because thread was interrupted or the reactor has no more events to process.]
24/07/03 11:10:24 WARN ConnectionHandler: onTransportError hostname[localhost:5671], connectionId[MF_9c3b23_1720001424716], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_9c3b23_1720001424716], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_9c3b23_1720001424716], hostname[localhost], open hasn't complete, stopping the reactor
24/07/03 11:10:24 WARN MessagingFactory: messagingFactory[MF_9c3b23_1720001424716], hostName[localhost], message[stopping the reactor because thread was interrupted or the reactor has no more events to process.]
24/07/03 11:10:24 WARN ConnectionHandler: onTransportError hostname[localhost:5671], connectionId[MF_777ada_1720001424733], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_777ada_1720001424733], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_777ada_1720001424733], hostname[localhost], open hasn't complete, stopping the reactor
24/07/03 11:10:24 WARN MessagingFactory: messagingFactory[MF_777ada_1720001424733], hostName[localhost], message[stopping the reactor because thread was interrupted or the reactor has no more events to process.]
24/07/03 11:10:24 WARN ConnectionHandler: onTransportError hostname[localhost:5671], connectionId[MF_0c9fe4_1720001424749], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_0c9fe4_1720001424749], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_0c9fe4_1720001424749], hostname[localhost], open hasn't complete, stopping the reactor
24/07/03 11:10:24 WARN MessagingFactory: messagingFactory[MF_0c9fe4_1720001424749], hostName[localhost], message[stopping the reactor because thread was interrupted or the reactor has no more events to process.]
24/07/03 11:10:24 WARN ConnectionHandler: onTransportError hostname[localhost:5671], connectionId[MF_593bf7_1720001424765], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_593bf7_1720001424765], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_593bf7_1720001424765], hostname[localhost], open hasn't complete, stopping the reactor
24/07/03 11:10:24 WARN MessagingFactory: messagingFactory[MF_593bf7_1720001424765], hostName[localhost], message[stopping the reactor because thread was interrupted or the reactor has no more events to process.]
24/07/03 11:10:24 WARN ConnectionHandler: onTransportError hostname[localhost:5671], connectionId[MF_76334d_1720001424780], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_76334d_1720001424780], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_76334d_1720001424780], hostname[localhost], open hasn't complete, stopping the reactor
24/07/03 11:10:24 WARN MessagingFactory: messagingFactory[MF_76334d_1720001424780], hostName[localhost], message[stopping the reactor because thread was interrupted or the reactor has no more events to process.]
24/07/03 11:10:24 WARN ConnectionHandler: onTransportError hostname[localhost:5671], connectionId[MF_54eab3_1720001424795], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_54eab3_1720001424795], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_54eab3_1720001424795], hostname[localhost], open hasn't complete, stopping the reactor
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_54eab3_1720001424795], hostname[localhost], error[org.apache.qpid.proton.engine.TransportException: connection aborted]
24/07/03 11:10:24 WARN MessagingFactory: onConnectionError messagingFactory[MF_54eab3_1720001424795], hostname[localhost], closing current connection
24/07/03 11:10:24 WARN MessagingFactory: messagingFactory[MF_54eab3_1720001424795], hostName[localhost], message[stopping the reactor because thread was interrupted or the reactor has no more events to process.]
24/07/03 11:10:24 ERROR MicroBatchExecution: Query [id = 33c12279-1a58-465b-900b-0ba12751cab0, runId = a5385b7c-17b4-4aa6-8524-b4c0fd5eafd1] terminated with error
com.microsoft.azure.eventhubs.EventHubException: org.apache.qpid.proton.engine.TransportException: connection aborted
	at com.microsoft.azure.eventhubs.impl.ExceptionUtil.toException(ExceptionUtil.java:74)
	at com.microsoft.azure.eventhubs.impl.MessagingFactory.onConnectionError(MessagingFactory.java:525)
	at com.microsoft.azure.eventhubs.impl.ConnectionHandler.onTransportError(ConnectionHandler.java:220)
	at org.apache.qpid.proton.engine.BaseHandler.handle(BaseHandler.java:191)
	at org.apache.qpid.proton.engine.impl.EventImpl.dispatch(EventImpl.java:108)
	at org.apache.qpid.proton.reactor.impl.ReactorImpl.dispatch(ReactorImpl.java:324)
	at org.apache.qpid.proton.reactor.impl.ReactorImpl.process(ReactorImpl.java:291)
	at com.microsoft.azure.eventhubs.impl.MessagingFactory$RunReactor.run(MessagingFactory.java:784)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base/java.lang.Thread.run(Thread.java:1583)
Traceback (most recent call last):
  File "/Users/victor/PycharmProjects/development/spark-docker.py", line 20, in <module>
    df.writeStream.format("console").outputMode("append").start().awaitTermination()
  File "/Users/victor/PycharmProjects/development/.venv-3.11/lib/python3.11/site-packages/pyspark/sql/streaming/query.py", line 221, in awaitTermination
    return self._jsq.awaitTermination()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/victor/PycharmProjects/development/.venv-3.11/lib/python3.11/site-packages/py4j/java_gateway.py", line 1322, in __call__
    return_value = get_return_value(
                   ^^^^^^^^^^^^^^^^^
  File "/Users/victor/PycharmProjects/development/.venv-3.11/lib/python3.11/site-packages/pyspark/errors/exceptions/captured.py", line 185, in deco
    raise converted from None
pyspark.errors.exceptions.captured.StreamingQueryException: [STREAM_FAILED] Query [id = 33c12279-1a58-465b-900b-0ba12751cab0, runId = a5385b7c-17b4-4aa6-8524-b4c0fd5eafd1] terminated with exception: org.apache.qpid.proton.engine.TransportException: connection aborted

[Saglodha]

Hi @victorfuzaro , Event Hubs emulator only supports interaction with latest SDKs. The PySpark connector’s incompatibility with the emulator stems from its use of older SDKs.

To tackle such use cases, we’re actively working on Kafka support for our emulator however there is no clear ETA yet. Once Kafka support is available, we recommend using the more regularly updated native Kafka connectors. Kindly let us know if there are additional concerns/queries.

[Saglodha]

We just released Kafka support in Event Hubs emulator @victorfuzaro . You can find more details here

[bingqiangxiang]

@Saglodha does this mean when the spark team upgrades the SDK to the latest version, then we can use PySpark to connect the event hub emulator? We also have such scenario to use spark to handle the data from event hub.