Kinesis Data Streams client

AmplifyKinesisClient is a standalone client for streaming data to Amazon 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

Getting started

Installation

Add the dependency to your pubspec.yaml:

dependencies:
  amplify_kinesis: ^2.11.0

Initialize the client

import 'package:amplify_kinesis/amplify_kinesis.dart';

final kinesis = await AmplifyKinesisClient.create(
    region: 'us-east-1',
    credentialsProvider: credentialsProvider,
);

Configuration options

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

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:

options: AmplifyKinesisClientOptions(
    flushStrategy: FlushNone(),
),

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

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');
}

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:

switch (await kinesis.flush()) {
    case Ok(:final value):
        print('Flushed ${value.recordsFlushed} records');
    case Error(:final error):
        print('Flush failed: $error');
}

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:

await kinesis.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.

kinesis.disable();
// Records are dropped, auto-flush paused

kinesis.enable();
// Collection and auto-flush resume

Close the client

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

await kinesis.close();

Advanced

Escape hatch

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

final sdkClient = kinesis.kinesisClient;
// Use sdkClient for direct Kinesis API calls

Error handling

All operations surface errors through a sealed exception hierarchy:

Operations return Result<T> with AmplifyKinesisException subtypes:

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
        }
}

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: