compress processor allocates a new gzip writer per message (~1 MB, 19 allocs)

Summary

The compress processor allocates a fresh compressor for every message. For gzip that costs
~1 MB and 19 allocations per message regardless of payload size, because a new
gzip.Writer allocates its full deflate window on first write and is then discarded.

Reusing the writer via Reset instead makes the same work ~20x faster and allocates
~640x less. Benchmarks below, added in-tree so they're reproducible.

We hit this in production: across four profiled Redpanda Connect pipelines, 49–54% of CPU time
is garbage collection, and the dominant allocator is this compressor. We make ~74,000 gzip'd
HTTP enrichment calls per second, so this is ~74,000 compressors built and thrown away per second.

Where it comes from

internal/impl/pure/algorithms.go — AddKnownCompressionAlgorithm synthesises a CompressFunc
from the registered CompressWriter, and that synthesised function runs per message:

a.CompressFunc = func(level int, b []byte) ([]byte, error) {
    var buf bytes.Buffer                       // (1) fresh output buffer
    wtr, err := a.CompressWriter(level, &buf)  // (2) fresh gzip.Writer  <- the expensive one
    ...
    return buf.Bytes(), err
}

and for gzip (same file, ~line 208):

CompressWriter: func(level int, w io.Writer) (io.Writer, error) {
    aw, err := gzip.NewWriterLevel(w, level)
    if err != nil {
        return nil, err
    }
    return &CombinedWriteCloser{Primary: aw, Sink: w}, nil   // (3) wrapper
},

gzip.NewWriterLevel itself is cheap (160 B, 1 alloc — it's lazy). The ~1 MB lands on the
first Write, when the deflate encoder allocates its window and tables. Since the writer is
discarded immediately, that cost is paid on every message:

BenchmarkNewWriterThenWrite-10    150080 ns/op    1076368 B/op    17 allocs/op
BenchmarkResetThenWrite-10          7214 ns/op        897 B/op     3 allocs/op

Benchmark

Added as internal/impl/pure/algorithms_alloc_bench_test.go. Payload is ~2.8 KB of JSON, the
shape of an HTTP request body. BenchmarkCompressGzipCurrent exercises the real
strToCompressAlg("gzip").CompressFunc; BenchmarkCompressGzipPooled is an illustrative
sync.Pool equivalent to size the headroom.

$ go test ./internal/impl/pure/ -run XXX -bench BenchmarkCompressGzip -benchmem -benchtime=3000x -count=5

BenchmarkCompressGzipCurrent-10    3000    148856 ns/op   18.66 MB/s   1076436 B/op   19 allocs/op
BenchmarkCompressGzipPooled-10     3000      7444 ns/op  371.45 MB/s      1704 B/op    5 allocs/op

	current	pooled
time	148,856 ns/op	7,444 ns/op	20x faster
allocated	1,076,436 B/op	~1,704 B/op	~630x less
allocations	19/op	5/op
throughput	18.7 MB/s	371.5 MB/s

Suggested direction

Pool the writer rather than constructing one per message. The constraint worth flagging: level
is a per-call argument, and gzip.Writer.Reset preserves the level it was constructed with — so
a single shared pool is incorrect. It needs to be keyed by level (and by algorithm).

Sketch, matching the benchmark:

var gzipPools sync.Map // map[int]*sync.Pool

// on the hot path: pool.Get() -> w.Reset(&buf) -> w.Write(src) -> w.Close()
//                  -> w.Reset(nil) -> pool.Put(w)

Points that need care in a real implementation:

• Reset before reuse, and Reset(nil) before returning to the pool so the writer doesn't
  retain a reference to the previous output buffer.
• Do not return a writer to the pool on an error path.
• The generic CompressFunc wrapper Close()s whatever CompressWriter returns, so a pooled
  implementation probably wants a pooled CombinedWriteCloser whose Close returns the writer —
  or a separate optional pooled path on KnownCompressionAlgorithm rather than changing the shape
  of CompressWriter.
• Consider capping the pooled bytes.Buffer so one very large message doesn't pin memory.

The included test TestPooledGzipRoundTripsConcurrentlyAtMixedLevels covers the obvious failure
mode — 24 goroutines compressing concurrently at levels 1, 5 and 9, asserting every output
decompresses back to its own input. It passes.

decompress has the same shape for readers and would presumably want the same treatment.

Environment

• benthos main @ d8319c9
• go1.26.6, darwin/arm64 (Apple silicon, 10 logical CPUs)
• github.com/klauspost/compress as currently pinned

Would you accept a PR?

Happy to implement this if the direction looks right — particularly on whether you'd prefer an
optional pooled path on KnownCompressionAlgorithm over changing CompressWriter, and whether
pooling the output buffer is wanted or out of scope.

Start in internal/impl/pure/algorithms.go at AddKnownCompressionAlgorithm and the gzip CompressWriter, then run internal/impl/pure/algorithms_alloc_bench_test.go and TestPooledGzipRoundTripsConcurrentlyAtMixedLevels. Done means the chosen pooling design preserves mixed-level concurrent round trips, avoids retaining prior buffers, and improves the reported gzip benchmark without breaking the existing compression path.