connectors: batches are tagged with the decoder's schema, not the payload's, so avro/proto/flatbuffer reach sinks as the wrong Payload variant

### Bug description

When a sink's `[[streams]]` entry sets `schema = "avro"`, `"proto"` or `"flatbuffer"`, the sink receives its messages as the wrong `Payload` variant.

The runtime tags each batch with the decoder's schema rather than the type the decoder returned. Those three decoders all return `Payload::Json` under the config the runtime gives them, so the tag and the bytes disagree. The SDK rebuilds the payload from the tag, so a sink is handed a `Payload::Avro` holding JSON.

What happens next depends on the sink. Sinks that only handle `Payload::Json` — ClickHouse, Delta, Elasticsearch, Meilisearch, Quickwit, Iceberg — log the message as an unsupported payload type and drop it. Sinks that accept the raw variants — S3, HTTP, SurrealDB — base64-encode the JSON text and store it as though it were Avro bytes.

Expected: the sink receives `Payload::Json`, because that is what the decoder produced.

**Root cause**

```rust
// core/connectors/runtime/src/sink.rs:359-363
let messages_metadata = MessagesMetadata {
partition_id,
current_offset,
schema: decoder.schema(), // Schema::Avro
};
```

The bytes sent alongside come from `Payload::try_into_vec()`, which for the `Payload::Json` the Avro decoder returned gives JSON (`core/connectors/sdk/src/lib.rs:163-165`). The same tag is put on `RawMessages` at `:731`. The SDK rebuilds from it (`sdk/src/sink.rs:199`, then `lib.rs:263`: `Schema::Avro => Payload::Avro(value)`).

No connector config avoids this. The runtime builds the Avro decoder with `..AvroConfig::default()` (`sink.rs:532-536`), so `extract_as_json` cannot be turned off from a connector file, and Proto and FlatBuffer get `Schema::decoder()` defaults with no knob at all.

Sources are unaffected: a source plugin tags its own `ProducedMessages.schema` and the encoder takes the `Payload` directly (`source.rs:430-435`), with no rebuild step.

**Which schemas are affected**

| `schema` | decoder returns | batch tagged | sink receives |
|---|---|---|---|
| `json`, `text`, `raw` | matching variant | matching | works |
| `avro` | `Payload::Json` (`decoders/avro.rs:169`) | `Avro` | `Payload::Avro` |
| `flatbuffer` | `Payload::Json` (`decoders/flatbuffer.rs:109`) | `FlatBuffer` | `Payload::FlatBuffer` |
| `proto` | `Payload::Json`, if the message is an `Any` (`decoders/proto.rs:529`) | `Proto` | `Payload::Raw` |

The three that work are the ones whose decoder output matches its own `schema()`. Avro and FlatBuffer default to `extract_as_json: true`; Proto with no schema configured falls through to `decode_as_any` (`proto.rs:264-265`), which also returns JSON. Proto ends up as `Payload::Raw` rather than `Payload::Proto` because `try_into_payload` re-decodes the bytes as a `prost_types::Any` and falls back to `Raw` when that fails (`lib.rs:260`).

**Impact**

For the Json-only sinks the batch is lost without an error being raised. The ClickHouse sink ends up with an empty body and returns `Ok(())` (`sinks/clickhouse_sink/src/sink.rs:146-152`), and offsets are committed with the poll request (`AutoCommit::When(AutoCommitWhen::PollingMessages)`, `runtime/src/sink.rs:522`) — the SDK documents that mode as sending "the commit with the poll request itself, before your code sees the batch" (`core/sdk/src/clients/consumer.rs:501`). So the offsets advance before the sink runs, and restarting the connector does not redeliver.

The batch is still counted as consumed. A run of 100 messages logs `Consumed 100 messages`, and the connector reports `Processed 0 messages.` at close.

Related to #3950 and PR #3954, which bind the sink's return status and let a sink defer the offset commit. Neither covers this: the ClickHouse sink returns `Ok(())` on an empty body, so even with `offset_commit = after_consuming` the batch would be committed as successfully written.

**Suggested fix**

Tag the batch with the payload's type rather than the decoder's. Changing each decoder's `schema()` would not be enough on its own: transforms run after decoding and can change the type again, and the Proto decoder returns either `Json` or `Raw` depending on the path it takes.

Adding a `Payload::schema()` and using it where `MessagesMetadata` and `RawMessages` are built (`runtime/src/sink.rs:362` and `:731`) covers all three formats in one place. I can send a PR either way; say which you prefer.

**Compatibility**

The fix changes what three sinks write. S3, HTTP and SurrealDB match on `Payload::Avro` and base64-encode it, so an avro stream reaches them today as base64 of JSON text, stored as though it were Avro bytes. Nothing reports this. After the fix they receive `Payload::Json` and write a JSON document, so anyone consuming their current output would see it change shape. That output is corrupt.

The Json-only sinks go from dropping the batch to writing it correctly. Streams configured `json`, `text` or `raw` are unaffected, because the decoder's schema and the payload's type already agree and the tag does not move.

Two related gaps:

- `StreamConsumerConfig` only exposes `avro_schema_json` and `avro_schema_path`, so flatbuffer and proto cannot be given a schema through the runtime at all.
- Nothing exercises a non-JSON schema through the runtime's sink path, and all ten configs under `runtime/example_config/connectors/` use `schema = "json"`. A test across the decoders would catch this class of problem.

### Affected area / component

Connectors

### Deployment

Compiled from source

### Versions

master @ `412014a` — `iggy-connectors` 0.5.0-edge.6, `iggy_connector_sdk` 0.4.0-edge.3, `iggy_connector_clickhouse_sink` 0.2.0-edge.4

### Hardware / environment

macOS arm64, local build; not hardware-dependent.

### Sample code

```toml
type = "sink"
key = "clickhouse"
enabled = true
version = 0
name = "ClickHouse sink"
path = "target/release/libiggy_connector_clickhouse_sink"
verbose = true

[[streams]]
stream = "repro"
topics = ["avro"]
schema = "avro"
avro_schema_json = '{"type":"record","name":"Event","fields":[{"name":"id","type":"long"},{"name":"name","type":"string"}]}'
batch_length = 100
poll_interval = "5ms"
consumer_group = "repro-sink"

[plugin_config]
url = "http://localhost:8123"
database = "default"
table = "repro"
insert_format = "row_binary"
```

Producer — an ordinary `IggyProducer`, with each payload an Avro datum for that schema:

```rust
let schema = apache_avro::Schema::parse_str(SCHEMA)?;
let record = apache_avro::types::Value::Record(vec![
("id".to_owned(), apache_avro::types::Value::Long(id)),
("name".to_owned(), apache_avro::types::Value::String(format!("row-{id}"))),
]);
let datum = apache_avro::writer::datum::GenericDatumWriter::builder(&schema)
.build()?
.write_value_to_vec(record)?;
let message = IggyMessage::builder().payload(datum.into()).build()?;
```

### Logs

100 Avro messages into a single-partition topic, ClickHouse sink in `row_binary` mode:

```
INFO iggy_connectors::sink: Processing 100 messages for sink connector with ID: 1
ERROR connector: connector_target="iggy_connector_clickhouse_sink::body"
RowBinary mode: skipping unsupported payload type at offset 0
ERROR ... at offset 1, 2, ... 99
ERROR connector: connector_target="iggy_connector_clickhouse_sink::sink"
ClickHouse sink ID: 1 — no serialisable messages in batch of 100
INFO iggy_connectors::sink: Consumed 100 messages in 675µs for sink connector with ID: 1
INFO connector: ClickHouse sink ID: 1 closed. Processed 0 messages.
```

`SELECT count() FROM default.repro` returns 0.

Restarting the connector against the same consumer group redelivers nothing: no batch, no errors, still no rows. The offsets were committed at poll time, so the 100 messages cannot be recovered without resetting them by hand.

Tagging the batch from the payload's type rather than the decoder's makes the same configuration land all 100 rows with the expected values, which confirms the cause.

### Iggy server config

Default server config; not involved.

### Reproduction

1. Run an Iggy server and a ClickHouse instance with
`CREATE TABLE repro (id Int64, name String) ENGINE = MergeTree ORDER BY id`.
2. Produce Avro-encoded records matching the schema above to `repro/avro`.
3. Run `iggy-connectors` with the sink config above.
4. The log fills with `skipping unsupported payload type` and the table stays
empty.
5. Restart the connector: nothing is redelivered, and the table is still empty.

### Contribution

- [x] I'm willing to submit a pull request to fix this bug

### Good first issue

- [ ] I think this could be a good first issue for a new contributor