Amazon Data Firehose client

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

Getting started

Installation

Add the dependency to your module's build.gradle.kts:

dependencies {
    implementation("com.amplifyframework:aws-kinesis:LATEST_VERSION")
}

Initialize the client

import com.amplifyframework.firehose.AmplifyFirehoseClient

val firehose = AmplifyFirehoseClient(
    context = applicationContext,
    region = "us-east-1",
    credentialsProvider = credentialsProvider
)

Configuration options

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

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:

options = AmplifyFirehoseClientOptions {
    flushStrategy = FlushStrategy.None
}

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

val result = firehose.record(
    data = "Hello Firehose".toByteArray(),
    streamName = "my-delivery-stream"
)
when (result) {
    is Result.Success -> { /* recorded successfully */ }
    is Result.Failure -> { /* 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:

when (val result = firehose.flush()) {
    is Result.Success -> println("Flushed ${result.data.recordsFlushed} records")
    is Result.Failure -> println("Flush error: ${result.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:

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.

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

firehose.enable()
// Collection and auto-flush resume

Advanced

Escape hatch

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

val sdkClient = firehose.firehoseClient
// Use sdkClient for direct Firehose API calls

Error handling

All operations surface errors through a sealed exception hierarchy:

Operations return Result<T, AmplifyFirehoseException>:

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

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: