GzipDecompressor: fix inflater memory leak

[pjfanning]

See #1133

GzipDecompressor allocates a new Inflater per gzip stream but did not explicitly release its native memory. Under collectors with relaxed off-heap reclamation such as ZGC, repeated request decompression can accumulate unreclaimed native buffers and eventually OOM.

• Inflater lifecycle cleanup

  • Add explicit Inflater.end() cleanup for GzipDecompressor
  • Make cleanup idempotent so termination paths can safely converge
  • Run cleanup from stage shutdown (postStop) to cover normal completion, failure, and cancellation

• Truncation handling

  • Release the inflater before failing truncated gzip streams
  • Preserve the existing Truncated GZIP stream behavior while ensuring native resources are not left behind

• Targeted coverage

  • Add focused tests that verify the inflater is released on:
    • successful decode completion
    • early downstream cancellation
    • truncation failure

private def cleanupInflater(): Unit =
  if (!inflaterEnded) {
    inflaterEnded = true
    inflater.end()
  }

override def postStop(): Unit = cleanupInflater()

override def onTruncation(): Unit = {
  cleanupInflater()
  failStage(new ZipException("Truncated GZIP stream"))
}

[He-Pin]

Maybe we can mark it as null ? instead of an inflaterEnded

[pjfanning]

I prefer the boolean to making inflater a var and then checking for nulls. The nulls approach makes cleanup harder.

    private def cleanupInflater(): Unit =
      if (!inflaterEnded) {
        inflaterEnded = true
        inflater.end()
      }

This is not thread-safe.

    private def cleanupInflater(): Unit =
      if (inflater != null) {
        inflater.end()
      }

This is thread-safe but introduces a local variable.

    private def cleanupInflater(): Unit = {
      val localInflater = inflater
      if (localInflater != null) {
        localInflater.end()
      }
   }

[pjfanning]

@He-Pin do you think is ok to merge?

[He-Pin]

lgtm