Leaving Kinesis: Shards Are Not Kafka Partitions

---
# v1, not v1beta2. Strimzi 1.0.0's CHANGELOG: "Remove the v1beta2 API (and
# v1alpha1 and v1beta2 for KafkaTopic and KafkaUser) from the CRDs and fully
# move to the v1 API" -- and "Use the v1 API as the stored API version".
# https://raw.githubusercontent.com/strimzi/strimzi-kafka-operator/1.0.0/CHANGELOG.md
apiVersion: kafka.strimzi.io/v1
kind: KafkaTopic
metadata:
  name: payments-events
  labels:
    strimzi.io/cluster: payments-kafka
spec:
  # NOT 12. A shard count is purchased throughput; a partition count is both
  # consumer parallelism AND the modulus every key is hashed against. Choose it
  # from key cardinality and target consumer parallelism, then freeze it for the
  # whole cutover: raising it later re-maps existing keys and moves no data.
  partitions: 24
  replicas: 3
  config:
    # The one direct translation of a Kinesis retention window: 7 days -- and
    # the steady-state value, NOT the cutover value. Kafka deletes a rolled
    # segment once that segment's NEWEST record timestamp is older than
    # retention.ms, and the backfill copier below deliberately stamps records
    # from ApproximateArrivalTimestamp, up to the full Kinesis retention window
    # old. Leave this at 7 days during a 7-day backfill and the oldest copied
    # segments are deletable the moment they roll, while the copier is still
    # running. So: set retention.ms: -1 ("If set to -1, no time limit is
    # applied") -- or a value well beyond retention span + cutover duration --
    # BEFORE the backfill starts, and restore the value below only after the
    # reader flip AND after dual-write stops.
    retention.ms: 604800000
    # Enforced per partition, so the topic ceiling is this times 24 = 1.5 TiB.
    retention.bytes: 68719476736
    # delete, not compact: tiered storage has no support for compacted topics,
    # so choosing compaction here would close that door. Tiering is NOT enabled
    # by this manifest -- that needs remote.storage.enable plus local.retention.*
    # here, remote.log.storage.system.enable on the broker, and a
    # RemoteStorageManager implementation, which Kafka does not ship.
    cleanup.policy: delete
    min.insync.replicas: 2
    # Availability is never silently preferred over durability on this topic.
    unclean.leader.election.enable: false
---
apiVersion: kafka.strimzi.io/v1
kind: KafkaUser
metadata:
  name: payments-reconciler
  labels:
    strimzi.io/cluster: payments-kafka
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: payments-events
        operations: [Describe, Read]
        host: "*"
      - resource:
          type: group
          name: payments-reconciler
        operations: [Read]
        host: "*"

# Batching is now a latency decision, not a billing one. Nothing here is
# chunked at 25 KB, so size batch.size against the broker and the payload.
bootstrap.servers={{ KAFKA_BOOTSTRAP }}
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="{{ KAFKA_USERNAME }}" password="{{ KAFKA_PASSWORD }}";

# linger.ms default changed from 0 to 5 in Apache Kafka 4.0: the producer waits
# 5 ms or until the batch reaches batch.size, whichever comes first. Raise it
# only against a measured latency budget, not out of habit from PUT Payload Units.
linger.ms=5
batch.size=65536

# Compression is applied to full batches, so batching efficacy and compression
# ratio move together. zstd for a throughput-shaped stream.
compression.type=zstd

# Durability half of the min.insync.replicas=2 contract on the topic.
acks=all
# Enabled by default when no conflicting configuration is set; stated here so a
# future edit cannot silently disable it. Requires acks=all, retries > 0 and
# max.in.flight.requests.per.connection <= 5.
enable.idempotence=true
max.in.flight.requests.per.connection=5

# Not only a producer setting: with an authorizer on, InitProducerId has to be
# authorized too. On a current broker that is satisfied by IdempotentWrite on
# the CLUSTER resource OR by Write on any topic (KIP-679), so a principal that
# already holds the topic Write above will not fail for lack of the cluster
# grant. Granting IdempotentWrite on CLUSTER explicitly -- kafka-acls has an
# --idempotent convenience flag for it -- is still worth doing: it states the
# intent where an auditor reads it instead of leaving it implied by a topic ACL
# that a later narrowing could remove.

#!/usr/bin/env bash
set -euo pipefail

STREAM="${KINESIS_STREAM:?set the stream name}"
STATE_DIR="${STATE_DIR:-./cutover-state}"
mkdir -p "$STATE_DIR"

# T is pinned ONCE, here, and never re-derived. Everything downstream reads it
# from this file: the backfill's stop point and the live tail's start point are
# the same instant by construction.
if [ ! -f "$STATE_DIR/boundary-T" ]; then
  date -u +%s > "$STATE_DIR/boundary-T"
fi
T=$(cat "$STATE_DIR/boundary-T")

summary=$(aws kinesis describe-stream-summary --stream-name "$STREAM" \
  --query 'StreamDescriptionSummary.[RetentionPeriodHours,RecordDistributionStrategy,OpenShardCount]' \
  --output text)
read -r retention_hours strategy open_shards <<< "$summary"

# RecordDistributionStrategy is only present for on-demand streams, so the CLI
# renders it as None on a provisioned one. That absence is not a third strategy
# and not a failed call: a provisioned stream routes by partition key and
# rejects AUTO outright, so resolve it to USER_PARTITION_KEY and say so.
if [ -z "$strategy" ] || [ "$strategy" = "None" ]; then
  strategy="USER_PARTITION_KEY (provisioned; field absent)"
fi

# A T older than the trim horizon does not fail: GetShardIterator returns a
# TRIM_HORIZON iterator instead, which replays the whole retained shard and
# looks exactly like success. Refuse to start rather than discover that later.
now=$(date -u +%s)
trim_horizon=$(( now - retention_hours * 3600 ))
if [ "$T" -le "$trim_horizon" ]; then
  echo "ABORT: boundary T has aged past the trim horizon. Re-pin it." >&2
  exit 1
fi

# What the backfill will actually copy: TRIM_HORIZON through T -- the entire
# retained span, not a recent slice of it. Recorded so the reconciliation gate
# can check its own window against it instead of taking one on trust. Pinned
# once, like T: a second run of this script would otherwise walk the recorded
# start forward while the window's start stays put, which can only make the
# coverage check more permissive.
if [ ! -f "$STATE_DIR/backfill-start" ]; then
  echo "$trim_horizon" > "$STATE_DIR/backfill-start"
fi

# The reconciliation window is [WINDOW_START, T). Pinned once, here, for the
# same reason T is: a window whose start moves cannot be reconciled twice. It
# DEFAULTS TO THE RETENTION SPAN because that is what the backfill copies -- a
# narrower window is a spot check, and the gate downstream will refuse to clear
# the reader flip on one.
WINDOW_HOURS="${WINDOW_HOURS:-$retention_hours}"
if [ ! -f "$STATE_DIR/window-start" ]; then
  echo $(( T - WINDOW_HOURS * 3600 )) > "$STATE_DIR/window-start"
fi

echo "stream=$STREAM strategy=$strategy shards=$open_shards T=$T"
echo "backfill span starts $(cat "$STATE_DIR/backfill-start")"
echo "reconciliation window starts $(cat "$STATE_DIR/window-start")"

# TWO SHARD LISTS, and conflating them is the quietest way to lose history.
#
# The BACKFILL list is the UNFILTERED one. FROM_TRIM_HORIZON is the ListShards
# default and "includes all the shards within the retention period of the data
# stream (trim to tip)" -- closed parents left behind by past splits and merges
# included. Those closed parents are where every pre-split record lives, so a
# backfill enumerated any other way silently copies no pre-split history at all
# on a stream that has resharded inside the retention window. Read each parent
# to exhaustion before its children, or per-key order breaks at the split.
aws kinesis list-shards --stream-name "$STREAM" \
  --shard-filter "Type=FROM_TRIM_HORIZON" \
  --query 'Shards[].[ShardId,ParentShardId]' --output text \
  > "$STATE_DIR/backfill-shards"
echo "backfill reads $(wc -l < "$STATE_DIR/backfill-shards") shards, closed parents included"

# The TAIL list is the FILTERED one, and the filter is not optional hygiene.
# AT_TIMESTAMP "includes all shards whose start timestamp is less than or equal
# to the given timestamp and end timestamp is greater than or equal to the
# given timestamp or still open" -- so at T it drops every parent closed before
# T. That is exactly right here and exactly wrong above. Without it a copier
# would tail shards that can never receive another record, because
# GetShardIterator on a closed shard returns a valid iterator for its last
# sequence number rather than an error.
aws kinesis list-shards --stream-name "$STREAM" \
  --shard-filter "Type=AT_TIMESTAMP,Timestamp=$T" \
  --query 'Shards[].ShardId' --output text | tr '\t' '\n' | while read -r shard; do
  # One AT_TIMESTAMP call per shard against the single pinned value. Every read
  # after this one follows the NextShardIterator that GetRecords returns, and
  # must be used inside its five-minute validity. These files are a record of
  # what was pinned, not a resumable handle: an iterator expires five minutes
  # after it is returned, so a backfill that restarts later calls
  # GetShardIterator again at AFTER_SEQUENCE_NUMBER of the last record copied.
  aws kinesis get-shard-iterator \
    --stream-name "$STREAM" \
    --shard-id "$shard" \
    --shard-iterator-type AT_TIMESTAMP \
    --timestamp "$T" \
    --query 'ShardIterator' --output text > "$STATE_DIR/tail-iterator.$shard"
  echo "pinned tail iterator for $shard"
done

"""Dual-write shim for a Kinesis-to-Kafka cutover.

Kinesis stays authoritative until the reader flip; Kafka runs as a shadow. The
Kinesis partition key is reused verbatim as the Kafka record key ONLY once the
stream's record distribution strategy is confirmed as USER_PARTITION_KEY. On
AUTO, Kinesis ignored the partition key and published no ordering guarantee, so
copying it would carry an assignment the service never made.

DescribeStreamSummary returns RecordDistributionStrategy only for streams in
on-demand capacity mode, where it is AUTO or USER_PARTITION_KEY. A provisioned
stream omits the field entirely; provisioned behaviour is USER_PARTITION_KEY,
which is why that is the default below.
"""

from __future__ import annotations

import json
import os
from dataclasses import dataclass


class CutoverRefused(RuntimeError):
    """A precondition of the cutover does not hold. Do not start."""


@dataclass(frozen=True)
class Cutover:
    stream_name: str
    topic: str
    frozen_partitions: int
    ack_timeout_s: float


def assert_user_partition_key(kinesis, stream_name: str) -> str:
    """Refuse to reuse the partition key on a stream that ignored it."""
    summary = kinesis.describe_stream_summary(StreamName=stream_name)
    # An absent field means PROVISIONED, not AUTO: the key was honoured. The
    # default below is therefore the correct fail direction, not a guess.
    strategy = summary["StreamDescriptionSummary"].get(
        "RecordDistributionStrategy", "USER_PARTITION_KEY"
    )
    if strategy != "USER_PARTITION_KEY":
        raise CutoverRefused(
            f"{stream_name} is on {strategy}: Kinesis ignored the partition key and "
            "ExplicitHashKey and published no ordering guarantee. There is no per-key "
            "assignment to carry. Choose the Kafka key from the payload deliberately."
        )
    return strategy


def assert_partition_count(producer, cutover: Cutover) -> None:
    """Key reuse preserves assignment only while the modulus holds."""
    live = len(producer.partitions_for(cutover.topic) or ())
    if live != cutover.frozen_partitions:
        raise CutoverRefused(
            f"{cutover.topic} now has {live} partitions, frozen at "
            f"{cutover.frozen_partitions}. hash(key) % partitions has moved, so "
            "existing keys may resolve to a different partition than their history."
        )


def dual_write(kinesis, producer, cutover: Cutover, key: str, payload: dict) -> None:
    body = json.dumps(payload, separators=(",", ":")).encode()
    # Kinesis first and synchronously: it is the system of record until the flip,
    # so a Kafka-only write is a divergence with no rollback target.
    kinesis.put_record(StreamName=cutover.stream_name, PartitionKey=key, Data=body)
    # send() only buffers the record and hands back a FutureRecordMetadata.
    # Discarding that future is how a shadow gap hides behind a successful
    # PutRecord: flush() blocks until the future is done -- including done with
    # an error -- and raises only its own KafkaTimeoutError, so the caller sees
    # a clean return on a produce that failed. Wait on the future instead, and
    # let the broker's error reach the caller that just wrote to Kinesis.
    future = producer.send(cutover.topic, key=key.encode(), value=body)
    future.get(timeout=cutover.ack_timeout_s)


def build_cutover() -> Cutover:
    return Cutover(
        stream_name=os.environ["KINESIS_STREAM"],
        topic=os.environ["KAFKA_TOPIC"],
        frozen_partitions=int(os.environ["KAFKA_FROZEN_PARTITIONS"]),
        ack_timeout_s=float(os.environ.get("KAFKA_ACK_TIMEOUT_S", "10")),
    )


def main() -> None:
    import boto3
    from kafka import KafkaProducer

    cutover = build_cutover()
    kinesis = boto3.client("kinesis")
    producer = KafkaProducer(
        bootstrap_servers=os.environ["KAFKA_BOOTSTRAP"].split(","),
        acks="all",
        enable_idempotence=True,
        # max_in_flight_requests_per_connection is deliberately NOT set here.
        # Do not carry the "<= 5" from the properties file above into this
        # client: that bound is the Java producer's. kafka-python 2.2.x raises
        # KafkaConfigurationError on any user-provided value other than 1 under
        # idempotence, and overrides its own default to 1 when you leave it
        # alone; 3.0.x aligned with the Java bound. Unset is correct on both.
    )
    # Both checks run at startup, before a single record is written.
    assert_user_partition_key(kinesis, cutover.stream_name)
    assert_partition_count(producer, cutover)
    dual_write(kinesis, producer, cutover, "acct-7781", {"event": "ping"})
    # Drain on the way out. NOT the acknowledgement check -- dual_write already
    # waited on its own future, because flush() completing is not proof that
    # any individual produce succeeded.
    producer.flush()


if __name__ == "__main__":
    main()

#!/usr/bin/env bash
set -euo pipefail

STREAM="${KINESIS_STREAM:?set the stream name}"
TOPIC="${KAFKA_TOPIC:?set the topic name}"
STATE_DIR="${STATE_DIR:-./cutover-state}"
# Both written once by the pin script. Reading them rather than re-deriving
# them is the whole point: a window whose edges move cannot be reconciled.
# UNITS: these are Unix SECONDS -- date -u +%s wrote them.
WINDOW_START=$(cat "$STATE_DIR/window-start")
WINDOW_END=$(cat "$STATE_DIR/boundary-T")
# A Kafka record timestamp is an integer of MILLISECONDS since epoch, so the
# Kafka-side counter is handed the millisecond pair and the Kinesis-side
# counter the seconds. Hand seconds to a counter comparing milliseconds and
# nothing falls in range: kafka_count is 0, delta equals kinesis_count, and the
# gate aborts a correct migration. It fails closed, so no data is lost -- but a
# gate that fires on good runs is how teams learn to ignore the gate.
WINDOW_START_MS=$(( WINDOW_START * 1000 ))
WINDOW_END_MS=$(( WINDOW_END * 1000 ))
# The span the backfill copied: TRIM_HORIZON through T. Counting a narrower
# window and then clearing the flip is the trap this file exists to close.
BACKFILL_START=$(cat "$STATE_DIR/backfill-start")

# PREREQUISITE, checked before this script is worth running: the target topic's
# retention.ms was set to -1 (or far beyond retention span + cutover duration)
# BEFORE the backfill started, and is still there. Kafka deletes a rolled
# segment once that segment's newest record timestamp is older than
# retention.ms, and the copier writes backdated timestamps -- so on a
# retention-span retention.ms the oldest copied segments are deletable as soon
# as they roll. This gate would then be counting history that is already being
# trimmed. Restore the intended retention.ms after the reader flip and after
# dual-write stops, never before.
#
# NOT SUPPLIED HERE. Each counter must print a single integer: the number of
# records whose own-side timestamp falls in the pinned window, in that side's
# own units -- [WINDOW_START, WINDOW_END) seconds for Kinesis,
# [WINDOW_START_MS, WINDOW_END_MS) milliseconds for Kafka.
#   Kinesis side: count on ApproximateArrivalTimestamp as GetRecords returns it
#     -- the server-side arrival clock. Never a timestamp inside the payload:
#     that one is the producer's and has no relation to when Kinesis stored it.
#   Kafka side: count on each record's timestamp over the same window. Whose
#     clock that is depends on message.timestamp.type -- leave it on the
#     default CreateTime, which keeps the value the producer supplied, and
#     have the backfill copier set that value from each record's own
#     ApproximateArrivalTimestamp. A Kafka record timestamp is an integer of
#     milliseconds since epoch; Kinesis documents the same instant as a Unix
#     epoch value with millisecond precision, written seconds-first
#     (1459799926.480), so the copier multiplies by 1000.
#     Do NOT set message.timestamp.type=LogAppendTime on this topic: the
#     broker then overwrites the producer's timestamp with its own local time
#     at append, and every backfill append happens AFTER T -- so a complete,
#     correct copy of [TRIM_HORIZON, T) lands outside the window this gate
#     counts and the gate fails on a good migration.
# Deliberately NOT a sum of end offsets: an offset is not a record count, a
# partition is not a shard, and that diff passes on a wrong backfill.
COUNT_KINESIS="${COUNT_KINESIS_CMD:?path to your Kinesis window counter}"
COUNT_KAFKA="${COUNT_KAFKA_CMD:?path to your Kafka window counter}"

# AWS guarantees neither accuracy nor monotonicity on ApproximateArrivalTimestamp,
# so an exact equality here would be a check that fails on correct backfills and
# teaches the team to ignore it. Declare the tolerance instead. 200 ppm is an
# EXAMPLE operator input, not a measured figure: set it from the duplicate and
# loss budget you are willing to defend, and write down why.
TOLERANCE_PPM="${TOLERANCE_PPM:-200}"
# Absolute floor, because a ppm of a small window truncates to zero and turns
# the gate back into the exact equality this exists to avoid.
TOLERANCE_FLOOR="${TOLERANCE_FLOOR:-5}"

kinesis_count=$("$COUNT_KINESIS" --stream "$STREAM" \
  --from "$WINDOW_START" --to "$WINDOW_END")
kafka_count=$("$COUNT_KAFKA" --topic "$TOPIC" \
  --from "$WINDOW_START_MS" --to "$WINDOW_END_MS")

delta=$(( kafka_count - kinesis_count ))
[ "$delta" -lt 0 ] && delta=$(( -delta ))
# Round UP: integer division of count * ppm / 1000000 is 0 below 5,000 records.
scaled=$(( (kinesis_count * TOLERANCE_PPM + 999999) / 1000000 ))
allowed=$(( scaled > TOLERANCE_FLOOR ? scaled : TOLERANCE_FLOOR ))

printf 'kinesis=%d kafka=%d delta=%d allowed=%d\n' \
  "$kinesis_count" "$kafka_count" "$delta" "$allowed"

if [ "$delta" -gt "$allowed" ]; then
  echo "ABORT: reconciliation outside tolerance. Do not flip readers." >&2
  exit 1
fi

# Inside tolerance is only half the question. The other half is WHAT was
# counted. A window narrower than the backfill certifies a slice and says
# nothing about the history behind it, so it does not clear the flip -- it is a
# spot check, and it exits non-zero so a pipeline cannot read it as a pass.
printf 'within tolerance over [%d, %d)\n' "$WINDOW_START" "$WINDOW_END"
if [ "$WINDOW_START" -gt "$BACKFILL_START" ]; then
  printf 'PARTIAL: backfill copied from %d; this window starts %ds later.\n' \
    "$BACKFILL_START" "$(( WINDOW_START - BACKFILL_START ))" >&2
  echo "A spot check does NOT clear the reader flip. Re-run with" >&2
  echo "WINDOW_HOURS set to the retention span, on a fresh state dir." >&2
  exit 2
fi
echo "window covers the full TRIM_HORIZON-to-T backfill: safe to flip readers"