Troubleshooting Kinesis: Common Issues and Solutions

Troubleshooting Kinesis: Common Issues and Solutions

Amazon Kinesis is a powerful platform for collecting, processing, and analyzing streaming data in real time. Whether you are using Kinesis Data Streams, Kinesis Data Firehose, or Kinesis Data Analytics, building streaming pipelines introduces a unique set of operational challenges. Latency spikes, throttling errors, consumer lag, and data loss are all problems that engineers encounter when running Kinesis at scale. This tutorial walks through the most common Kinesis issues, explains why they happen, and provides practical solutions with code examples you can apply immediately.

Why Troubleshooting Kinesis Matters

Streaming systems differ from batch systems because they operate continuously. A small misconfiguration can compound over time, leading to backpressure, dropped records, or silent data loss. Because Kinesis is often the backbone of real-time analytics, fraud detection, or event-driven architectures, any disruption can have immediate business impact. Understanding how to diagnose and resolve Kinesis issues quickly is essential for maintaining reliable data pipelines.

1. ProvisionedThroughputExceededException

This is the most common error developers encounter with Kinesis Data Streams. It occurs when your application exceeds the shard capacity limits. Each shard supports 1 MB per second of write throughput and 2 MB per second of read throughput, with up to 1,000 records per second for writes.

Common Causes

• Too many records being sent to a single shard
• Records exceeding the 1 MB maximum size
• Insufficient shard count for your workload
• Hot shard distribution caused by poor partition key selection

Solution: Exponential Backoff and Resharding

The immediate fix is to implement retry logic with exponential backoff. The long-term fix is to increase shard count or improve your partition key strategy to distribute traffic evenly.

import boto3
import time
import random
from botocore.exceptions import ClientError

kinesis = boto3.client('kinesis')

def put_record_with_retry(stream_name, data, partition_key, max_retries=5):
    for attempt in range(max_retries):
        try:
            response = kinesis.put_record(
                StreamName=stream_name,
                Data=data,
                PartitionKey=partition_key
            )
            return response
        except ClientError as e:
            error_code = e.response['Error']['Code']
            if error_code == 'ProvisionedThroughputExceededException':
                sleep_time = min(2 ** attempt + random.random(), 30)
                print(f"Throttled. Retrying in {sleep_time:.2f}s (attempt {attempt + 1})")
                time.sleep(sleep_time)
            else:
                raise
    raise Exception(f"Failed after {max_retries} retries")

To increase shard capacity, you can use the UpdateShardCount API:

response = kinesis.update_shard_count(
    StreamName='my-stream',
    TargetShardCount=10,
    ScalingType='UNIFORM_SCALING'
)

2. Consumer Lag and Read Throughput Issues

Consumer lag happens when your consumers cannot process records as fast as producers write them. The iterator age metric (GetRecords.IteratorAgeMilliseconds) is the key indicator. If this value keeps growing, your consumer is falling behind.

Diagnosing with CloudWatch Metrics

• GetRecords.IteratorAgeMilliseconds — should stay below 10,000 ms (10 seconds)
• GetRecords.Bytes — actual bytes read per second
• GetRecords.Records — number of records read per second
• PutRecords.Success — ratio of successful writes

Solution: Use Enhanced Fan-Out Consumers

Standard consumers share the 2 MB per second read throughput per shard. Enhanced fan-out consumers get a dedicated 2 MB per second channel each, which dramatically improves read performance.

import boto3

kinesis = boto3.client('kinesis')

# Register a consumer with enhanced fan-out
consumer = kinesis.register_stream_consumer(
    StreamARN='arn:aws:kinesis:us-east-1:123456789012:stream/my-stream',
    ConsumerName='my-enhanced-consumer'
)

consumer_arn = consumer['Consumer']['ConsumerARN']

# Subscribe to the shard using SubscribeToShard
import asyncio
from botocore.config import Config

async def consume_shard(shard_id):
    client = boto3.client(
        'kinesis',
        config=Config(region_name='us-east-1')
    )
    
    response = client.describe_stream(
        StreamName='my-stream',
        ShardId=shard_id
    )
    
    iterator = client.get_shard_iterator(
        StreamName='my-stream',
        ShardId=shard_id,
        ShardIteratorType='LATEST',
        ConsumerARN=consumer_arn
    )['ShardIterator']
    
    while True:
        records = client.subscribe_to_shard(
            ConsumerARN=consumer_arn,
            ShardId=shard_id,
            StartingPosition={'Type': 'AFTER_SEQUENCE_NUMBER', 'SequenceNumber': 'last-seq'}
        )
        # Process records here
        for event in records['EventStream']:
            print(event)

3. Partial Failures with PutRecords Batch API

The PutRecords API accepts up to 500 records or 5 MB per request. However, individual records within a batch can fail while others succeed. Ignoring these partial failures leads to silent data loss.

Solution: Inspect Individual Record Responses

import boto3
import json

kinesis = boto3.client('kinesis')

def put_records_batch(stream_name, records):
    formatted = [
        {'Data': json.dumps(r), 'PartitionKey': str(r['id'])}
        for r in records
    ]
    
    response = kinesis.put_records(
        StreamName=stream_name,
        Records=formatted
    )
    
    failed_records = response.get('FailedRecordCount', 0)
    
    if failed_records > 0:
        retry_batch = []
        for idx, record_result in enumerate(response['Records']):
            if 'ErrorCode' in record_result:
                print(f"Record {idx} failed: {record_result['ErrorCode']} - {record_result.get('ErrorMessage', '')}")
                retry_batch.append(records[idx])
        
        if retry_batch:
            print(f"Retrying {len(retry_batch)} failed records...")
            return put_records_batch(stream_name, retry_batch)
    
    print(f"Successfully wrote {len(records)} records")
    return response

# Example usage
records = [{'id': i, 'event': f'event_{i}'} for i in range(100)]
put_records_batch('my-stream', records)

4. Kinesis Data Firehose Delivery Failures

Firehose automatically delivers data to destinations like S3, Redshift, Elasticsearch, or HTTP endpoints. When the destination is unavailable or slow, Firehose buffers data. If buffering limits are exceeded, records can be dropped.

Common Firehose Issues

• Destination service throttling or downtime
• Buffer size or interval misconfiguration
• Lambda transformation failures returning errors
• Permission issues preventing delivery to S3 or other targets

Solution: Tune Buffer Settings and Monitor Delivery

import boto3

firehose = boto3.client('firehose')

# Update delivery stream buffer settings
response = firehose.update_destination(
    DeliveryStreamName='my-delivery-stream',
    CurrentDeliveryStreamVersionId='1',
    DestinationId='destinationId-000000000001',
    S3DestinationUpdate={
        'BufferingHints': {
            'SizeInMBs': 64,        # Buffer up to 64 MB
            'IntervalInSeconds': 300  # Or flush every 5 minutes
        },
        'CompressionFormat': 'GZIP',
        'EncryptionConfiguration': {
            'NoEncryptionConfig': 'NoEncryption'
        },
        'CloudWatchLoggingOptions': {
            'Enabled': True,
            'LogGroupName': '/aws/kinesisfirehose/my-delivery-stream',
            'LogStreamName': 'S3Delivery'
        }
    }
)

For Lambda transformation failures, ensure your Lambda function returns properly formatted responses:

import base64
import json

def lambda_handler(event, context):
    output_records = []
    
    for record in event['records']:
        try:
            payload = base64.b64decode(record['data']).decode('utf-8')
            data = json.loads(payload)
            
            # Transform the data
            transformed = {
                'timestamp': data.get('timestamp'),
                'value': data.get('value'),
                'processed': True
            }
            
            output_records.append({
                'recordId': record['recordId'],
                'result': 'Ok',
                'data': base64.b64encode(json.dumps(transformed).encode('utf-8')).decode('utf-8')
            })
        except Exception as e:
            # Mark as processing failed - Firehose will retry
            output_records.append({
                'recordId': record['recordId'],
                'result': 'ProcessingFailed',
                'data': record['data']
            })
    
    return {'records': output_records}

5. Shard Iterator Expiration

Shard iterators expire after 5 minutes. If your consumer takes too long between GetRecords calls, the iterator becomes invalid and you receive a ResourceNotFoundException.

Solution: Cache Sequence Numbers and Recreate Iterators

import boto3
import time

kinesis = boto3.client('kinesis')
last_sequence_number = None

def get_shard_iterator(stream_name, shard_id, sequence_number=None):
    if sequence_number:
        return kinesis.get_shard_iterator(
            StreamName=stream_name,
            ShardId=shard_id,
            ShardIteratorType='AFTER_SEQUENCE_NUMBER',
            StartingSequenceNumber=sequence_number
        )['ShardIterator']
    else:
        return kinesis.get_shard_iterator(
            StreamName=stream_name,
            ShardId=shard_id,
            ShardIteratorType='LATEST'
        )['ShardIterator']

def consume_stream(stream_name, shard_id):
    global last_sequence_number
    shard_iterator = get_shard_iterator(stream_name, shard_id, last_sequence_number)
    
    while True:
        try:
            response = kinesis.get_records(ShardIterator=shard_iterator, Limit=10000)
            
            for record in response['Records']:
                process_record(record)
                last_sequence_number = record['SequenceNumber']
            
            shard_iterator = response['NextShardIterator']
            
            if not response['Records']:
                time.sleep(0.5)
                
        except kinesis.exceptions.ResourceNotFoundException:
            print("Shard iterator expired. Recreating...")
            shard_iterator = get_shard_iterator(stream_name, shard_id, last_sequence_number)

def process_record(record):
    print(f"Processing: {record['Data']}")

6. Hot Shards and Uneven Data Distribution

If your partition key strategy concentrates too much data on a few shards, those shards become bottlenecks while others remain underutilized. This is called a hot shard problem.

Solution: Use a Hash-Based Partition Key Strategy

import hashlib
import random

def generate_partition_key(record):
    """
    Generate a well-distributed partition key.
    Avoid using high-cardinality fields that cluster data.
    """
    # Option 1: Use a hash of a unique identifier
    base = record.get('user_id', str(random.randint(0, 999999)))
    return hashlib.md5(base.encode()).hexdigest()

# Option 2: Explicit shard routing with a composite key
def generate_composite_key(record):
    region = record.get('region', 'unknown')
    random_suffix = random.randint(0, 99)
    return f"{region}:{random_suffix}"

7. Permission and IAM Issues

Many Kinesis problems stem from incorrect IAM permissions. Producers need kinesis:PutRecord and kinesis:PutRecords, while consumers need kinesis:GetRecords, kinesis:GetShardIterator, and kinesis:DescribeStream.

Solution: Apply Least Privilege IAM Policies

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kinesis:PutRecord",
                "kinesis:PutRecords"
            ],
            "Resource": "arn:aws:kinesis:us-east-1:123456789012:stream/my-stream"
        },
        {
            "Effect": "Allow",
            "Action": [
                "kinesis:GetRecords",
                "kinesis:GetShardIterator",
                "kinesis:DescribeStream",
                "kinesis:DescribeStreamSummary",
                "kinesis:ListShards"
            ],
            "Resource": "arn:aws:kinesis:us-east-1:123456789012:stream/my-stream"
        }
    ]
}

8. Monitoring and Alerting Best Practices

Proactive monitoring prevents most Kinesis issues from escalating. Set up CloudWatch alarms for critical metrics before they become production incidents.

Recommended CloudWatch Alarms

import boto3

cloudwatch = boto3.client('cloudwatch')

def create_kinesis_alarms(stream_name):
    # Alarm for high iterator age (consumer lag)
    cloudwatch.put_metric_alarm(
        AlarmName=f'{stream_name}-HighIteratorAge',
        ComparisonOperator='GreaterThanThreshold',
        EvaluationPeriods=3,
        MetricName='GetRecords.IteratorAgeMilliseconds',
        Namespace='AWS/Kinesis',
        Period=60,
        Statistic='Average',
        Threshold=30000,  # 30 seconds
        AlarmDescription='Consumer is falling behind',
        Dimensions=[{'Name': 'StreamName', 'Value': stream_name}],
        AlarmActions=['arn:aws:sns:us-east-1:123456789012:kinesis-alerts']
    )
    
    # Alarm for write throttling
    cloudwatch.put_metric_alarm(
        AlarmName=f'{stream_name}-WriteThrottling',
        ComparisonOperator='GreaterThanThreshold',
        EvaluationPeriods=2,
        MetricName='WriteProvisionedThroughputExceeded',
        Namespace='AWS/Kinesis',
        Period=60,
        Statistic='Sum',
        Threshold=0,
        AlarmDescription='Write throttling detected',
        Dimensions=[{'Name': 'StreamName', 'Value': stream_name}],
        AlarmActions=['arn:aws:sns:us-east-1:123456789012:kinesis-alerts']
    )
    
    # Alarm for low get records success rate
    cloudwatch.put_metric_alarm(
        AlarmName=f'{stream_name}-LowGetSuccess',
        ComparisonOperator='LessThanThreshold',
        EvaluationPeriods=3,
        MetricName='GetRecords.Success',
        Namespace='AWS/Kinesis',
        Period=60,
        Statistic='Average',
        Threshold=0.95,
        AlarmDescription='GetRecords success rate below 95%',
        Dimensions=[{'Name': 'StreamName', 'Value': stream_name}],
        AlarmActions=['arn:aws:sns:us-east-1:123456789012:kinesis-alerts']
    )

create_kinesis_alarms('my-stream')

9. Using the Kinesis Producer Library (KPL) for High Throughput

For high-throughput producers, the KPL handles batching, retries, and aggregation automatically. It aggregates multiple records into a single Kinesis record, increasing effective throughput beyond the 1,000 records per second limit.

from amazon_kclpy import kpl
import threading
import time

# Configure KPL
config = {
    'AggregationEnabled': 'true',
    'AggregationMaxCount': '4294967295',
    'AggregationMaxSize': '51200',
    'CollectionMaxCount': '500',
    'CollectionMaxSize': '5242880',
    'Region': 'us-east-1',
    'StreamName': 'my-stream'
}

producer = kpl.KinesisProducer(config)

def send_records():
    for i in range(10000):
        producer.add_user_record(
            'my-stream',
            f'partition-key-{i % 100}',
            f'{{"id": {i}, "data": "sample"}}'.encode('utf-8')
        )
    # Flush and wait for completion
    producer.flush()
    while producer.outstanding_records_count() > 0:
        time.sleep(0.1)

send_records()

10. Debugging with Kinesis Client Library (KCL) Checkpointing

The KCL manages checkpointing automatically, but improper checkpoint handling can cause reprocessing or data gaps. Always checkpoint after processing, not before.

from amazon_kclpy import kcl

class RecordProcessor(kcl.RecordProcessorBase):
    def __init__(self):
        self._shard_id = None
        self._checkpoint_counter = 0
    
    def initialize(self, shard_id):
        self._shard_id = shard_id
        print(f"Initialized processor for shard: {shard_id}")
    
    def process_records(self, records, checkpointer):
        for record in records:
            try:
                # Process the record
                data = record.get('data')
                sequence_number = record.get('sequenceNumber')
                print(f"Processing record: {sequence_number}")
                
                # Checkpoint every 100 records
                self._checkpoint_counter += 1
                if self._checkpoint_counter >= 100:
                    checkpointer.checkpoint(sequence_number)
                    self._checkpoint_counter = 0
                    
            except Exception as e:
                print(f"Error processing record: {e}")
                # Do not checkpoint on failure - will retry
    
    def shutdown(self, checkpointer, reason):
        if reason == 'TERMINATE':
            checkpointer.checkpoint()
        print(f"Shutting down processor for shard: {self._shard_id}")

Best Practices Summary

• Always implement retry logic with exponential backoff and jitter for producers
• Use enhanced fan-out consumers for latency-sensitive applications
• Monitor IteratorAgeMilliseconds and set alarms at 30 seconds
• Handle partial failures in PutRecords by inspecting individual record responses
• Choose partition keys that distribute data evenly across shards
• Use the KPL for high-throughput producers to leverage aggregation
• Checkpoint after processing, never before, to avoid data gaps
• Enable CloudWatch logging for Firehose delivery streams
• Apply least privilege IAM policies scoped to specific streams
• Test resharding during off-peak hours and monitor split/merge operations

Conclusion

Troubleshooting Amazon Kinesis requires a combination of proactive monitoring, proper configuration, and robust error handling in your application code. By understanding the root causes of common issues like throttling, consumer lag, partial batch failures, and shard iterator expiration, you can build resilient streaming pipelines that handle real-world conditions gracefully. The key is to treat every potential failure point as an expected scenario: implement retries, inspect individual record results, monitor CloudWatch metrics continuously, and design your partition key strategy for even distribution. With these practices in place, your Kinesis-based architectures will be well-equipped to handle scale, recover from transient failures, and deliver data reliably to downstream consumers.