title: Spark Cloudfiles Autoloader BlobStorage Azure java.io.IOException: Attempted read from closed stream
body: <p>I learn to use the new autoloader streaming method on SPARK 3 and I have this issue.
  Here i'm trying to listen simple json files but my stream never start.</p>
  <p>My code (creds removed) :</p>
  <pre class="lang-py prettyprint-override"><code>from pyspark.sql.types import StructType, StringType, IntegerType
  
  azdls_connection_string = &quot;My connection string&quot;
  queue_name = &quot;queu-name&quot;
  
  stream_schema = StructType() \
    .add(&quot;timestamp&quot;, StringType(), False) \
    .add(&quot;temperature&quot;, IntegerType(), False)
  
  ressource_group = &quot;&quot;
  cloudfiles_subid_telem = &quot;&quot;
  cloudfiles_clientid_telem = &quot;&quot;
  cloudfiles_clientsecret_telem = &quot;&quot;
  tenantid = &quot;&quot;
  conainer_name = &quot;mydb&quot;
  abs_fs = &quot;abfss://&quot; + conainer_name + &quot;@&quot; + dls_name + &quot;.dfs.core.windows.net&quot;
  
  read_stream = (
    spark.readStream.format(&quot;cloudFiles&quot;)
    .option(&quot;cloudFiles.useNotifications&quot;, True)
    .option(&quot;cloudFiles.format&quot;, &quot;json&quot;)
    .option(&quot;cloudFiles.connectionString&quot;, azdls_connection_string)
    .option(&quot;cloudFiles.resourceGroup&quot;, ressource_group)
    .option(&quot;cloudFiles.subscriptionId&quot;, cloudfiles_subid_telem)
    .option(&quot;cloudFiles.tenantId&quot;, tenantid)
    .option(&quot;cloudFiles.clientId&quot;, cloudfiles_clientid_telem)
    .option(&quot;cloudFiles.clientSecret&quot;, cloudfiles_clientsecret_telem)
    .option(&quot;cloudFiles.region&quot;, &quot;francecentral&quot;)
    .schema(stream_schema)
    .option(&quot;cloudFiles.includeExistingFiles&quot;, False)
    .load(abs_fs + &quot;/input&quot;)
  )
  
  checkpoint_path = abs_fs + &quot;/checkpoints&quot;
  out_path = abs_fs + &quot;/out&quot;
  
  df = read_stream.writeStream.format(&quot;delta&quot;) \
    .option(&quot;checkpointLocation&quot;, checkpoint_path) \
    .start(out_path)
  </code></pre>
  <p>And when i try to start my streaming i got a error. My permissions are correctly set because my Azure Queue is created. I don't find any informations in the documentation of autloader on the databricks website about this error.</p>
  <p>And here is my error :</p>
  <pre><code>java.io.IOException: Attempted read from closed stream.
  at org.apache.http.impl.io.ContentLengthInputStream.read(ContentLengthInputStream.java:165)
      at org.apache.http.conn.EofSensorInputStream.read(EofSensorInputStream.java:135)
      at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:284)
      at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:326)
      at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:178)
      at java.io.InputStreamReader.read(InputStreamReader.java:184)
      at java.io.Reader.read(Reader.java:140)
      at org.apache.commons.io.IOUtils.copyLarge(IOUtils.java:2001)
      at org.apache.commons.io.IOUtils.copyLarge(IOUtils.java:1980)
      at org.apache.commons.io.IOUtils.copy(IOUtils.java:1957)
      at org.apache.commons.io.IOUtils.copy(IOUtils.java:1907)
      at org.apache.commons.io.IOUtils.toString(IOUtils.java:778)
      at org.apache.commons.io.IOUtils.toString(IOUtils.java:759)
      at com.databricks.sql.aqs.EventGridClient.prettyResponse(EventGridClient.scala:428)
      at com.databricks.sql.aqs.EventGridClient.com$databricks$sql$aqs$EventGridClient$$errorResponse(EventGridClient.scala:424)
      at com.databricks.sql.aqs.EventGridClient$$anonfun$createEventSubscription$3.applyOrElse(EventGridClient.scala:235)
      at com.databricks.sql.aqs.EventGridClient$$anonfun$createEventSubscription$3.applyOrElse(EventGridClient.scala:229)
      at com.databricks.sql.aqs.EventGridClient.executeRequest(EventGridClient.scala:387)
      at com.databricks.sql.aqs.EventGridClient.createEventSubscription(EventGridClient.scala:226)
      at com.databricks.sql.aqs.autoIngest.AzureEventNotificationSetup.$anonfun$setupEventGridSubscription$1(AzureEventNotificationSetup.scala:135)
      at scala.Option.getOrElse(Option.scala:189)
      at com.databricks.sql.aqs.autoIngest.AzureEventNotificationSetup.setupEventGridSubscription(AzureEventNotificationSetup.scala:121)
      at com.databricks.sql.aqs.autoIngest.AzureEventNotificationSetup.&lt;init&gt;(AzureEventNotificationSetup.scala:75)
      at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
      at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
      at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
      at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
      at com.databricks.sql.fileNotification.autoIngest.EventNotificationSetup$.$anonfun$create$1(EventNotificationSetup.scala:66)
      at com.databricks.sql.fileNotification.autoIngest.ResourceManagementUtils$.unwrapInvocationTargetException(ResourceManagementUtils.scala:42)
      at com.databricks.sql.fileNotification.autoIngest.EventNotificationSetup$.create(EventNotificationSetup.scala:50)
      at com.databricks.sql.fileNotification.autoIngest.CloudFilesSourceProvider.$anonfun$createSource$2(CloudFilesSourceProvider.scala:162)
      at scala.Option.getOrElse(Option.scala:189)
      at com.databricks.sql.fileNotification.autoIngest.CloudFilesSourceProvider.createSource(CloudFilesSourceProvider.scala:154)
      at org.apache.spark.sql.execution.datasources.DataSource.createSource(DataSource.scala:306)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$1.$anonfun$applyOrElse$1(MicroBatchExecution.scala:93)
      at scala.collection.mutable.HashMap.getOrElseUpdate(HashMap.scala:86)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$1.applyOrElse(MicroBatchExecution.scala:90)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$1.applyOrElse(MicroBatchExecution.scala:88)
      at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDown$1(TreeNode.scala:322)
      at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:80)
      at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:322)
      at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDown(LogicalPlan.scala:29)
      at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown(AnalysisHelper.scala:166)
      at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown$(AnalysisHelper.scala:164)
      at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29)
      at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29)
      at org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:311)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution.logicalPlan$lzycompute(MicroBatchExecution.scala:88)
      at org.apache.spark.sql.execution.streaming.MicroBatchExecution.logicalPlan(MicroBatchExecution.scala:68)
      at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:346)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:269)
  </code></pre>
  <p>Thank you in advance</p>