search

Azure / azure-event-hubs-emulator-installer

This repository acts as a central hub for handling issues, collecting feedback, and facilitating discussions related to the Azure Event Hubs emulator.
https://learn.microsoft.com/azure/event-hubs/overview-emulator
MIT License
24 stars 8 forks source link

Unable to connect to the event hub emulator using PySpark #24

Open victorfuzaro opened 2 months ago

victorfuzaro commented 2 months ago

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==

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:

Image Platform-Architecture Used:

Docker Version:

Arguments && Environment variables to start Emulator:

Emulator Launch Method:

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 commented 1 month ago

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.