DE Mastery GitHub

Apache Kafka · KFK-14

Operating & troubleshooting a cluster

From your world. an incident runbook: observe the symptom, diagnose the cause, apply the fix, verify health restored

← All exercises

Six broken-cluster scenarios — diagnose and fix

There are six classic Kafka incidents. For each incident, implement a diagnose_X (read the state and return a small structured dict) and a fix_X (apply the smallest change). The grader follows these steps:

  1. It creates each fault deterministically.
  2. It asserts that your diagnosis names the fault.
  3. It applies your fix.
  4. It checks again that health is restored.

Edit troubleshoot.py.

The six incidents

# diagnose / fix Fault the grader creates Health check after fix
1 diagnose_under_replicated / fix_under_replicated kills a broker on an RF=3 topic URP back to 0 (broker restarted)
2 diagnose_min_isr / fix_min_isr min.insync.replicas=3 with a broker down acks=all produce succeeds (floor lowered)
3 diagnose_retention / fix_retention retention.ms=1000 (far too low) retention ≥ the floor
4 diagnose_lag / fix_lag a group committed behind the end lag drains to 0 (offsets reset)
5 diagnose_advertised_listeners / fix_advertised_listeners advertised.listeners = 0.0.0.0/wrong host rewritten to the reachable host
6 diagnose_throttle / fix_throttle replication throttle below a floor raised to ≥ the floor

Diagnosis return shapes

Each diagnose_X returns a dict with an "issue" key. The value is the problem name when the fault is present. The value is "healthy" when the fault is not present.

  • diagnose_under_replicated(admin, topic) → {"issue": "under_replicated"|"healthy", "under_replicated_partitions": [...]}
  • diagnose_min_isr(admin, topic) → {"issue": "min_isr_too_high"|"healthy", "min_insync_replicas", "smallest_isr"}
  • diagnose_retention(admin, topic, floor_ms) → {"issue": "retention_too_low"|"healthy", "retention_ms"}
  • diagnose_lag(bootstrap, topic, group) → {"issue": "lagging"|"healthy", "total_lag"}
  • diagnose_advertised_listeners(props, reachable_host) → {"issue": "wrong_advertised_listeners"|"healthy", "unreachable": [...]}
  • diagnose_throttle(throttle_config, floor_bytes_per_s) → {"issue": "throttle_too_low"|"healthy", "rates": [...]}

Notes

  • The admin object is a confluent_kafka AdminClient. The cluster object is the de_testkit cluster (run_tool, start_broker, bootstrap_servers). Fixes use the bundled CLI tools (kafka-configs.sh, kafka-consumer-groups.sh) or cluster.start_broker.
  • Scenarios 5 and 6 are pure. They work on a config/props dict and need no broker. So you can always grade part of the suite on a bare machine.
  • retention.ms = -1 means infinite retention and is healthy. Only a non-negative value below the floor is a fault.

Starter code

troubleshoot.py
"""Starter — KFK-14. Implement the six diagnose/fix pairs. The stubs below
currently fail the tests (gate G1)."""

from __future__ import annotations


# 1. Under-replicated partitions
def diagnose_under_replicated(admin, topic, timeout=15.0):
    """Return {'issue': 'under_replicated'|'healthy', 'under_replicated_partitions': [...]}."""
    # TODO
    return {"issue": "healthy", "under_replicated_partitions": []}


def fix_under_replicated(cluster, dead_broker_id):
    """Bring the offline broker back (cluster.start_broker)."""
    # TODO
    pass


# 2. min.insync.replicas too high for the available ISR
def diagnose_min_isr(admin, topic, timeout=15.0):
    """Return {'issue': 'min_isr_too_high'|'healthy', 'min_insync_replicas', 'smallest_isr'}."""
    # TODO
    return {"issue": "healthy", "min_insync_replicas": 1, "smallest_isr": 1}


def fix_min_isr(cluster, topic, new_min):
    """Lower topic min.insync.replicas via kafka-configs.sh --alter."""
    # TODO
    pass


# 3. Retention below a floor
def diagnose_retention(admin, topic, floor_ms, timeout=15.0):
    """Return {'issue': 'retention_too_low'|'healthy', 'retention_ms'}."""
    # TODO
    return {"issue": "healthy", "retention_ms": -1}


def fix_retention(cluster, topic, new_ms):
    """Set topic retention.ms via kafka-configs.sh --alter."""
    # TODO
    pass


# 4. Stuck / lagging consumer group
def diagnose_lag(consumer_bootstrap, topic, group_id, timeout=15.0):
    """Return {'issue': 'lagging'|'healthy', 'total_lag'}."""
    # TODO
    return {"issue": "healthy", "total_lag": 0}


def fix_lag(cluster, topic, group_id, timeout=30.0):
    """Reset the group to the latest offset (kafka-consumer-groups.sh)."""
    # TODO
    pass


# 5. Wrong advertised.listeners (pure)
def diagnose_advertised_listeners(props, reachable_host):
    """Return {'issue': 'wrong_advertised_listeners'|'healthy', 'unreachable': [...]}."""
    # TODO
    return {"issue": "healthy", "unreachable": []}


def fix_advertised_listeners(props, reachable_host):
    """Return a copy of props with advertised.listeners rewritten to reachable_host."""
    # TODO
    return dict(props)


# 6. Reassignment throttle too low (pure)
def diagnose_throttle(throttle_config, floor_bytes_per_s):
    """Return {'issue': 'throttle_too_low'|'healthy', 'rates': [...]}."""
    # TODO
    return {"issue": "healthy", "rates": []}


def fix_throttle(throttle_config, floor_bytes_per_s):
    """Return a copy raising the throttled rates to at least floor_bytes_per_s."""
    # TODO
    return dict(throttle_config)

Hints

Hint 1

Each scenario is a diagnose_X + fix_X pair. The diagnose functions READ state (AdminClient metadata/configs, or a props/throttle dict) and return a small structured dict describing the problem. The fix functions apply the smallest change and the test re-checks health.

Hint 2

Live ones: under_replicated -> a partition's ISR shorter than its replicas (fix: start the dead broker); min_isr -> topic min.insync.replicas higher than the number of in-sync replicas (fix: lower it via kafka-configs.sh --alter); retention -> retention.ms below the floor (fix: set it to the floor); lag -> committed behind end (fix: reset the group to the end so lag drains).

Hint 3

Pure ones (no broker): advertised_listeners -> the advertised host is 0.0.0.0 or not the reachable host (fix: rewrite it to reachable_host:port, keeping the protocol/port); throttle -> leader/follower replication throttled rate below the floor or set to -1/unset (fix: return a config raising it to at least the floor).

Run it locally
uv run de start KFK-14/01-broken-clusters
# edit workspace/KFK-14/01-broken-clusters/
uv run de test