---
title: "Amazon Data Firehose client"
section: "frontend/analytics"
platforms: ["android", "flutter", "swift"]
gen: 2
last-updated: "2026-05-11T17:44:11.000Z"
url: "https://docs.amplify.aws/react/frontend/analytics/firehose/"
---
`AmplifyFirehoseClient` is a standalone client for streaming data to [Amazon Data Firehose](https://aws.amazon.com/firehose/) delivery streams. It provides:
- Local persistence for offline support
- Automatic retry for failed records
- Automatic batching (up to 500 records or 4 MB per request)
- Interval-based automatic flushing (default: every 30 seconds)
- Enable/disable toggle that silently drops new records while preserving cached ones
This is a standalone client, separate from the Amplify Analytics category plugin. It communicates directly with the Firehose API using `PutRecordBatch`.
Before using this client, ensure your backend is configured with the required IAM permissions. See [Set up Amazon Data Firehose](/[platform]/build-a-backend/add-aws-services/analytics/firehose/).
## Getting started
### Installation
Add the dependency to your module's `build.gradle.kts`:
```kotlin
dependencies {
implementation("com.amplifyframework:aws-kinesis:LATEST_VERSION")
}
```
Add `AmplifyFirehoseClient` to your project using Swift Package Manager. In Xcode, go to **File > Add Package Dependencies** and enter the repository URL for the Amplify Swift SDK.
Add the dependency to your `pubspec.yaml`:
```yaml
dependencies:
amplify_firehose: ^2.11.0
```
### Initialize the client
```kotlin
import com.amplifyframework.firehose.AmplifyFirehoseClient
val firehose = AmplifyFirehoseClient(
context = applicationContext,
region = "us-east-1",
credentialsProvider = credentialsProvider
)
```
```swift
import AmplifyFirehoseClient
let firehose = try AmplifyFirehoseClient(
region: "us-east-1",
credentialsProvider: credentialsProvider
)
```
```dart
import 'package:amplify_firehose/amplify_firehose.dart';
final firehose = await AmplifyFirehoseClient.create(
region: 'us-east-1',
credentialsProvider: credentialsProvider,
);
```
The Flutter client automatically resolves the local storage path using `path_provider`. On web, it uses IndexedDB with an in-memory fallback.
### Configuration options
You can customize the client behavior by passing an options object:
| Option | Default | Description |
|---|---|---|
| `cacheMaxBytes` | 5 MB | Maximum size of the local record cache in bytes. |
| `maxRetries` | 5 | Maximum retry attempts per record before it is discarded. |
| `flushStrategy` | `FlushStrategy.Interval(30.seconds)` | Automatic flush interval. Use `FlushStrategy.None` for manual-only flushing. |
| `configureClient` | `null` | Escape hatch to customize the underlying AWS SDK `FirehoseClient`. |
| Option | Default | Description |
|---|---|---|
| `cacheMaxBytes` | 5 MB | Maximum size of the local record cache in bytes. |
| `maxRetries` | 5 | Maximum retry attempts per record before it is discarded. |
| `flushStrategy` | `.interval(30)` | Automatic flush interval in seconds. Use `.none` for manual-only flushing. |
| `configureClient` | `nil` | Closure to customize the underlying `FirehoseClientConfiguration`. |
```kotlin
import com.amplifyframework.firehose.AmplifyFirehoseClient
import com.amplifyframework.firehose.AmplifyFirehoseClientOptions
import com.amplifyframework.recordcache.FlushStrategy
import kotlin.time.Duration.Companion.seconds
val firehose = AmplifyFirehoseClient(
context = applicationContext,
region = "us-east-1",
credentialsProvider = credentialsProvider,
options = AmplifyFirehoseClientOptions {
cacheMaxBytes = 10L * 1024 * 1024 // 10 MB
maxRetries = 5
flushStrategy = FlushStrategy.Interval(30.seconds)
configureClient {
retryStrategy { maxAttempts = 10 }
}
}
)
```
To disable automatic flushing:
```kotlin
options = AmplifyFirehoseClientOptions {
flushStrategy = FlushStrategy.None
}
```
```swift
let firehose = try AmplifyFirehoseClient(
region: "us-east-1",
credentialsProvider: credentialsProvider,
options: .init(
cacheMaxBytes: 10 * 1_024 * 1_024, // 10 MB
maxRetries: 5,
flushStrategy: .interval(30),
configureClient: { config in
// Customize the underlying FirehoseClientConfiguration
}
)
)
```
To disable automatic flushing:
```swift
options: .init(flushStrategy: .none)
```
| Option | Default | Description |
|---|---|---|
| `cacheMaxBytes` | 5 MB | Maximum size of the local record cache in bytes. |
| `maxRetries` | 5 | Maximum retry attempts per record before it is discarded. |
| `flushStrategy` | `FlushInterval(interval: Duration(seconds: 30))` | Automatic flush interval. Use `FlushNone()` for manual-only flushing. |
```dart
import 'package:amplify_firehose/amplify_firehose.dart';
final firehose = await AmplifyFirehoseClient.create(
region: 'us-east-1',
credentialsProvider: credentialsProvider,
options: const AmplifyFirehoseClientOptions(
cacheMaxBytes: 10 * 1024 * 1024, // 10 MB
maxRetries: 5,
flushStrategy: FlushInterval(
interval: Duration(seconds: 30),
),
),
);
```
To disable automatic flushing:
```dart
options: const AmplifyFirehoseClientOptions(
flushStrategy: FlushNone(),
)
```
## Usage
### Record data
Use `record()` to persist data to the local cache. Records are sent to Firehose during the next flush cycle (automatic or manual).
```kotlin
val result = firehose.record(
data = "Hello Firehose".toByteArray(),
streamName = "my-delivery-stream"
)
when (result) {
is Result.Success -> { /* recorded successfully */ }
is Result.Failure -> { /* handle error */ }
}
```
```swift
let result = try await firehose.record(
data: "Hello Firehose".data(using: .utf8)!,
streamName: "my-delivery-stream"
)
```
```dart
import 'dart:convert';
import 'dart:typed_data';
final result = await firehose.record(
data: Uint8List.fromList(utf8.encode('Hello Firehose')),
streamName: 'my-delivery-stream',
);
switch (result) {
case Ok():
// recorded successfully
case Error(:final error):
// handle error
}
```
Records submitted while the client is disabled are silently dropped.
### Flush records
The client automatically flushes cached records at the configured interval (default: 30 seconds). You can also trigger a manual flush:
```kotlin
when (val result = firehose.flush()) {
is Result.Success -> println("Flushed ${result.data.recordsFlushed} records")
is Result.Failure -> println("Flush error: ${result.error}")
}
```
```swift
let flushResult = try await firehose.flush()
print("Flushed \(flushResult.recordsFlushed) records")
```
```dart
switch (await firehose.flush()) {
case Ok(:final value):
print('Flushed ${value.recordsFlushed} records');
case Error(:final error):
print('Flush error: $error');
}
```
Each flush sends at most one batch per stream (up to 500 records or 4 MB). Remaining records are picked up in subsequent flush cycles. If a flush is already in progress, the call returns immediately with `flushInProgress: true`.
Manual flushes work even when the client is disabled, allowing you to drain cached records without re-enabling collection.
### Clear cache
Delete all cached records from local storage:
```kotlin
firehose.clearCache()
```
```swift
let cleared = try await firehose.clearCache()
```
```dart
final result = await firehose.clearCache();
```
### Enable and disable
You can toggle record collection and automatic flushing at runtime. When disabled, new records are silently dropped but already-cached records remain in storage.
```kotlin
firehose.disable()
// Records are dropped, auto-flush paused
firehose.enable()
// Collection and auto-flush resume
```
```swift
await firehose.disable()
// Records are dropped, auto-flush paused
await firehose.enable()
// Collection and auto-flush resume
```
```dart
firehose.disable();
// Records are dropped, auto-flush paused
firehose.enable();
// Collection and auto-flush resume
```
### Close the client
When you're done with the client, close it to release resources:
```dart
await firehose.close();
```
After closing, all operations return an error. Create a new client instance if needed.
## Advanced
### Escape hatch
Access the underlying AWS SDK `FirehoseClient` for operations not covered by this client's API:
```kotlin
val sdkClient = firehose.firehoseClient
// Use sdkClient for direct Firehose API calls
```
```swift
let sdkClient = firehose.getFirehoseClient()
// Use sdkClient for direct Firehose API calls
```
```dart
final sdkClient = firehose.firehoseClient;
// Use sdkClient for direct Firehose API calls
```
### Error handling
All operations surface errors through a sealed exception hierarchy:
| Error type | Description |
|---|---|
| `AmplifyFirehoseValidationException` | Record input validation failed (oversized record). |
| `AmplifyFirehoseLimitExceededException` | Local cache is full. Call `flush()` or `clearCache()` to free space. |
| `AmplifyFirehoseStorageException` | Local database error. |
| `AmplifyFirehoseUnknownException` | Unexpected or uncategorized error. |
Operations return `Result`:
```kotlin
when (val result = firehose.record(...)) {
is Result.Success -> { /* success */ }
is Result.Failure -> when (result.error) {
is AmplifyFirehoseValidationException -> { /* invalid input */ }
is AmplifyFirehoseLimitExceededException -> { /* cache full */ }
is AmplifyFirehoseStorageException -> { /* database error */ }
is AmplifyFirehoseUnknownException -> { /* unexpected error */ }
}
}
```
| Error type | Description |
|---|---|
| `FirehoseError.validation` | Record input validation failed (oversized record). |
| `FirehoseError.cacheLimitExceeded` | Local cache is full. Call `flush()` or `clearCache()` to free space. |
| `FirehoseError.cache` | Local database error. |
| `FirehoseError.unknown` | Unexpected or uncategorized error. |
Operations throw `FirehoseError`:
```swift
do {
try await firehose.record(
data: payload,
streamName: "stream"
)
} catch let error as FirehoseError {
switch error {
case .validation(let desc, _, _):
print("Validation error: \(desc)")
case .cacheLimitExceeded:
print("Cache full")
case .cache(let desc, _, _):
print("Storage error: \(desc)")
case .unknown(let desc, _, _):
print("Unknown error: \(desc)")
}
}
```
| Error type | Description |
|---|---|
| `FirehoseValidationException` | Record input validation failed (oversized record). |
| `FirehoseLimitExceededException` | Local cache is full. Call `flush()` or `clearCache()` to free space. |
| `FirehoseStorageException` | Local database error. |
| `FirehoseClientClosedException` | Operation attempted on a closed client. |
| `FirehoseUnknownException` | Unexpected or uncategorized error. |
Operations return `Result` which can be pattern-matched:
```dart
final result = await firehose.record(
data: payload,
streamName: 'stream',
);
switch (result) {
case Ok():
// success
case Error(:final error):
switch (error) {
case FirehoseValidationException():
print('Validation error: ${error.message}');
case FirehoseLimitExceededException():
print('Cache full');
case FirehoseStorageException():
print('Storage error: ${error.message}');
case FirehoseClientClosedException():
print('Client is closed');
case FirehoseUnknownException():
print('Unknown error: ${error.message}');
}
}
```
### Retry behavior
- All `PutRecordBatch` error codes (`ServiceUnavailableException`, `InternalFailure`) are treated as retryable.
- Each failed record's retry count is incremented after each attempt.
- Records exceeding `maxRetries` (default: 5) are permanently deleted from the cache.
- SDK-level Firehose errors are logged and skipped per-stream, so other streams can still flush.
- Non-SDK errors (network failures, storage errors) abort the flush entirely.
### Firehose service limits
The client enforces these limits before sending to the service:
| Limit | Value |
|---|---|
| Max records per `PutRecordBatch` request | 500 |
| Max single record size | 1,000 KiB |
| Max total payload per `PutRecordBatch` request | 4 MB |