Page MenuHomePhabricator

ERROR AsyncEventQueue: Listener DatahubSparkListener threw an exception
Open, HighPublic

Description

As per

sudo -u analytics kerberos-run-command analytics yarn logs -applicationId application_1750705250302_855817 > pageview_actor_hourly__compute_pageview_actor_hourly__20250720.application_1750705250302_855817

We seem to be encountering an error with emission of lineage:

25/07/21 17:26:21 ERROR AsyncEventQueue: Listener DatahubSparkListener threw an exception
datahub.shaded.org.apache.kafka.common.KafkaException: Failed to construct kafka producer
        at datahub.shaded.org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:430)
        at datahub.shaded.org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:298)
        at datahub.client.kafka.KafkaEmitter.<init>(KafkaEmitter.java:55)
        at datahub.spark.DatahubEventEmitter.getEmitter(DatahubEventEmitter.java:85)
        at datahub.spark.DatahubEventEmitter.emitMcps(DatahubEventEmitter.java:401)
        at datahub.spark.DatahubEventEmitter.emitCoalesced(DatahubEventEmitter.java:190)
        at datahub.spark.DatahubSparkListener.onApplicationEnd(DatahubSparkListener.java:279)
        at org.apache.spark.scheduler.SparkListenerBus.doPostEvent(SparkListenerBus.scala:57)
        at org.apache.spark.scheduler.SparkListenerBus.doPostEvent$(SparkListenerBus.scala:28)
        at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37)
        at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37)
        at org.apache.spark.util.ListenerBus.postToAll(ListenerBus.scala:117)
        at org.apache.spark.util.ListenerBus.postToAll$(ListenerBus.scala:101)
        at org.apache.spark.scheduler.AsyncEventQueue.super$postToAll(AsyncEventQueue.scala:105)
        at org.apache.spark.scheduler.AsyncEventQueue.$anonfun$dispatch$1(AsyncEventQueue.scala:105)
        at scala.runtime.java8.JFunction0$mcJ$sp.apply(JFunction0$mcJ$sp.java:23)
        at scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)
        at org.apache.spark.scheduler.AsyncEventQueue.org$apache$spark$scheduler$AsyncEventQueue$$dispatch(AsyncEventQueue.scala:100)
        at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.$anonfun$run$1(AsyncEventQueue.scala:96)
        at org.apache.spark.util.Utils$.tryOrStopSparkContext(Utils.scala:1381)
        at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.run(AsyncEventQueue.scala:96)
Caused by: datahub.shaded.org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers
        at datahub.shaded.org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:84)
        at datahub.shaded.org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:408)
        ... 20 more

Something similar is mentioned in T386862#10700176 , although filing this as a standalone item (please feel free to mark as duplicate if it makes sense to handle that on the other task, of course).

I'm unclear if this was a transient error due to the dataset production failing or if it was simply coincidental.

The Airflow job's execution that spawned the job was:

spark3-submit --driver-cores 2 --conf spark.executorEnv.SPARK_HOME=/usr/lib/spark3
    --conf spark.executorEnv.SPARK_CONF_DIR=/etc/spark3/conf --master yarn --conf
    spark.datahub.emitter=kafka --conf spark.extraListeners=datahub.spark.DatahubSparkListener
    --conf spark.datahub.kafka.bootstrap=kafka-jumbo-eqiad.external-services.svc.cluster.local:9092
    --conf spark.datahub.kafka.schema_registry_url=https://datahub-gms.discovery.wmnet:30443/schema-registry/api/
    --conf spark.datahub.flow_name=airflow_pageview_actor_hourly__compute_pageview_actor_hourly
    --conf spark.datahub.log.mcps=false --conf spark.dynamicAllocation.maxExecutors=128
    --conf spark.executor.memoryOverhead=4G --conf spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2
    --conf spark.yarn.appMasterEnv.SPARK_CONF_DIR=/etc/spark3/conf --conf spark.yarn.appMasterEnv.SPARK_HOME=/usr/lib/spark3
    --jars hdfs:///wmf/cache/artifacts/airflow/analytics/acryl-spark-lineage-0.2.16.jar
    --executor-cores 2 --executor-memory 8G --driver-memory 4G --keytab airflow.keytab
    --principal analytics/airflow.discovery.wmnet --name pageview_actor_hourly__compute_pageview_actor_hourly__20250720
    --class org.apache.spark.sql.hive.thriftserver.WMFSparkSQLCLIDriver --queue production
    --deploy-mode client hdfs:///wmf/cache/artifacts/airflow/analytics/wmf-sparksqlclidriver-1.0.0.jar
    -f hdfs://analytics-hadoop/wmf/refinery/current/hql/pageview/actor/pageview_actor.hql
    -d refinery_hive_jar_path=hdfs:///wmf/cache/artifacts/airflow/analytics/refinery-hive-0.2.54-shaded.jar
    -d source_table=wmf.webrequest -d actor_label_table=wmf.webrequest_actor_label_hourly
    -d destination_table=wmf.pageview_actor -d coalesce_partitions=32 -d year=2025
    -d month=7 -d day=20 -d hour=3

(By the way, the command was re-run manually, as discussed on Slack.)

Event Timeline

Just a hunch, but this may also pertain to DataHub sometimes being down. There are data-platform-alerts intermittently indicating DatahubNextServiceUnavailable .

Milimetric added subscribers: BTullis, Milimetric.

cc @BTullis Andrew says might be a problem with k8s networking

Gehel triaged this task as High priority.Sep 2 2025, 1:42 PM

Hmm. This is interesting.

Caused by: datahub.shaded.org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers

The command contains the following cofig entry...

--conf spark.datahub.kafka.bootstrap=kafka-jumbo-eqiad.external-services.svc.cluster.local:9092

That address should resolve to all of the IPv4 and IPv6 address of the kafka-jumbo brokers, like this...

btullis@deploy1003:~$ kube-env airflow-main-deploy dse-k8s-eqiad

btullis@deploy1003:~$ kubectl exec -it airflow-task-shell-6d5bdb94fc-c5tjd -- bash

runuser@airflow-task-shell-6d5bdb94fc-c5tjd:/opt/airflow$ host kafka-jumbo-eqiad.external-services.svc.cluster.local
kafka-jumbo-eqiad.external-services.svc.cluster.local has address 10.64.132.21
kafka-jumbo-eqiad.external-services.svc.cluster.local has address 10.64.135.16
kafka-jumbo-eqiad.external-services.svc.cluster.local has address 10.64.131.16
kafka-jumbo-eqiad.external-services.svc.cluster.local has address 10.64.160.16
kafka-jumbo-eqiad.external-services.svc.cluster.local has address 10.64.134.9
kafka-jumbo-eqiad.external-services.svc.cluster.local has address 10.64.154.15
kafka-jumbo-eqiad.external-services.svc.cluster.local has address 10.64.130.10
kafka-jumbo-eqiad.external-services.svc.cluster.local has address 10.64.0.126
kafka-jumbo-eqiad.external-services.svc.cluster.local has address 10.64.136.11
kafka-jumbo-eqiad.external-services.svc.cluster.local has IPv6 address 2620:0:861:10e:10:64:135:16
kafka-jumbo-eqiad.external-services.svc.cluster.local has IPv6 address 2620:0:861:10f:10:64:136:11
kafka-jumbo-eqiad.external-services.svc.cluster.local has IPv6 address 2620:0:861:109:10:64:130:10
kafka-jumbo-eqiad.external-services.svc.cluster.local has IPv6 address 2620:0:861:101:10:64:0:126
kafka-jumbo-eqiad.external-services.svc.cluster.local has IPv6 address 2620:0:861:128:10:64:160:16
kafka-jumbo-eqiad.external-services.svc.cluster.local has IPv6 address 2620:0:861:10b:10:64:132:21
kafka-jumbo-eqiad.external-services.svc.cluster.local has IPv6 address 2620:0:861:10d:10:64:134:9
kafka-jumbo-eqiad.external-services.svc.cluster.local has IPv6 address 2620:0:861:10a:10:64:131:16
kafka-jumbo-eqiad.external-services.svc.cluster.local has IPv6 address 2620:0:861:122:10:64:154:15

Is this a transient error, or does it happen every time?