title: Failure Message: java.io.IOException: Failed to write 0 bytes in 'gs://<my_bucket_name>/events/1645565167200/.temp-beam-0a2f3ce9-1df3-45a5-a7f0-1cb1'
body: <p>I am using Spotify's flink on kubernetes operator and running a simple Apache Beam pipeline which reads from Kafka and writes to GCS in parquet format.
  I am new to Apache Beam, flink and still trying to figure out the concepts.</p>
  <p>The pipeline has a window of 2 minutes.
  I use <code>withCreateTime(Duration.standardMinutes(0))</code> to get Kafka create time and <code>withNaming(FileNaming.getNaming())</code> to create directories with yyyy-MM-dd-HH format to write files for hourly window in respective directories.</p>
  <p>I also use <code>withTempDirectory()</code> to place temporary files.</p>
  <p>When I run the pipeline and there is a huge lag of millions of records in Kafka I get the following exception after writing the first batch of files.</p>
  <pre><code>ERROR org.apache.beam.sdk.io.FileBasedSink$Writer - Beginning write to gs://my_bucket_name/events/1645565167200/.temp-beam-0a2f3ce9-1df3-45a5-a7f0-149a57848a85/cb125f19-61c1-46c1-bc24-89239206c75b failed, closing channel.
  
  
  java.io.IOException: Failed to write 0 bytes in 'gs://my_bucket_name/events/1645565167200/.temp-beam-0a2f3ce9-1df3-45a5-a7f0-149a57848a85/cb125f19-61c1-46c1-bc24-89239206c75b'
  234
      at com.google.cloud.hadoop.util.BaseAbstractGoogleAsyncWriteChannel.write(BaseAbstractGoogleAsyncWriteChannel.java:187) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  233
      at java.nio.channels.Channels.writeFullyImpl(Channels.java:78) ~[?:1.8.0_312]
  232
      at java.nio.channels.Channels.writeFully(Channels.java:101) ~[?:1.8.0_312]
  231
      at java.nio.channels.Channels.access$000(Channels.java:61) ~[?:1.8.0_312]
  230
      at java.nio.channels.Channels$1.write(Channels.java:174) ~[?:1.8.0_312]
  229
      at org.apache.beam.sdk.io.parquet.ParquetIO$Sink$BeamOutputStream.write(ParquetIO.java:1303) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  228
      at org.apache.beam.sdk.io.parquet.ParquetIO$Sink$BeamOutputStream.write(ParquetIO.java:1298) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  227
      at org.apache.parquet.hadoop.ParquetFileWriter.start(ParquetFileWriter.java:394) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  226
      at org.apache.parquet.hadoop.ParquetWriter.&lt;init&gt;(ParquetWriter.java:293) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  225
      at org.apache.parquet.hadoop.ParquetWriter$Builder.build(ParquetWriter.java:641) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  224
      at org.apache.beam.sdk.io.parquet.ParquetIO$Sink.open(ParquetIO.java:1233) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  223
      at org.apache.beam.sdk.io.FileIO$Write$ViaFileBasedSink$1$1.prepareWrite(FileIO.java:1377) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  222
      at org.apache.beam.sdk.io.FileBasedSink$Writer.open(FileBasedSink.java:961) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  221
      at org.apache.beam.sdk.io.WriteFiles$WriteShardsIntoTempFilesFn.processElement(WriteFiles.java:921) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  220
      at org.apache.beam.sdk.io.WriteFiles$WriteShardsIntoTempFilesFn$DoFnInvoker.invokeProcessElement(Unknown Source) ~[?:?]
  219
      at org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:232) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  218
      at org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:188) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  217
      at org.apache.beam.runners.flink.metrics.DoFnRunnerWithMetricsUpdate.processElement(DoFnRunnerWithMetricsUpdate.java:62) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  216
      at org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator.processElement(DoFnOperator.java:645) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  215
      at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
  214
      at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
  213
      at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
  212
      at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
  211
      at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
  210
      at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:50) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
  209
      at org.apache.beam.runners.flink.FlinkStreamingTransformTranslators$ToGroupByKeyResult.flatMap(FlinkStreamingTransformTranslators.java:1416) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  208
      at org.apache.beam.runners.flink.FlinkStreamingTransformTranslators$ToGroupByKeyResult.flatMap(FlinkStreamingTransformTranslators.java:1388) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  207
      at org.apache.flink.streaming.api.operators.StreamFlatMap.processElement(StreamFlatMap.java:47) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
  206
      at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
  205
      at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
  204
      at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
  203
      at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
  202
      at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
  201
      at org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator$BufferedOutputManager.emit(DoFnOperator.java:1171) ~[blob_p-ed6ce4c16e52d488153c17d014c718f2ca9e97c5-bfc37bda64724db1bd6699a32347c2d6:?]
  </code></pre>
  <p>Please advice</p>