title: Akka stream throws ParsingException when start decompressing .gz file from middle offset
body: <p>I am reading <strong>.gz</strong> file from <strong>AWS S3 bucket</strong> using Akka streaming, The application reads chunk of data and process it.</p>
  <p>The following code is working fine when I read file from beginning but if I start file from particular offset then Decompression fails and throws ParsingException.</p>
  <p><strong>Code:</strong></p>
  <pre><code>object StreamApp extends App {
  
      // Run 1
      //startStreaming(0)
      // Run 2
      startStreaming(33339)
  
      def startStreaming(pointer: Long): Unit = {
          println(&quot;Stream App Starting&quot;)
  
          implicit val system = ActorSystem()
          val materializerSettings = ActorMaterializerSettings(system)
          implicit val materializer = ActorMaterializer(materializerSettings)
          implicit val dispatcher: ExecutionContextExecutor = system.dispatcher
          implicit val timeout: Timeout = Timeout(1 seconds)
          
          val accessKey = &quot;aws-access-key&quot;
          val secretAccessKey = &quot;aws-secret-access-key&quot;
  
          val awsCredentials = new AWSStaticCredentialsProvider(
              new BasicAWSCredentials(accessKey, secretAccessKey)
          )
          
          val s3Region: String = &quot;s3-region-name&quot;
          val s3Bucket: String = &quot;s3-bucket-name&quot;
          val s3DataFile: String = &quot;s3-object-path.gz&quot;
          
          val settings = new S3Settings(MemoryBufferType, None, awsCredentials, s3Region, false)
  
          val s3Client = new S3Client(settings)(system, materializer)
  
          val currentOffset: Long = pointer
          val source: Source[ByteString, NotUsed] =
              s3Client
                  .download(s3Bucket, s3DataFile, ByteRange.fromOffset(currentOffset))
  
          val flowDecompress: Flow[ByteString, ByteString, NotUsed] =
              Flow[ByteString].via(
                  Compression.gunzip()
              )
  
          val flowToString: Flow[ByteString, String, NotUsed] =
              Flow[ByteString].map(_.utf8String)
  
          val sink: Sink[String, Future[Done]] = Sink.foreach(println)
  
          val (killSwitch, graph) =
              source
                  .via(flowDecompress)
                  .via(flowToString)
                  .viaMat(KillSwitches.single)(Keep.right)
                  .toMat(sink)(Keep.both)
                  .run()
  
          graph.onComplete {
              case Success(_) =&gt;
                  println(&quot;Stream App &gt;&gt; File Data Extractor &gt;&gt; Stream completed successfully&quot;)
                  killSwitch.shutdown()
          }
      }
  }
  </code></pre>
  <p>The exception as follows:</p>
  <pre><code>[ERROR] [02/26/2018 12:13:10.682] [default-akka.actor.default-dispatcher-2] [akka.dispatch.Dispatcher] Failure(akka.stream.impl.io.ByteStringParser$ParsingException: Parsing failed in step ReadHeaders) (of class scala.util.Failure)
  scala.MatchError: Failure(akka.stream.impl.io.ByteStringParser$ParsingException: Parsing failed in step ReadHeaders) (of class scala.util.Failure)
      at StreamApp$$anonfun$startStreaming$1.apply(StreamApp.scala:71)
      at StreamApp$$anonfun$startStreaming$1.apply(StreamApp.scala:71)
      at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36)
      at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55)
      at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:91)
      at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91)
      at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91)
      at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72)
      at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:90)
      at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40)
      at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:43)
      at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
      at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
      at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
      at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
  </code></pre>
  <p>I debug the code and following is my observation</p>
  <p>Execution on this <a href="https://github.com/akka/akka-http/blob/master/akka-http/src/main/scala/akka/http/scaladsl/coding/Gzip.scala#L81" rel="nofollow noreferrer">line</a>, the <strong>readByte()</strong> returns 3 and hence fails to understand the input is in .gz format</p>
  <p>I am not able to decompress the .gz file's data chunk, I want guideline about this problem, if possible then please provide the pointers and correct me if I am wrong.</p>
  <p><strong>Version:</strong></p>
  <blockquote>
  <p>Akka Version = &quot;2.5.9&quot;</p>
  <p>Akka Http Version = &quot;10.0.11&quot;</p>
  <p>Alpakka Version = &quot;0.14&quot;</p>
  </blockquote>