Aiven-Open / Aiven-Open/cloud-storage-connectors-for-apache-kafka

Azure Blob Sink task remains RUNNING and retries indefinitely after non-recoverable Azure SDK errors

Ouverte
#612 8 commentaires 0 réactions 0 personnes assignées Voir sur GitHub
Langage dominant
Java
Étoiles
58
Forks
39
Merge moyen
2 j 10 h
PR mergées (30 j)
5

Description

### Description

We are using Kafka Connect with the Aiven Azure Blob Storage Sink Connector.

When the underlying Azure SDK returns an error during `flush()` or offset commit, the connector task remains in the `RUNNING` state. Kafka Connect rewinds to the last committed offsets and the connector repeatedly attempts to write the same data again.

For non-recoverable errors such as `ContainerNotFound`, this results in an apparently endless retry loop instead of eventually failing the task.

### Observed behaviour

In our case, the configured Azure Blob Storage container did not exist. The Azure SDK returned:

- HTTP status: `404`
- Azure error code: `ContainerNotFound`
- Message: `The specified container does not exist`

The exception is raised from:

`io.aiven.kafka.connect.azure.sink.AzureBlobSinkTask.flushFile(AzureBlobSinkTask.java:160)`

The connector subsequently reports:

`Offset commit failed, rewinding to last committed offsets`

However, the task continues running and repeatedly attempts to process and flush the same records.

This causes:

- High CPU utilisation, probably due to the large number of repeatedly generated exceptions and stack traces
- Continuously increasing consumer lag because offsets cannot be committed
- Excessive error logging
- The connector appearing healthy or `RUNNING`, although it is unable to deliver any data
- No clear terminal failure that monitoring or operations can react to

### Expected behaviour

The connector should distinguish between transient and non-recoverable Azure errors.

For a potentially transient error, the connector may retry the operation a bounded number of times, ideally with a configurable delay or backoff.

If the operation still cannot be completed after the configured retry limit, the connector task should transition to the `FAILED` state.

For clearly non-recoverable errors such as `ContainerNotFound`, the connector should either:

1. Fail the task immediately; or
2. Retry only a small, configurable number of times and then fail the task.

The task should not remain in the `RUNNING` state while endlessly retrying an operation that cannot succeed without an external configuration change.

### Steps to reproduce

1. Configure an Aiven Azure Blob Storage Sink Connector.
2. Configure a destination container that does not exist.
3. Start the connector.
4. Produce records to the configured Kafka topic.
5. Wait for the connector to flush data to Azure Blob Storage.
6. Observe that Azure returns `404 ContainerNotFound`.
7. Observe that Kafka Connect rewinds to the last committed offsets.
8. Observe that the task remains `RUNNING` and continues retrying the same operation.

### Actual result

The connector repeatedly executes the flush and offset commit cycle:

1. `AzureBlobSinkTask.flushFile()` attempts to write to Azure.
2. The Azure SDK returns `BlobStorageException: 404 ContainerNotFound`.
3. The connector wraps the error in `IOException` and `ConnectException`.
4. `WorkerSinkTask` logs that the offset commit failed.
5. Kafka Connect rewinds to the last committed offsets.
6. The task remains running and repeats the operation.

### Desired retry behaviour

It would be useful to provide connector-level settings similar to:

- Maximum number of write/flush retries
- Initial retry delay
- Maximum retry delay
- Retry backoff strategy
- Retryable Azure status codes or exception types
- Immediate failure for explicitly non-retryable Azure errors

After the retry limit is exhausted, the exception should cause the Kafka Connect task to enter the `FAILED` state.

### Relevant log excerpt

```text
[2026-07-10 15:48:12,575] ERROR
WorkerSinkTask{id=-0}
Offset commit failed, rewinding to last committed offsets
(org.apache.kafka.connect.runtime.WorkerSinkTask:450)

org.apache.kafka.connect.errors.ConnectException:
java.io.IOException:
com.azure.storage.blob.models.BlobStorageException:
Status code 404
Code: ContainerNotFound
Message: The specified container does not exist.

at io.aiven.kafka.connect.azure.sink.AzureBlobSinkTask.flushFile(
AzureBlobSinkTask.java:160
)
at io.aiven.kafka.connect.azure.sink.AzureBlobSinkTask.flush(
AzureBlobSinkTask.java:130
)
at org.apache.kafka.connect.sink.SinkTask.preCommit(
SinkTask.java:139
)
at org.apache.kafka.connect.runtime.WorkerSinkTask.commitOffsets(
WorkerSinkTask.java:445
)
```

If you agree with the proposed behaviour, I can investigate the retry handling and try to submit a pull request. Any guidance on the preferred implementation would be appreciated.

Guide de contribution

Ouvrir le guide de contribution

Évaluation

Cette issue n'a pas encore été évaluée.

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.