---
title: "Kinesis Data Streams client"
section: "frontend/analytics"
platforms: ["swift", "android", "flutter"]
gen: 2
last-updated: "2026-04-06T15:34:36.000Z"
url: "https://docs.amplify.aws/react/frontend/analytics/kinesis/"
---

`AmplifyKinesisClient` is a standalone client for streaming data to [Amazon Kinesis Data Streams](https://aws.amazon.com/kinesis/data-streams/). It provides:

- Local persistence for offline support
- Automatic retry for failed records
- Automatic batching (up to 500 records or 10 MB per request)
- Interval-based automatic flushing (default: every 30 seconds)
- Enable/disable toggle that silently drops new records while preserving cached ones

<Callout>

This is a standalone client, separate from the Amplify Analytics category plugin. It communicates directly with the Kinesis Data Streams API using `PutRecords`.

</Callout>

<Callout>

Before using this client, ensure your backend is configured with the required IAM permissions. See [Set up Kinesis Data Streams](/[platform]/build-a-backend/add-aws-services/analytics/kinesis/).

</Callout>

## Getting started

### Installation

<!-- Platform: android -->
Add the dependency to your module's `build.gradle.kts`:

```kotlin
dependencies {
    implementation("com.amplifyframework:aws-kinesis:LATEST_VERSION")
}
```
<!-- /Platform -->

<!-- Platform: swift -->
Add `AmplifyKinesisClient` 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.
<!-- /Platform -->

<!-- Platform: flutter -->
Add the dependency to your `pubspec.yaml`:

```yaml
dependencies:
  amplify_kinesis: ^2.11.0
```
<!-- /Platform -->

### Initialize the client

<!-- Platform: android -->
```kotlin
import com.amplifyframework.kinesis.AmplifyKinesisClient

val kinesis = AmplifyKinesisClient(
    context = applicationContext,
    region = "us-east-1",
    credentialsProvider = credentialsProvider
)
```
<!-- /Platform -->

<!-- Platform: swift -->
```swift
import AmplifyKinesisClient

let kinesis = try AmplifyKinesisClient(
    region: "us-east-1",
    credentialsProvider: credentialsProvider
)
```
<!-- /Platform -->

<!-- Platform: flutter -->
```dart
import 'package:amplify_kinesis/amplify_kinesis.dart';

final kinesis = await AmplifyKinesisClient.create(
    region: 'us-east-1',
    credentialsProvider: credentialsProvider,
);
```
<!-- /Platform -->

### Configuration options

You can customize the client behavior by passing an options object:

<!-- Platform: android -->
| 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 `KinesisClient`. |
<!-- /Platform -->

<!-- Platform: swift -->
| 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 `KinesisClientConfiguration`. |
<!-- /Platform -->

<!-- Platform: flutter -->
| 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. |
<!-- /Platform -->

<!-- Platform: android -->
```kotlin
import com.amplifyframework.kinesis.AmplifyKinesisClient
import com.amplifyframework.kinesis.AmplifyKinesisClientOptions
import com.amplifyframework.recordcache.FlushStrategy
import kotlin.time.Duration.Companion.seconds

val kinesis = AmplifyKinesisClient(
    context = applicationContext,
    region = "us-east-1",
    credentialsProvider = credentialsProvider,
    options = AmplifyKinesisClientOptions {
        cacheMaxBytes = 10L * 1024 * 1024  // 10 MB
        maxRetries = 3
        flushStrategy = FlushStrategy.Interval(60.seconds)
        configureClient {
            retryStrategy { maxAttempts = 10 }
        }
    }
)
```

To disable automatic flushing:

```kotlin
options = AmplifyKinesisClientOptions {
    flushStrategy = FlushStrategy.None
}
```
<!-- /Platform -->

<!-- Platform: swift -->
```swift
let kinesis = try AmplifyKinesisClient(
    region: "us-east-1",
    credentialsProvider: credentialsProvider,
    options: .init(
        cacheMaxBytes: 10 * 1_024 * 1_024,  // 10 MB
        maxRetries: 3,
        flushStrategy: .interval(60),
        configureClient: { config in
            // Customize the underlying KinesisClientConfiguration
        }
    )
)
```

To disable automatic flushing:

```swift
options: .init(flushStrategy: .none)
```
<!-- /Platform -->

<!-- Platform: flutter -->
```dart
final kinesis = await AmplifyKinesisClient.create(
    region: 'us-east-1',
    credentialsProvider: credentialsProvider,
    options: AmplifyKinesisClientOptions(
        cacheMaxBytes: 10 * 1024 * 1024,  // 10 MB
        maxRetries: 3,
        flushStrategy: FlushInterval(interval: Duration(seconds: 60)),
    ),
);
```

To disable automatic flushing:

```dart
options: AmplifyKinesisClientOptions(
    flushStrategy: FlushNone(),
),
```
<!-- /Platform -->

## Usage

### Record data

Use `record()` to persist data to the local cache. Records are sent to Kinesis during the next flush cycle (automatic or manual).

<!-- Platform: android -->
```kotlin
val result = kinesis.record(
    data = "Hello Kinesis".toByteArray(),
    partitionKey = "partition-1",
    streamName = "my-stream"
)
when (result) {
    is Result.Success -> { /* recorded successfully */ }
    is Result.Failure -> { /* handle error */ }
}
```
<!-- /Platform -->

<!-- Platform: swift -->
```swift
let result = try await kinesis.record(
    data: "Hello Kinesis".data(using: .utf8)!,
    partitionKey: "partition-1",
    streamName: "my-stream"
)
```
<!-- /Platform -->

<!-- Platform: flutter -->
```dart
final result = await kinesis.record(
    data: Uint8List.fromList(utf8.encode('Hello Kinesis')),
    partitionKey: 'partition-1',
    streamName: 'my-stream',
);
switch (result) {
    case Ok(): print('Recorded');
    case Error(:final error): print('Error: $error');
}
```
<!-- /Platform -->

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:

<!-- Platform: android -->
```kotlin
when (val result = kinesis.flush()) {
    is Result.Success -> println("Flushed ${result.data.recordsFlushed} records")
    is Result.Failure -> println("Flush error: ${result.error}")
}
```
<!-- /Platform -->

<!-- Platform: swift -->
```swift
let flushResult = try await kinesis.flush()
print("Flushed \(flushResult.recordsFlushed) records")
```
<!-- /Platform -->

<!-- Platform: flutter -->
```dart
switch (await kinesis.flush()) {
    case Ok(:final value):
        print('Flushed ${value.recordsFlushed} records');
    case Error(:final error):
        print('Flush failed: $error');
}
```
<!-- /Platform -->

Each flush sends at most one batch per stream (up to 500 records or 10 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:

<!-- Platform: android -->
```kotlin
kinesis.clearCache()
```
<!-- /Platform -->

<!-- Platform: swift -->
```swift
let cleared = try await kinesis.clearCache()
```
<!-- /Platform -->

<!-- Platform: flutter -->
```dart
await kinesis.clearCache();
```
<!-- /Platform -->

### 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.

<!-- Platform: android -->
```kotlin
kinesis.disable()
// Records are dropped, auto-flush paused

kinesis.enable()
// Collection and auto-flush resume
```
<!-- /Platform -->

<!-- Platform: swift -->
```swift
await kinesis.disable()
// Records are dropped, auto-flush paused

await kinesis.enable()
// Collection and auto-flush resume
```
<!-- /Platform -->

<!-- Platform: flutter -->
```dart
kinesis.disable();
// Records are dropped, auto-flush paused

kinesis.enable();
// Collection and auto-flush resume
```
<!-- /Platform -->

<!-- Platform: flutter -->
### Close the client

When you're done with the client, release its resources. The client cannot be reused after closing.

```dart
await kinesis.close();
```
<!-- /Platform -->

## Advanced

### Escape hatch

Access the underlying AWS SDK `KinesisClient` for operations not covered by this client's API:

<!-- Platform: android -->
```kotlin
val sdkClient = kinesis.kinesisClient
// Use sdkClient for direct Kinesis API calls
```
<!-- /Platform -->

<!-- Platform: swift -->
```swift
let sdkClient = kinesis.getKinesisClient()
// Use sdkClient for direct Kinesis API calls
```
<!-- /Platform -->

<!-- Platform: flutter -->
```dart
final sdkClient = kinesis.kinesisClient;
// Use sdkClient for direct Kinesis API calls
```
<!-- /Platform -->

### Error handling

All operations surface errors through a sealed exception hierarchy:

<!-- Platform: android -->
| Error type | Description |
|---|---|
| `AmplifyKinesisValidationException` | Record input validation failed (oversized record, invalid partition key). |
| `AmplifyKinesisLimitExceededException` | Local cache is full. Call `flush()` or `clearCache()` to free space. |
| `AmplifyKinesisStorageException` | Local database error. |
| `AmplifyKinesisUnknownException` | Unexpected or uncategorized error. |

Operations return `Result<T, AmplifyKinesisException>`:

```kotlin
when (val result = kinesis.record(...)) {
    is Result.Success -> { /* success */ }
    is Result.Failure -> when (result.error) {
        is AmplifyKinesisValidationException -> { /* invalid input */ }
        is AmplifyKinesisLimitExceededException -> { /* cache full */ }
        is AmplifyKinesisStorageException -> { /* database error */ }
        is AmplifyKinesisUnknownException -> { /* unexpected error */ }
    }
}
```
<!-- /Platform -->

<!-- Platform: swift -->
| Error type | Description |
|---|---|
| `KinesisError.validation` | Record input validation failed (oversized record, invalid partition key). |
| `KinesisError.cacheLimitExceeded` | Local cache is full. Call `flush()` or `clearCache()` to free space. |
| `KinesisError.cache` | Local database error. |
| `KinesisError.unknown` | Unexpected or uncategorized error. |

Operations throw `KinesisError`:

```swift
do {
    try await kinesis.record(
        data: payload,
        partitionKey: "key",
        streamName: "stream"
    )
} catch let error as KinesisError {
    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)")
    }
}
```
<!-- /Platform -->

<!-- Platform: flutter -->
| Error type | Description |
|---|---|
| `KinesisValidationException` | Record input validation failed (oversized record, invalid partition key). |
| `KinesisLimitExceededException` | Local cache is full. Call `flush()` or `clearCache()` to free space. |
| `KinesisStorageException` | Local database error. |
| `KinesisUnknownException` | Unexpected or uncategorized error. |
| `ClientClosedException` | The client was closed and cannot be used. |

Operations return `Result<T>` with `AmplifyKinesisException` subtypes:

```dart
switch (await kinesis.record(...)) {
    case Ok(): break;
    case Error(:final error):
        switch (error) {
            case KinesisValidationException(): // invalid input
            case KinesisLimitExceededException(): // cache full
            case KinesisStorageException(): // database error
            case KinesisUnknownException(): // unexpected error
            case ClientClosedException(): // client was closed
        }
}
```
<!-- /Platform -->

### Retry behavior

- All `PutRecords` error codes (`ProvisionedThroughputExceededException`, `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 Kinesis errors are logged and skipped per-stream, so other streams can still flush.
- Non-SDK errors (network failures, storage errors) abort the flush entirely.

### Kinesis service limits

The client enforces these limits before sending to the service:

| Limit | Value |
|---|---|
| Max records per `PutRecords` request | 500 |
| Max single record size | 10 MB |
| Max total payload per `PutRecords` request | 10 MB |
| Max partition key length | 256 characters |