Dataset<Row> readDf = spark
        .readStream()
        .format("kafka")
        .options(...)
        .load();

readDf.writeStream()
      .outputMode(OutputMode.Update())
      .foreachBatch(...)
      .trigger(Trigger.ProcessingTime(1000L))
      .start();

I can receive a bunch of events, one per User ID. For each user event I need to read a Delta table from blob store 1 and write it as a parquet file (some transformation involved as well) into blob store 2. Since, the data replication is independent for each user event I am using the Java multithreading via parallelStream().

userEvents.parallelStream().forEach(userEvent -> {
      Dataset<Row> readEventDf = spark
           .read()
           .format("delta")
           .load(readPath);

//perform some transformation on the above dataframe

      readEventDf.write()
        .format("parquet")
        .mode(SaveMode.Overwrite)
        .save(writePath);
});

However, when during testing I am observing a weird behaviour. I pass 2 user events, say User-1 and User-2, and I could see 2 parallel threads processing them via the Java parallelStream() method. The processing is completely successful for one thread for User-1 but failing for the other thread for User-2 at the read() step itself. I get the error as, Failed to find data source: delta. Please find packages at...

I checked the SparkSession, SparkContext and SqlContext objects active in both the threads are exactly the same. Also, the Spark Conf properties set as also exactly the same.

If I remove the parallelism by changing the method call from parallelStream() to normal sequential stream(), then everything works fine. Hence, it is evident that there is no issue with the spark/delta libraries used or connection to the blob store/file format.

Can someone help here to explain this behaviour and how to resolve it?