chore(codecs): centralize events_dropped emission for batch encoding errors

[pront]

Summary

After #25156 (Parquet encoding) landed, events_dropped emission across batch encoding error paths is inconsistent -- Parquet covers most paths but misses some, Arrow IPC covers almost none. Every new error path requires the author to remember to add emission manually, which is easy to miss.

This PR moves events_dropped emission to a single wrapper at the batch encoding chokepoint (encode_input), so all current and future error paths are covered automatically. Individual internal events keep their error logs and component_errors_total counters for diagnostics.

Risk: The non_log_count partial drop in Parquet (non-log events filtered before encoding) can double-count if the batch also fails to encode. This is a rare intersection and acceptable.

Vector configuration

Used to verify metrics across three error scenarios. Base config (swap batch_encoding and schema file per scenario):

sources:
  demo:
    type: demo_logs
    format: json
    interval: 0.5
  internal_metrics:
    type: internal_metrics
    scrape_interval_secs: 5

sinks:
  s3:
    type: aws_s3
    inputs: ["demo"]
    bucket: test-bucket
    region: us-east-1
    endpoint: http://localhost:4566
    batch:
      max_events: 2
      timeout_secs: 2
    batch_encoding:
      codec: parquet
      schema_mode: strict          # or relaxed for scenarios 2 & 3
      schema_file: /tmp/test.schema
    encoding:
      codec: text
    compression: none
    auth:
      access_key_id: test
      secret_access_key: test
  metrics_console:
    type: console
    inputs: ["internal_metrics"]
    encoding:
      codec: json

How did you test this PR?

• make check-clippy -- clean
• cargo nextest run -p codecs --features arrow,parquet -- 277 passed
• cargo nextest run --package vector --lib sinks::aws_s3 -- 2 passed
• cargo nextest run --package vector --lib sinks::clickhouse -- 29 passed
• cargo nextest run --package vector --lib sinks::util::encoding -- 10 passed
• Local run with three error scenarios, all emitting Events dropped and component_errors_total:

Scenario 1 -- Strict mode (extra field rejected)

Schema: message logs { required binary message (STRING); }

codecs::internal_events: Failed writing bytes.
  error=SerializingError(Strict schema mode: event contains field 'host' not in schema)

component_events_dropped: Events dropped intentional=false count=2 reason="Failed writing bytes."

{"name":"component_errors_total", "tags":{"component_id":"s3", "error_type":"encoder_failed", "stage":"sending"}, "counter":{"value":5.0}}

Scenario 2 -- Null constraint violation (required field missing)

Schema: message logs { required binary message (STRING); required binary nonexistent_field (STRING); }

codecs::internal_events: Schema constraint violation.
  error=Null value for non-nullable field 'nonexistent_field'

codecs::internal_events: Failed writing bytes.
  error=SerializingError(Null value for non-nullable field 'nonexistent_field')

component_events_dropped: Events dropped intentional=false count=2 reason="Failed writing bytes."

{"name":"component_errors_total", "tags":{"error_code":"encoding_null_constraint", "error_type":"encoder_failed", "stage":"sending"}, "counter":{"value":5.0}}

Scenario 3 -- Type mismatch (string field vs int64 schema)

Schema: message logs { required int64 message; }

This is a build_record_batch ArrowJsonDecode error that previously had zero events_dropped emission. Now covered by the centralized wrapper.

codecs::internal_events: Failed writing bytes.
  error=SerializingError(Arrow JSON decoding error: failed to parse "..." as Int64)

component_events_dropped: Events dropped intentional=false count=2 reason="Failed writing bytes."

{"name":"component_errors_total", "tags":{"error_type":"encoder_failed", "stage":"sending"}, "counter":{"value":5.0}}

Change Type

• Bug fix
• New feature
• Dependencies
• Non-functional (chore, refactoring, docs)
• Performance

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: feat(aws_s3 sink): add Parquet encoder with schema_file and auto infer schema support #25156 (Parquet encoding PR)
• Related: fix(s3 sink): encoding failures after split_input silently ack batches as delivered #25191 (pre-existing silent ack bug on encode failures)

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 2caa7428ec

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: a0def5bf25

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".