// 生のKinesis APIを使った場合（大変な実装例）
val kinesisClient = KinesisClient.builder().build()

// 1. ストリーム情報を取得してシャード一覧を取得
val streamDescription = kinesisClient.describeStream(
    DescribeStreamRequest.builder().streamName("user-events").build()
)

streamDescription.streamDescription().shards().forEach { shard ->
    // 2. 各シャードのイテレーターを取得
    val shardIterator = kinesisClient.getShardIterator(
        GetShardIteratorRequest.builder()
            .streamName("user-events")
            .shardId(shard.shardId())
            .shardIteratorType(ShardIteratorType.LATEST)
            .build()
    ).shardIterator()
    
    // 3. 継続的にレコードを取得・処理
    var currentIterator = shardIterator
    while (currentIterator != null) {
        val response = kinesisClient.getRecords(
            GetRecordsRequest.builder()
                .shardIterator(currentIterator)
                .build()
        )
        
        // 4. レコード処理
        response.records().forEach { record ->
            try {
                processRecord(record)
                
                // 5. 処理位置を手動で記録（DynamoDBなどに）
                saveCheckpoint(shard.shardId(), record.sequenceNumber())
            } catch (e: Exception) {
                // エラーハンドリングも自前実装
                handleProcessingError(record, e)
            }
        }
        
        currentIterator = response.nextShardIterator()
        
        // 6. シャードの分割・結合をチェック
        if (isShardClosed(shard)) {
            handleShardClosure(shard)
        }
        
        // 7. 他のインスタンスとの協調処理
        if (shouldGiveUpShard(shard)) {
            transferShardToOtherInstance(shard)
        }
        
        Thread.sleep(1000) // ポーリング間隔
    }
}

// KCLを使った場合
class MyRecordProcessor : ShardRecordProcessor {
    override fun processRecords(processRecordsInput: ProcessRecordsInput) {
        // ビジネスロジックだけに集中できる
        processRecordsInput.records().forEach { record ->
            when (record.eventType) {
                "USER_LOGIN" -> handleUserLogin(record)
                "PURCHASE" -> processPurchase(record)
                "PAGE_VIEW" -> trackPageView(record)
                else -> logger.warn("Unknown event: ${record.eventType}")
            }
        }
        
        // チェックポイントも1行で完了
        processRecordsInput.checkpointer().checkpoint()
    }
}

[ユーザーイベント] → [Kinesis Stream A] → [Application A]
[システムログ]     → [Kinesis Stream B] → [Application B]  
[メトリクス]       → [Kinesis Stream C] → [Application C]