feat(kafka source): add decompression option for producer-compressed payloads

[cjford]

Summary

Adds decompression config option to kafka sources. This allows unwrapping of application-level (not kafka protocol-level) compression from message payloads, so that their original forms can be used for framing/decoding and downstream components. Supports gzip, zlib, and zstd algorithms, and furthermore allows optional zstd dictionary_path if one was used during compression.

On decompression failure, a component error is logged and the message skipped. Its offset is still committed so that partition consumption continues, which is consistent with existing behavior for decoding failures.

This pattern should be re-usable for other sources, but needs some wiring in each including specific failure behavior. I also explored more general alternative implementations (e.g. decompression as part of framing, separate pre-framing Decoder step), but found them not workable universally, particularly with streaming inputs.

Code generated using Claude Fable 5

Vector configuration

sources:
  in:
    type: kafka
    bootstrap_servers: localhost:9092
    group_id: vector-test
    topics: ["vector-test"]
    auto_offset_reset: beginning
    decoding:
      codec: json
    decompression:
      algorithm: zstd
      dictionary_path: /tmp/vector-test.dict

sinks:
  out:
    type: console
    inputs: [in]
    encoding:
      codec: json

How did you test this PR?

Tested manually with local (macOS aarch64) kafka/vector setup, using zstd to compress raw messages, kcat to publish them to kafka, and a console sink to verify output. Covered various scenarios:

• Plain payload regression check
• gzip/zlib/zstd decompression works as expected
• zstd decompression with dictionary works as expected
• Failed decompression skips message and commits offset
• Multi-message payload is correctly decompressed -> framed
• Size limits enforced for both upfront allocation and decompressed size
• Error messaging for mismatched compression algorithms, invalid dictionary path, mismatched dictionaries, dictionary_path with unsupported algorithm

Is this a breaking change?

• Yes
• No

Does this PR include user facing changes?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the no-changelog label to this PR.

References

• Related: ZStd compression should allow the configuration and use of custom dictionaries #19435 - Also involves zstd dictionary support, but specifically for sink compression while this PR is limited to decompression

[github-actions]

All contributors have signed the CLA ✍️ ✅
_{Posted by the CLA Assistant Lite bot.}

[cjford]

I have read the CLA Document and I hereby sign the CLA