Support pre-decompression header skipping in DecompressionEngine

[yavmanis]

Problem Statement

Some data producers (log forwarders, Kafka connectors, ETL tools) prepend an application-level metadata header to compressed payloads before writing to storage. The current DecompressionEngine implementations have no mechanism to handle these headers, causing decompression failures.

Example File Format

A Palo Alto Networks firewall log forwarder writes files to S3 in this format:

{"compression": "snappy", "topic": "fwd_d6b6a0e7-5519-442a-b69c-60c759501fa3_1435757634.firewall.hipmatch", "key": "0-0-panw.firewall.hipmatch-genericjson", "num_bytes": 1260, "num_records": 10}
<snappy-compressed binary payload>

• Line 1: Plain-text JSON metadata (application-level, not part of any compression spec)
• Remaining bytes: Raw Snappy block-compressed NDJSON records

With compression: "snappy", SnappyDecompressionEngine passes the entire stream (including the header) to SnappyInputStream, which fails due to this custom header

Proposed Solution

Add a decompression_options configuration at the codec level (data-prepper-plugins/common) that allows skipping bytes from the raw stream before decompression is applied. This will be implemented as a decorator around existing DecompressionEngine implementations, keeping them unchanged and stateless.

Configuration

compression: "snappy"
decompression_options:
  skip_lines: 1

Available options (mutually exclusive — exactly one may be specified):

Option	Description
skip_lines	Skip N newline-delimited lines before decompressing
skip_bytes	Skip a fixed number of bytes before decompressing
skip_until_delimiter	Skip all bytes until the first occurrence of a delimiter string
skip_until_pattern	Skip bytes matching a regex pattern at the start of the stream

If decompression_options is omitted, existing behavior is preserved.

Example Pipeline

s3-pipeline:
  source:
    s3:
      compression: "snappy"
      decompression_options:
        skip_lines: 1
      codec:
        newline:
  sink:
    - opensearch:
        index: "firewall-logs"

[bagmarnikhil]

Thanks for the write-up. The approach is correct. A skip before decompression is necessary. I have two suggestions:

1. Add one option, not four. Only skip_lines is useful for this case.

• Do not add skip_bytes. The length of the JSON header changes for each batch. A fixed byte count corrupts most objects.
• Do not add skip_until_delimiter: "\n". It does the same thing as skip_lines.
• Do not add skip_until_pattern. A regular expression on raw compressed bytes is not reliable.

Start with skip_lines. Add the other options only if a use case needs them.

2. Use a different name. The name skip_lines is already in the newline codec. That option runs after decompression. Use a name such as skip_header_lines to prevent confusion.

source:
  s3:
     compression: "snappy"
     decompression_options:
        skip_header_lines: 1
     codec:
        newline: