A Kafka certification question asks:

A company created a production topic prod_topic. Its retention should be 10 days. What's the most efficient way?

Correct: B.

The point of the question is topic-level vs broker-level, and the right unit (retention.ms, not retention.days). Instead of shelling out to kafka-configs.sh, this notebook does it programmatically from a cell with the Kafka AdminClient (kafka-python) — the API the CLI is a thin wrapper over — and then proves it landed as a topic-level override by reading the config's source.

Option	Verdict	Why
A retention.days=10 (topic)	❌	retention.days is not a valid config key
B retention.ms=864000000 (topic)	✅	correct key + unit, scoped to one topic
C retention.days=10 (broker)	❌	invalid key and broker-wide
D retention.ms=864000000 (broker)	❌	broker default — hits every topic, and the broker key is log.retention.ms

Prerequisites

The infra/ksqldb Docker stack must be running — we talk straight to the Kafka broker on localhost:9092 (not ksqlDB this time):

docker compose -f infra/ksqldb/docker-compose.yml up -d

Uses kafka-python (same library as the producer demo). Run top to bottom.

Connect the AdminClient + a retention reader

describe_configs returns each config as a tuple (name, value, read_only, source, is_sensitive, synonyms). The source is the tell: 5 = a default (broker/cluster-wide), 1 = a per-topic override. Watching it flip from 5 → 1 is the whole B-vs-D distinction, live.

from kafka.admin import KafkaAdminClient, ConfigResource, ConfigResourceType, NewTopic
from kafka.errors import KafkaError, TopicAlreadyExistsError

BOOTSTRAP = "localhost:9092"
TOPIC = "prod_topic"

# Config-source codes from the Kafka protocol (DescribeConfigs).
CONFIG_SOURCE = {
    1: "DYNAMIC_TOPIC_CONFIG (per-topic override)",
    4: "STATIC_BROKER_CONFIG",
    5: "DEFAULT_CONFIG (broker/cluster default)",
}

try:
    admin = KafkaAdminClient(bootstrap_servers=BOOTSTRAP, request_timeout_ms=8000)
    print("admin connected ->", BOOTSTRAP)
except KafkaError as e:
    print(f"[broker unreachable at {BOOTSTRAP}] {type(e).__name__}: {e}")
    print("Bring the stack up: docker compose -f infra/ksqldb/docker-compose.yml up -d")


def retention():
    """Return (value, source_code) for the topic's retention.ms."""
    res = ConfigResource(ConfigResourceType.TOPIC, TOPIC)
    described = admin.describe_configs([res])[0]
    for entry in described.resources[0][4]:
        if entry[0] == "retention.ms":
            return entry[1], entry[3]
    return None, None


def show(label):
    value, source = retention()
    print(f"{label}: retention.ms={value}  "
          f"({(int(value) / 86_400_000):g} days)  "
          f"source={source} -> {CONFIG_SOURCE.get(source, source)}")

Output:

[stdout]
admin connected -> localhost:9092

Setup — make sure prod_topic exists

Idempotent: create it if it isn't there.

try:
    admin.create_topics([NewTopic(TOPIC, num_partitions=1, replication_factor=1)])
    print("created", TOPIC)
except TopicAlreadyExistsError:
    print("already exists:", TOPIC)

Output:

[stdout]
created prod_topic

Before — retention is the inherited default

A fresh topic has no override, so retention.ms reads as the default (604800000 ms = 7 days), sourced from the broker/cluster config.

show("before")

Output:

[stdout]
before: retention.ms=604800000  (7 days)  source=5 -> DEFAULT_CONFIG (broker/cluster default)

The answer (B) — set the topic-level retention.ms

10 days = 10 × 24 × 60 × 60 × 1000 = 864000000 ms. alter_configs with a ConfigResource(TOPIC, ...) writes the override to this topic only — the programmatic equivalent of:

kafka-configs --bootstrap-server localhost:9092 \
  --alter --entity-type topics --entity-name prod_topic \
  --add-config retention.ms=864000000

We re-read after a short beat (a describe issued the instant after the alter can return stale data).

import time

TEN_DAYS_MS = 10 * 24 * 60 * 60 * 1000  # 864000000
assert TEN_DAYS_MS == 864_000_000

resp = admin.alter_configs([
    ConfigResource(ConfigResourceType.TOPIC, TOPIC,
                   configs={"retention.ms": str(TEN_DAYS_MS)}),
])
err = resp.resources[0][0]
print("alter_configs error_code:", err, "(0 = success)")

time.sleep(0.5)
show("after ")

Output:

[stdout]
alter_configs error_code: 0 (0 = success)
after : retention.ms=864000000  (10 days)  source=1 -> DYNAMIC_TOPIC_CONFIG (per-topic override)

Why B — and why the source code proves it

source flipped 5 → 1: from DEFAULT_CONFIG (a broker/cluster-wide default) to DYNAMIC_TOPIC_CONFIG (an override on this topic). That's the exact thing the exam is testing:

• B is right — retention.ms is the real key, 864000000 is 10 days in ms, and scoping it to prod_topic leaves every other topic untouched (source 1).
• A / C use retention.days, which isn't a Kafka config key at all (some UIs show days, but the wire config is always retention.ms).
• D would change the broker default — hitting every topic that doesn't override it — and the broker-level key isn't even retention.ms, it's log.retention.ms / log.retention.hours. Wrong scope and wrong name.

API notes

• kafka-python's alter_configs uses the legacy AlterConfigs API, which replaces a resource's dynamic configs with what you pass. For one key on an otherwise-default topic that's fine; if a topic already has several overrides, read them first (or use a broker/client that supports IncrementalAlterConfigs) so you don't clear the others.
• retention.ms = -1 means infinite retention — a common accidental inherit from a broker-wide change.

Teardown

Drop the demo topic so the stack is clean. (In real life you'd keep prod_topic and its override — this is a lab.)

admin.delete_topics([TOPIC])
print("deleted", TOPIC)
admin.close()

Output:

[stdout]
DeleteTopicsResponse_v3(throttle_time_ms=0, topic_error_codes=[(topic='prod_topic', error_code=0)])
deleted prod_topic

🛠️ Ran and generated by PySpark DrillBench.

https://preview.redd.it/3u0jur0pt26h1.png?width=1524&format=png&auto=webp&s=b3fa9b4f0d0417847dcb7787b80ff29b423736b3