Sinks commit the consumer-group offset before delivery is confirmed (data loss on write failure)

[mattp5657]

Description

Every sink connector's consumer is configured with
AutoCommit::When(AutoCommitWhen::PollingMessages)
(core/connectors/runtime/src/sink.rs:522), which commits the read offset
to the server the moment a batch is polled off the topic — before the
plugin's consume() runs. If consume() fails (the downstream write
errors), the offset is already committed, so that batch is never
redelivered. No error, no duplicate, no gap signal.

This is shared runtime code, not per-plugin: clickhouse_sink,
delta_sink, doris_sink, elasticsearch_sink, http_sink,
iceberg_sink, influxdb_sink, mongodb_sink, postgres_sink,
quickwit_sink, s3_sink, stdout_sink, surrealdb_sink — all 13 sink
plugins go through this same consumer setup and are affected identically.

Net effect: at-most-once delivery where at-least-once is expected.

Affected area / component

Connectors

Proposed solution

Switch the consumer to AutoCommit::Disabled and call
consumer.store_offset() explicitly after consume() returns Ok.
Runtime-only change — no plugin rebuild needed, no FFI surface touched.

Alternatives considered

No response

Contribution

I'm willing to submit a pull request to fix this bug.