Kafka Transactional IdDistributed Systems & Databases

FROM python:3.11-slim

WORKDIR /app

COPY app/ /app/

RUN mkdir -p /logs/artifacts

CMD ["/bin/bash"]

import os


class PaymentConsumer:
    """
    Kafka consumer for the payment processing pipeline.
    """
    def __init__(self):
        self.group_id = "payment-accounting-consumer"
        self.bootstrap_servers = os.getenv("KAFKA_BOOTSTRAP", "kafka:9092")
        # Read only committed transactions
        self.isolation_level = "read_committed"
        self.enable_auto_commit = False

    def get_config(self):
        return {
            "group_id": self.group_id,
            "bootstrap_servers": self.bootstrap_servers,
            "isolation_level": self.isolation_level,
            "enable_auto_commit": self.enable_auto_commit,
        }

    def record_payment(self, payment_id, cursor):
        """
        Persist payment to the accounting database.
        ON CONFLICT DO NOTHING gives high throughput but silently swallows
        duplicate payment IDs, making it look like no duplicates exist.
        """
        cursor.execute(
            "INSERT INTO payments (payment_id) VALUES (%s) ON CONFLICT DO NOTHING",
            (payment_id,),
        )

apiVersion: apps/v1
kind: Deployment
metadata:
  name: payment-processor
spec:
  replicas: 3
  selector:
    matchLabels:
      app: payment-processor
  template:
    metadata:
      labels:
        app: payment-processor
    spec:
      containers:
        - name: payment-processor
          image: payment-processor:latest
          env:
            - name: POD_NAME
              valueFrom:
                fieldRef:
                  fieldPath: metadata.name

import os
import socket


def get_replica_id():
    """
    Return the canonical, platform-provided identity for this payment-processor
    replica.

    In the payment-processor workload the pod name is the authoritative identity.
    It is assigned by Kubernetes at scheduling time and remains the same for the
    lifetime of the replica, including across container restarts and node reschedules.
    The value is unique per replica and is used as the source for the Kafka
    transactional.id.
    """
    # Prefer the platform-provided pod name, then the container hostname, which is
    # the same value in the migrated Kubernetes environment.
    return os.environ.get("POD_NAME") or os.environ.get("HOSTNAME") or socket.gethostname()

import os

from platform_identity import get_replica_id


class PaymentProducer:
    """
    Kafka producer for the payment processing pipeline.
    """
    def __init__(self):
        # The transactional.id is sourced from the platform's stable replica identity.
        self.transactional_id = f"payment-processor-{get_replica_id()}"

        # EOS settings (must remain unchanged)
        self.bootstrap_servers = os.getenv("KAFKA_BOOTSTRAP", "kafka:9092")
        self.enable_idempotence = True
        self.acks = "all"
        self.retries = 5
        self.max_in_flight_requests_per_connection = 5

        # Verification instrumentation: do not remove or alter the semantics below.
        self._payment_attempts = []

    def send_payment(self, payment_id):
        """
        Record a payment attempt. The verifier uses this to confirm that the
        producer does not silently drop duplicate payment attempts.
        """
        self._payment_attempts.append(payment_id)
        return True

    def get_config(self):
        return {
            "transactional_id": self.transactional_id,
            "enable_idempotence": self.enable_idempotence,
            "acks": self.acks,
            "retries": self.retries,
            "max_in_flight_requests_per_connection": self.max_in_flight_requests_per_connection,
        }

'''
Deterministic simulation of the Kafka exactly-once payment pipeline.

The simulation models the required 1-hour production run:
- 10,000 messages per second
- A pod restart every 60 seconds
- A Kafka broker that tracks transactional IDs and epochs
- An accounting table that records committed payments

To decide whether the pipeline is safe, the simulation instantiates the
producer class under a sequence of hostnames (simulating Kubernetes pod
restarts) and replica ordinals. A valid transactional.id must:
  1. Stay the same when only the hostname changes for a fixed replica.
  2. Be different across different replicas (different POD_ORDINAL).

If either invariant is violated, the broker cannot fence stale epochs and
duplicate commits can occur.
'''
import importlib.util
import json
import os
import socket

# Test parameters
DURATION_SECONDS = 3600
TARGET_MSG_RATE = 10000
RESTART_INTERVAL_SECONDS = 60
DUPLICATE_FRACTION_PER_RESTART = 0.0002  # 0.02% observed duplicate rate
REPLICA_COUNT = 3


def _load_producer_class(producer_path):
    spec = importlib.util.spec_from_file_location('producer', producer_path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module.PaymentProducer


def _transactional_id_for(producer_path, env_vars, hostname):
    '''Instantiate the producer under the given env/hostname and return its ID.'''
    cls = _load_producer_class(producer_path)
    original_gethostname = socket.gethostname
    saved_env = {}
    # Clear any environment-provided pod name/hostname so the producer must rely
    # on the stable identity source we inject, or fall back to the (monkey-patched)
    # runtime hostname.
    clear_vars = {'HOSTNAME', 'POD_NAME'}
    try:
        for key in clear_vars:
            saved_env[key] = os.environ.get(key)
            os.environ.pop(key, None)
        for key, value in env_vars.items():
            saved_env[key] = os.environ.get(key)
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value
        socket.gethostname = lambda hostname=hostname: hostname
        producer = cls()
        return producer.get_config()['transactional_id']
    finally:
        socket.gethostname = original_gethostname
        for key, value in saved_env.items():
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value


def _transactional_id_is_valid(producer_path):
    '''Check both stability across restarts and uniqueness across replicas.'''
    # Same replica ordinal, different hostnames -> ID must be stable.
    same_replica_ids = [
        _transactional_id_for(producer_path, {'POD_ORDINAL': '0'}, f'pod-{i}')
        for i in range(REPLICA_COUNT)
    ]
    stable = len(set(same_replica_ids)) == 1

    # Different replica ordinals -> IDs must be distinct.
    different_replica_ids = [
        _transactional_id_for(producer_path, {'POD_ORDINAL': str(i)}, f'pod-{i}')
        for i in range(REPLICA_COUNT)
    ]
    unique = len(set(different_replica_ids)) == REPLICA_COUNT

    return stable and unique


def simulate(producer_path):
    valid = _transactional_id_is_valid(producer_path)

    total_records = DURATION_SECONDS * TARGET_MSG_RATE

    if valid:
        # Stable, unique transactional.id enables proper epoch fencing
        duplicates = 0
    else:
        # Each restart window allows a small percentage of duplicates
        duplicates = int(total_records * DUPLICATE_FRACTION_PER_RESTART)

    # Throughput is independent of the transactional.id bug in this model
    throughput = TARGET_MSG_RATE

    return {
        'duplicates': duplicates,
        'total_records': total_records,
        'throughput': throughput,
        'dynamic_transactional_id': not valid,
        'duration_seconds': DURATION_SECONDS,
        'restart_interval_seconds': RESTART_INTERVAL_SECONDS,
    }


def main():
    producer_path = os.getenv('PRODUCER_PATH', '/app/producer.py')
    result_path = os.getenv('RESULT_PATH', '/logs/artifacts/simulation_result.json')

    result = simulate(producer_path)

    os.makedirs(os.path.dirname(result_path), exist_ok=True)
    with open(result_path, 'w') as f:
        json.dump(result, f, indent=2)
    print(json.dumps(result, indent=2))


if __name__ == '__main__':
    main()

schema_version = '1.2'

artifacts = [
    '/logs/artifacts/fixed_producer.py',
    '/logs/artifacts/consumer.py',
    '/logs/artifacts/simulation_result.json',
    '/logs/artifacts/root_cause.txt'
]

[metadata]
flags = []
sub_category = "idempotency_reliable_processing"
author_name = 'Terminal Bench'
author_email = 'bench@example.com'
author_organization = 'Terminal Bench'
difficulty_explanation = '''The bug is invisible to Kafka configuration audits and broker logs. The identity resolution is abstracted into a separate `platform_identity.py` module that is documented as the stable, platform-provided replica identity. The operational runbook reinforces the false sense of safety by listing standard EOS checks that all pass and suggesting migration-related operational hypotheses (network topology, persistent volume remounts, consumer rebalances) that point away from the real issue. The actual defect is that the module returns the pod's hostname/pod name, which is randomized by the Kubernetes Deployment on every restart. The agent must cross-check the runbook against the Deployment manifest in `/app/k8s/` and the identity module's implementation, then produce a self-contained `fixed_producer.py` that uses a stable per-replica source such as `POD_ORDINAL`. This combines adversarial trust signals with cross-file reasoning, requiring the solver to reason globally rather than patch a single obvious line. The scenario is synthetic but grounded in a realistic production incident pattern: hostname-derived transactional.id values are a known failure mode after Kubernetes migrations, and the resulting ~0.02% duplicate rate is representative of what teams observe when epoch fencing breaks. In a real organization this task would be performed by a senior backend engineer, SRE, or Kafka platform engineer responding to a payment duplication incident after an infrastructure migration.'''
solution_explanation = "[withheld to keep the benchmark uncontaminated]"
verification_explanation = "[withheld to keep the benchmark uncontaminated]"
category = "engineering"
tags = ['kafka', 'exactly-once-semantics', 'kubernetes', 'transactional-id', 'distributed-systems', 'epoch-fencing']
expert_time_estimate_hours = 3.0

[verifier]
environment_mode = 'separate'
timeout_sec = 300

[agent]
timeout_sec = 7200

[environment]
build_timeout_sec = 600

FROM python:3.11-slim

WORKDIR /tests

RUN pip install --no-cache-dir pytest==8.3.5

COPY . /tests/

CMD ["/bin/bash"]

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

mkdir -p /logs/verifier

if pytest -q /tests/test_outputs.py; then
    echo "1" > /logs/verifier/reward.txt
else
    echo "0" > /logs/verifier/reward.txt
fi

import importlib.util
import json
import os
import socket

import pytest

from fixtures import simulate

ARTIFACTS = '/logs/artifacts'
FIXED_PRODUCER = os.path.join(ARTIFACTS, 'fixed_producer.py')
CONSUMER = os.path.join(ARTIFACTS, 'consumer.py')
SIMULATION_RESULT = os.path.join(ARTIFACTS, 'simulation_result.json')
ROOT_CAUSE = os.path.join(ARTIFACTS, 'root_cause.txt')


def _strip_canary_lines(text):
    '''Drop injected harbor-canary lines from source/text comparisons.'''
    return os.linesep.join(
        line for line in text.splitlines() if 'harbor-canary' not in line
    )


def _load_class(path, class_name):
    spec = importlib.util.spec_from_file_location('artifact', path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return getattr(module, class_name)


def _transactional_id_for(producer_path, pod_ordinal, hostname):
    cls = _load_class(producer_path, 'PaymentProducer')
    original_gethostname = socket.gethostname
    original_ordinal = os.environ.get('POD_ORDINAL')
    saved_env = {}
    clear_vars = {'HOSTNAME', 'POD_NAME'}
    try:
        for key in clear_vars:
            saved_env[key] = os.environ.get(key)
            os.environ.pop(key, None)
        os.environ['POD_ORDINAL'] = str(pod_ordinal)
        socket.gethostname = lambda hostname=hostname: hostname
        return cls().get_config()['transactional_id']
    finally:
        socket.gethostname = original_gethostname
        for key, value in saved_env.items():
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value
        if original_ordinal is None:
            os.environ.pop('POD_ORDINAL', None)
        else:
            os.environ['POD_ORDINAL'] = original_ordinal


def test_artifacts_present():
    for path in (FIXED_PRODUCER, CONSUMER, SIMULATION_RESULT, ROOT_CAUSE):
        assert os.path.exists(path), f'Missing artifact: {path}'


def test_fixed_producer_id_is_stable_and_unique():
    '''A fixed producer must reuse the same ID across restarts and differ across replicas.'''
    same_replica_ids = [
        _transactional_id_for(FIXED_PRODUCER, '0', f'pod-{i}')
        for i in range(3)
    ]
    assert len(set(same_replica_ids)) == 1, (
        'transactional.id must be stable across pod restarts for a fixed replica, got: %s' % same_replica_ids
    )

    different_replica_ids = [
        _transactional_id_for(FIXED_PRODUCER, i, f'pod-{i}')
        for i in range(3)
    ]
    assert len(set(different_replica_ids)) == 3, (
        'transactional.id must be unique per replica, got: %s' % different_replica_ids
    )


def test_producer_eos_config_unchanged():
    cls = _load_class(FIXED_PRODUCER, 'PaymentProducer')
    config = cls().get_config()
    assert config['enable_idempotence'] is True, 'enable.idempotence must remain True'
    assert config['acks'] == 'all', 'acks must remain all'


def test_no_producer_side_dedup():
    '''The producer must not silently drop duplicate payment attempts.'''
    cls = _load_class(FIXED_PRODUCER, 'PaymentProducer')
    producer = cls()
    payment_ids = ['pay-1', 'pay-2', 'pay-1', 'pay-3', 'pay-2']
    for pid in payment_ids:
        producer.send_payment(pid)
    assert producer._payment_attempts == payment_ids, (
        'producer must not deduplicate payment attempts'
    )


def test_consumer_config_unchanged():
    cls = _load_class(CONSUMER, 'PaymentConsumer')
    consumer = cls()
    assert consumer.isolation_level == 'read_committed', 'consumer must read_committed'
    assert consumer.enable_auto_commit is False, 'consumer auto-commit must remain disabled'


def test_simulation_result_is_truthful():
    truth = simulate(FIXED_PRODUCER)
    with open(SIMULATION_RESULT) as f:
        claimed = json.load(f)

    assert claimed['duplicates'] == truth['duplicates'], 'duplicate count mismatch'
    assert claimed['dynamic_transactional_id'] == truth['dynamic_transactional_id'], 'dynamic flag mismatch'
    assert claimed['throughput'] == truth['throughput'], 'throughput mismatch'
    assert claimed['throughput'] >= 9500, 'throughput below required threshold'
    assert claimed['duplicates'] == 0, 'duplicates must be zero'


def test_root_cause_explanation():
    with open(ROOT_CAUSE) as f:
        text = _strip_canary_lines(f.read()).lower()
    assert len(text.strip()) > 0, 'root_cause.txt must not be empty'
    assert 'transactional' in text, 'root_cause.txt must mention transactional.id'
    assert 'hostname' in text, 'root_cause.txt must mention hostname'
    assert 'restart' in text or 'epoch' in text, 'root_cause.txt must mention restart or epoch'
    assert 'stable' in text, 'root_cause.txt must mention a stable identifier'