RabbitMQ messages are acknowledged before Kafka producer confirmation, causing data loss

# What happened?

The source connector acknowledges RabbitMQ deliveries before Kafka has persisted the corresponding records.

`AmqpFragment#getReceiver` opens a ProtonJ2 receiver with default options:

```java
connection.openReceiver(address)
```

With ProtonJ2 1.1.0, the defaults are `autoAccept=true` and `autoSettle=true`. As a result, `AmqpSourceData#getNativeItemIterator` calling `receiver.tryReceive()` schedules an `Accepted` and settled disposition before the delivery is returned. This happens before message extraction, `SourceRecord` creation, `SourceTask#poll`, and Kafka producer acknowledgement.

Neither `AmqpSourceTask` nor `framework-for-source-connectors` 0.3.0 handles `SourceTask#commitRecord`, so the RabbitMQ acknowledgement is not tied to Kafka delivery success.

This creates an at-most-once handoff and can permanently lose messages when:

- The worker stops or crashes after `tryReceive()` but before Kafka producer acknowledgement.
- The Kafka producer, converter, or serializer fails.
- Records are still buffered by the source task when it stops.
- `AmqpExtractor` catches a `ClientException` or `IOException` and returns an empty result.

Relevant locations at commit `be44dce`:

- `common/src/main/java/io/aiven/kafka/connect/amqp/common/config/AmqpFragment.java:167-170`
- `source/src/main/java/io/aiven/kafka/connect/amqp/source/AmqpSourceData.java:82-102`
- `source/src/main/java/io/aiven/kafka/connect/amqp/source/AmqpSourceTask.java:33-72`
- `source/src/main/java/io/aiven/kafka/connect/amqp/source/extractor/AmqpExtractor.java:107-124`

Minimal reproduction:

1. Publish one message to RabbitMQ.
2. Poll it through `AmqpSourceTask`.
3. Stop the task before invoking any commit callback.
4. Connect a new receiver to the same RabbitMQ address.

The message is not redelivered. In an A/B integration test, changing only the receiver to `autoAccept(false)` caused the same message ID to be redelivered, confirming the cause.

# What did you expect to happen?

RabbitMQ deliveries should remain unsettled until Kafka confirms that the corresponding record has been persisted. If the task stops or fails before that confirmation, RabbitMQ should redeliver the message.

The connector should provide at-least-once delivery. A failure between Kafka acknowledgement and RabbitMQ settlement may cause a duplicate, but it must not cause data loss.

# What else do we need to know?

Suggested fix:

- Open the receiver with `autoAccept(false)` and use manual settlement.
- Track pending source-record-to-delivery mappings.
- Call `Delivery.accept()` from `commitRecord(record, metadata)` only after a successful Kafka write.
- Release pending deliveries on failure and shutdown.
- Do not acknowledge `metadata == null` without an explicit filter, reject, or DLQ policy; it can also represent a tolerated producer failure.
- Bound unsettled deliveries using manual credit or equivalent backpressure.
- Propagate extraction errors or route them through an explicit reject/DLQ policy instead of returning an empty stream.

Acceptance criteria:

- An uncommitted delivery is redelivered after task restart.
- A successfully committed delivery is not redelivered.
- Producer and extraction failures do not remove the RabbitMQ message.
- The number of outstanding unsettled deliveries remains bounded.

Observed with the current `main` branch at `be44dce`, ProtonJ2 1.1.0, and RabbitMQ 4.2.x.

Start by tracing receiver setup in common/src/main/java/io/aiven/kafka/connect/amqp/common/config/AmqpFragment.java and delivery handling in AmqpSourceData.java, AmqpSourceTask.java, and AmqpExtractor.java. Run the minimal restart reproduction and inspect commitRecord behavior. Done means uncommitted deliveries are redelivered, committed deliveries are not, failures preserve messages, and unsettled deliveries remain bounded.

索引モデルが issue の本文から書いたものです。