kafka

FastForwardPolicy

FastForwardPolicy(grace_budgets=FAST_FORWARD_GRACE_BUDGETS, grace_floor=FAST_FORWARD_GRACE_FLOOR, clock=time.monotonic)

Decide when a live stream should give up on its backlog and skip ahead to current data.

Every stream has a latency budget: how old its data may grow and still be useful (its channels' max_latency, plus a stride). Older data is dropped as late anyway, so a consumer stuck further behind real time than its budget is doing pure waste -- decoding blocks nothing will use, with no way to catch up. Better to skip to fresh data and take one bounded gap.

Feed each delivered block's age to :meth:observe. Once a stream has stayed over budget for its whole grace period (grace_budgets times its budget, at least grace_floor seconds), it answers :attr:LagAction.FORWARD; the caller performs the seek and reports it via :meth:forwarded. The first observation with every stream back within budget answers :attr:LagAction.RECOVERED, once, for logging. Lag is tracked per stream, so a healthy stream cannot hide a stuck one.

Used by :class:KafkaReader for direct-Kafka clients, and by the arrakis-server for the readers it pools for its own clients.

Source code in arrakis/kafka.py

def __init__(
    self,
    grace_budgets: float = FAST_FORWARD_GRACE_BUDGETS,
    grace_floor: float = FAST_FORWARD_GRACE_FLOOR,
    clock: Callable[[], float] = time.monotonic,
):
    self.grace_budgets = grace_budgets
    self.grace_floor = grace_floor
    self._clock = clock
    # when each stream's data first went over budget (monotonic
    # seconds); streams within budget have no entry
    self._behind_since: dict[str, float] = {}
    # a skip happened and recovery has not yet been observed
    self._forwarded = False

forwarded

forwarded()

Record that the consumer skipped ahead: every stream starts fresh, and the next fully within-budget observation reports recovery.

Source code in arrakis/kafka.py

def forwarded(self) -> None:
    """Record that the consumer skipped ahead: every stream starts
    fresh, and the next fully within-budget observation reports
    recovery."""
    self._behind_since.clear()
    self._forwarded = True

grace

grace(budget_ns)

The grace period for the given latency budget, in seconds.

Source code in arrakis/kafka.py

def grace(self, budget_ns: int) -> float:
    """The grace period for the given latency budget, in seconds."""
    return max(self.grace_floor, self.grace_budgets * budget_ns / Time.SECONDS)

observe

observe(stream_id, lag_ns, budget_ns)

Record one delivered block's age against its stream's budget.

Source code in arrakis/kafka.py

def observe(self, stream_id: str, lag_ns: int, budget_ns: int) -> LagAction:
    """Record one delivered block's age against its stream's budget."""
    if self.grace_budgets <= 0 or budget_ns <= 0:
        return LagAction.OK
    if lag_ns <= budget_ns:
        self._behind_since.pop(stream_id, None)
        if self._forwarded and not self._behind_since:
            self._forwarded = False
            return LagAction.RECOVERED
        return LagAction.OK
    now = self._clock()
    since = self._behind_since.setdefault(stream_id, now)
    if now - since < self.grace(budget_ns):
        return LagAction.OK
    return LagAction.FORWARD

KafkaReader

KafkaReader(url, metadata, start=None, *, fast_forward=False)

Bases: StreamReader

A connection object to read data from Kafka.

Parameters:

Name	Type	Description	Default
url	str	URL of Kafka broker to connect to.	required
metadata	dict[str, Channel]]	Dictionary of channel metadata for request.	required
start	int | None	GPS start time of stream in nanoseconds, defaults to "now".	None
fast_forward	bool	When True, a live stream that has fallen further behind real time than its latency budget allows -- for long enough that it clearly cannot catch up -- skips ahead to current data via :meth:fast_forward, trading the unusable backlog for a gap. The skip and the recovery are logged. Default False: only the caller knows whether skipping is appropriate (bounded readers catch up on purpose; the arrakis-server runs this policy itself), so the user-facing entry points arm it for live requests.	False

Source code in arrakis/kafka.py

def __init__(
    self,
    url: str,
    metadata: dict[str, Channel],
    start: int | None = None,
    *,
    fast_forward: bool = False,
):
    self.url = url
    self.metadata = metadata
    self.start = start
    self._policy = FastForwardPolicy() if fast_forward else None

    # track stream -> index -> channel
    self._name_lookup: dict[str, dict[int, Channel]] = defaultdict(defaultdict)
    # filters for partitions
    self._partition_filter_id_sets = defaultdict(set)

    for channel in self.metadata.values():
        partition_id = channel.partition_id
        assert partition_id is not None
        partition_index = channel.partition_index
        assert partition_index is not None
        stream_id = self._id(partition_id)
        self._name_lookup[stream_id][partition_index] = channel
        self._partition_filter_id_sets[partition_id].add(partition_index)

    # per-stream latency budget, mirroring the muxer's queue
    # timeouts: the largest channel max_latency plus a stride of
    # headroom, or 0 (no budget, never fast-forwarded) when no
    # channel declares one
    self._budgets: dict[str, int] = {}
    for stream_id, index_chan_map in self._name_lookup.items():
        channels = list(index_chan_map.values())
        latencies = [ch.max_latency for ch in channels if ch.max_latency]
        stride = math.lcm(*(ch.stride or 0 for ch in channels))
        self._budgets[stream_id] = max(latencies) + stride if latencies else 0

    # pre-compute Arrow arrays for filtering to avoid repeated
    # array creation
    self._partition_filter_id_arrays = {
        partition_id: pyarrow.array(channel_id, type=pyarrow.int32())
        for partition_id, channel_id in self._partition_filter_id_sets.items()
    }

    # Optimize IPC stream reader creation/destruction overhead
    self._ipc_options = pyarrow.ipc.IpcReadOptions(use_threads=False)

    # create Kafka consumer
    queue_kbytes = min(
        self.QUEUE_KBYTES_MAX,
        max(
            self.QUEUE_KBYTES_MIN,
            self.QUEUE_KBYTES_PER_PARTITION * len(self._partition_filter_id_sets),
        ),
    )
    consumer_settings = {
        "bootstrap.servers": url,
        "group.id": generate_groupid(),
        "message.max.bytes": 10_000_000,  # 10 MB
        "queued.min.messages": 100,
        "queued.max.messages.kbytes": queue_kbytes,
        "fetch.queue.backoff.ms": self.FETCH_QUEUE_BACKOFF_MS,
        "enable.auto.commit": False,
    }
    # librdkafka internal debugging, e.g. "cgrp,topic,metadata"
    # (add "fetch" only for short sessions: it logs per request).
    # Diagnostic switch — not intended as a standing setting.
    kafka_debug = os.getenv("ARRAKIS_KAFKA_DEBUG")
    if kafka_debug:
        consumer_settings["debug"] = kafka_debug
    # route librdkafka logs through the arrakis logger so they
    # respect the application's logging configuration
    self._consumer = Consumer(consumer_settings, logger=logger)
    logger.debug("Kafka consumer: %s", consumer_settings)
    # topics requested but not present on the broker; their
    # channels are served as gaps until the topics appear
    self._missing_topics: set[str] = set()
    self._next_missing_check = 0.0
    # transient consume-error episode: count since the last report
    # and when that report was logged (None outside an episode)
    self._consume_errors = 0
    self._consume_error_logged_at: float | None = None
    self._topic_to_partition: dict[str, str] = {}
    for partition_id in self._partition_filter_id_sets:
        channel = next(
            ch for ch in self.metadata.values() if ch.partition_id == partition_id
        )
        t = topic_name(partition_id, channel.replay_id)
        self._topic_to_partition[t] = partition_id
    self._topics = list(self._topic_to_partition)
    logger.debug("kafka topics: %s", self._topics)
    # watermark cache for exhausted(): (topic, partition) -> high
    self._watermarks: dict[tuple[str, int], int] = {}
    self._watermarks_at = float("-inf")

enter

enter()

Assign the consumer to the requested topics.

Partitions are assigned directly rather than subscribed via a consumer group: every reader is a single-member group with no committed offsets, so group membership buys nothing and costs the coordinator dependency (slow or failing joins starve the reader entirely) plus rebalance churn when any requested topic does not exist. Missing topics are excluded — their channels are served as gaps by the muxer — and re-checked periodically so data starts flowing if a topic appears.

Source code in arrakis/kafka.py

def enter(self) -> None:
    """Assign the consumer to the requested topics.

    Partitions are assigned directly rather than subscribed via a
    consumer group: every reader is a single-member group with no
    committed offsets, so group membership buys nothing and costs
    the coordinator dependency (slow or failing joins starve the
    reader entirely) plus rebalance churn when any requested topic
    does not exist.  Missing topics are excluded — their channels
    are served as gaps by the muxer — and re-checked periodically
    so data starts flowing if a topic appears.
    """
    logger.debug("initiating kafka assignments...")
    now = time_as_ns(gpsnow())
    existing = self._existing_topics()
    missing = set(self._topics) - set(existing)
    if missing:
        logger.warning(
            "%d requested topic(s) do not exist on the broker and "
            "will be served as gaps until they appear: %s",
            len(missing),
            sorted(missing),
        )
    self._missing_topics = missing
    self._next_missing_check = time.monotonic() + self.MISSING_TOPIC_RECHECK

    # if start time is specified, point the assignment at the
    # data requested
    if self.start and self.start <= now:
        if not existing:
            return
        # convert to UNIX time in ms
        offset_time = int(gps2unix(self.start // Time.SECONDS) * 1000)
        # get offsets corresponding to times
        partitions = [
            TopicPartition(topic, partition=0, offset=offset_time)
            for topic in existing
        ]
        partitions = self._offsets_for_times(partitions)
        # An unresolved offset (no message at/after the requested
        # time yet — e.g. a fresh topic whose data is still being
        # written) must not decay to auto.offset.reset (latest),
        # which would silently skip data: start from the beginning
        # instead; the muxer drops blocks older than requested.
        for tp in partitions:
            if tp.offset < 0:
                logger.debug(
                    "no offset at requested time for %s; "
                    "starting from the beginning",
                    tp.topic,
                )
                tp.offset = OFFSET_BEGINNING
        self._consumer.assign(partitions)

    # FIXME: start times in the future are being handled as if a
    # start time was not specified
    elif existing:
        self._consumer.assign(
            [
                TopicPartition(topic, partition=0, offset=OFFSET_END)
                for topic in existing
            ]
        )

exhausted

exhausted()

Whether everything currently in Kafka has been consumed.

True when every assigned partition's position has reached its high watermark. Topics missing from the broker are not assigned and therefore count as exhausted — they have nothing to consume. Watermarks are fetched from the broker at most every WATERMARK_REFRESH seconds; a False result may be spurious (stale watermark, broker hiccup), so callers should simply re-check later, while True is reliable for the data Kafka held at the last refresh.

Bounded readers use this as their terminal state: once the log is exhausted, data absent from the requested range is absent for good, and waiting for it will not end.

Source code in arrakis/kafka.py

def exhausted(self) -> bool:
    """Whether everything currently in Kafka has been consumed.

    True when every assigned partition's position has reached its
    high watermark.  Topics missing from the broker are not
    assigned and therefore count as exhausted — they have nothing
    to consume.  Watermarks are fetched from the broker at most
    every WATERMARK_REFRESH seconds; a False result may be
    spurious (stale watermark, broker hiccup), so callers should
    simply re-check later, while True is reliable for the data
    Kafka held at the last refresh.

    Bounded readers use this as their terminal state: once the
    log is exhausted, data absent from the requested range is
    absent for good, and waiting for it will not end.
    """
    assignment = self._consumer.assignment()
    if not assignment:
        return True
    keys = {(tp.topic, tp.partition) for tp in assignment}
    if (
        time.monotonic() - self._watermarks_at > self.WATERMARK_REFRESH
        or keys != set(self._watermarks)
    ):
        watermarks: dict[tuple[str, int], int] = {}
        try:
            for tp in assignment:
                low, high = self._consumer.get_watermark_offsets(
                    tp, timeout=self.OFFSETS_TIMEOUT
                )
                # a fully purged partition's offsets do not restart
                # at 0; low == high means nothing readable
                watermarks[tp.topic, tp.partition] = high if high > low else 0
        except KafkaException as exc:
            logger.debug("watermark refresh failed: %s", exc)
            return False
        self._watermarks = watermarks
        self._watermarks_at = time.monotonic()

    for tp in self._consumer.position(assignment):
        high = self._watermarks[tp.topic, tp.partition]
        if high == 0:
            continue
        if tp.offset < 0 or tp.offset < high:
            # nothing consumed yet, or still behind the head
            return False
    return True

fast_forward

fast_forward(time_ns=None)

Skip the assigned partitions ahead, dropping the backlog.

Seeks the assigned partitions to the first message at or after time_ns (GPS nanoseconds), or to the end of the partition when time_ns is None or no message that recent exists yet. A partition that has already consumed past its target stays where it is. Messages between the current position and the target are never delivered. Intended for a consumer that has fallen behind real time further than it can catch up: reading the backlog only produces data too old to serve.

Returns:

Type	Description
int	The number of partitions moved (0 when nothing is assigned).

Source code in arrakis/kafka.py

def fast_forward(self, time_ns: int | None = None) -> int:
    """Skip the assigned partitions ahead, dropping the backlog.

    Seeks the assigned partitions to the first message at or after
    ``time_ns`` (GPS nanoseconds), or to the end of the partition
    when ``time_ns`` is None or no message that recent exists yet.
    A partition that has already consumed past its target stays
    where it is.  Messages between the current position and the
    target are never delivered.  Intended for a consumer that has
    fallen behind real time further than it can catch up: reading
    the backlog only produces data too old to serve.

    Returns
    -------
    int
        The number of partitions moved (0 when nothing is assigned).
    """
    assignment = self._consumer.assignment()
    if not assignment:
        return 0
    positions = {
        (tp.topic, tp.partition): tp.offset
        for tp in self._consumer.position(assignment)
    }
    targets = [
        TopicPartition(tp.topic, tp.partition, OFFSET_END) for tp in assignment
    ]
    if time_ns is not None:
        offset_time = int(gps2unix(time_ns / Time.SECONDS) * 1000)
        requested = [
            TopicPartition(tp.topic, tp.partition, offset_time) for tp in assignment
        ]
        try:
            resolved = self._offsets_for_times(requested)
        except RuntimeError as exc:
            logger.warning(
                "offset lookup for fast-forward failed, seeking to end: %s", exc
            )
        else:
            by_partition = {(tp.topic, tp.partition): tp.offset for tp in resolved}
            for tp in targets:
                partition = (tp.topic, tp.partition)
                offset = by_partition.get(partition, -1)
                if offset >= 0:
                    # never rewind a partition that is past the target
                    tp.offset = max(
                        offset, positions.get(partition, OFFSET_INVALID)
                    )
    moved = sum(
        1
        for tp in targets
        if tp.offset != positions.get((tp.topic, tp.partition), OFFSET_INVALID)
    )
    # re-assigning drops the fetch queues, so messages already
    # buffered below the target are discarded along with the rest
    self._consumer.assign(targets)
    self._watermarks_at = float("-inf")
    return moved

LagAction

Bases: Enum

What a :meth:FastForwardPolicy.observe observation calls for.

generate_groupid

generate_groupid()

Generate a random Kafka group ID.

Source code in arrakis/kafka.py

def generate_groupid() -> str:
    """Generate a random Kafka group ID."""
    return random_alphanum(16)

random_alphanum

random_alphanum(n)

Generate a random alpha-numeric sequence of N characters.

Source code in arrakis/kafka.py

def random_alphanum(n: int) -> str:
    """Generate a random alpha-numeric sequence of N characters."""
    alphanum = string.ascii_uppercase + string.digits
    return "".join(random.SystemRandom().choice(alphanum) for _ in range(n))