Principal Engineer El Kitabı — Python, Reactive, Kafka & Dağıtık Sistemler

7 Mart 2026 · netologist · 40 dakika, 8397 kelime · #distributed-systems #kafka #note #python #reactive-programming #resilience

Python · Reactive Programming · Kafka Event Streaming · Distributed Systems · Resilience

Büyük ölçekli, Python tabanlı dağıtık sistemleri tasarlamak, inşa etmek ve işletmek isteyen mühendisler için author-level derinlikte teknik referans.

1. Author Seviyesinde Python

1.1 Data Model — Python’ın Gerçek Sözleşmesi

Python’daki “magic method"lar sadece syntactic sugar değildir; interpreter’ın senin objelerinle müzakere ettiği gerçek arayüz sözleşmesidir. __new__ ile __init__ arasındaki farkı, descriptor protokolünü ve MRO’yu (Method Resolution Order, C3 linearization) anlamak, orta seviye ile ileri seviye Python bilgisini birbirinden ayıran şeydir.

class Descriptor:
    """Data descriptor: hem __get__ hem __set__ tanımlar."""
    def __set_name__(self, owner, name):
        self._name = f"_{name}"

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self._name, None)

    def __set__(self, obj, value):
        if not isinstance(value, (int, float)):
            raise TypeError(f"{self._name} sayısal olmalı")
        setattr(obj, self._name, value)


class Point:
    x = Descriptor()
    y = Descriptor()

    def __init__(self, x, y):
        self.x = x
        self.y = y

Neden önemli: @property, ORM’ler (SQLAlchemy, Django) ve dataclasses hepsi descriptor protokolü üzerine inşa edilmiştir. Data descriptor’lar (__set__ tanımlayanlar) instance __dict__‘ine göre önceliklidir; non-data descriptor’lar (sadece __get__, örneğin fonksiyon/metodlar) önceliksizdir. Instance attribute’ların methodları gölgeleyebilmesinin ama property’leri gölgeleyememesinin tam sebebi budur.

1.2 Metaclass’lar — Gerçekten Ne Zaman Gerekli

Metaclass “bir sınıfın sınıfı"dır. Bunları tüm bir sınıf hiyerarşisi boyunca invariant zorlamak veya otomatik kayıt (registration) için kullan — estetik için değil.

class PluginMeta(type):
    registry: dict[str, type] = {}

    def __new__(mcs, name, bases, namespace, **kwargs):
        cls = super().__new__(mcs, name, bases, namespace)
        if bases:  # base sınıfın kendisini atla
            PluginMeta.registry[name] = cls
        return cls


class Plugin(metaclass=PluginMeta):
    pass


class CSVExporter(Plugin):
    pass

# PluginMeta.registry == {"CSVExporter": <class CSVExporter>}

__init_subclass__ (PEP 487), meşru metaclass kullanım senaryolarının %90’ını çok daha az karmaşıklıkla çözer:

class Plugin:
    registry: dict[str, type] = {}

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        Plugin.registry[cls.__name__] = cls

Pratik kural: önce __init_subclass__‘a yönel. Metaclass’a sadece sınıf oluşturulma anının kendisine müdahale etmen gerektiğinde (örn. sınıf objesi oluşmadan önce namespace’i validate etmek) veya sıralı/özel namespace’ler için __prepare__ kontrolü gerektiğinde başvur.

1.3 Concurrency Modelleri: GIL, Thread’ler, Process’ler ve Async

Python’ın concurrency hikayesinde üç ayrı eksen var, ve bunları birbirine karıştırmak en yaygın mimari hatadır:

Model	En uygun olduğu yer	Kısıtlama
`threading`	I/O-bound işler	GIL bytecode yürütmesini seri hale getirir (çoğunlukla — bkz. 1.3.1)
`multiprocessing`	CPU-bound işler	Process spawn/IPC overhead’i, pickling maliyeti
`asyncio`	Yüksek eşzamanlı I/O (binlerce bağlantı)	Tek thread’li; bloklayan tek bir çağrı her şeyi durdurur

1.3.1 GIL ve Kaldırılması (PEP 703 / Python 3.13+)

Global Interpreter Lock, aynı anda sadece bir thread’in Python bytecode’u çalıştırmasını garanti eder. Bu, veri yapılarını daha yüksek seviyede race condition’lardan korumaz (x += 1 atomik değildir) ve I/O ile bazı C-extension çağrıları (NumPy vb.) sırasında serbest bırakılır.

Python 3.13, deneysel free-threaded build‘i tanıttı (--disable-gil, PEP 703). 3.13/3.14 itibarıyla bu hâlâ opt-in ve gerçek ekosistem maliyetleri var (C-extension ABI uyumluluğu, tek-thread’li kodda per-object locking overhead’i). 2026’daki bir principal engineer duruşu: free-threading’i, tüm dependency zincirini kontrol etmiyorsan ve benchmark etmediysen, henüz varsayılan bir production kabulü değil, geleceğe yönelik tasarlanacak bir kapasite olarak ele al.

1.3.2 asyncio İç Yapısı

Event loop, şunların etrafında kurulmuş tek-thread’li cooperative bir scheduler’dır:

Future’lar: eventual bir sonucu temsil eden low-level awaitable.
Task’lar: bir coroutine’i saran, loop.call_soon ile zamanlanan bir Future.
Selector’lar: select/epoll/kqueue tabanlı I/O hazır olma bildirimi.

import asyncio

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} tamamlandı"

async def main():
    async with asyncio.TaskGroup() as tg:  # Python 3.11+: structured concurrency
        t1 = tg.create_task(fetch("A", 1))
        t2 = tg.create_task(fetch("B", 2))
    print(t1.result(), t2.result())

asyncio.run(main())

Structured concurrency (TaskGroup, PEP 654 ExceptionGroup), asyncio’nun en kötü tarihsel tuzağını düzeltti: exception’ları sessizce kaybolan öksüz (orphaned) task’lar. Bir TaskGroup, alt task’lardan biri başarısız olursa kardeşlerinin iptal edilmesini ve tüm exception’ların bir ExceptionGroup‘a toplanmasını garanti eder — artık “fırlat ve unut” sızıntısı yok.

async def riskli():
    raise ValueError("patladı")

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(riskli())
            tg.create_task(asyncio.sleep(5))
    except* ValueError as eg:
        print("Yakalandı:", eg.exceptions)

Kütüphaneler arası gerçekten structured, cancellation-safe concurrency için pek çok production ekibi hâlâ anyio‘yu (asyncio/trio’yu birleştiren) veya doğrudan trio‘yu tercih ediyor; çünkü trio’nun nursery modeli structured concurrency’yi yapısal olarak zorunlu kılıyor, unstructured bir çekirdeğe sonradan eklenmiş opt-in bir özellik olarak değil.

1.3.3 Ölçekte Yaygın asyncio Tuzakları

Loop’u bloklamak: bir coroutine içindeki herhangi bir senkron, CPU-ağır veya bloklayan I/O çağrısı, tüm eşzamanlı task’ları durdurur. loop.run_in_executor(None, blocking_fn) veya bir ThreadPoolExecutor ile offload et.
Sınırsız task oluşturma: asyncio.gather(*[coro() for _ in range(1_000_000)]) belleği/FD’leri tüketir. Bir semaphore veya sınırlı worker pool deseni kullan.
Cancellation cooperative’dir: bir task sadece .cancel() çağrısından sonra bir await noktasına ulaştığında durur. Temizliği try/finally içine sar, ve CancelledError‘ın genelde yutulmayıp yeniden fırlatılması gerektiğinin farkında ol.

sem = asyncio.Semaphore(50)

async def bounded_fetch(url):
    async with sem:
        return await fetch_url(url)

async def main(urls):
    async with asyncio.TaskGroup() as tg:
        for url in urls:
            tg.create_task(bounded_fetch(url))

1.4 Ölçekte Typing

Büyük kod tabanları için modern Python (3.11+) typing:

from typing import Protocol, TypeVar, Generic, ParamSpec, overload
from collections.abc import Callable

T = TypeVar("T")
P = ParamSpec("P")

class Repository(Protocol[T]):
    """Structural typing: kalıtım gerekmez, statik kontrollü duck typing."""
    async def get(self, id: str) -> T | None: ...
    async def save(self, entity: T) -> None: ...

class EventHandler(Generic[T]):
    def __init__(self, handler: Callable[[T], None]) -> None:
        self._handler = handler

def retry(fn: Callable[P, T]) -> Callable[P, T]:
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        for attempt in range(3):
            try:
                return fn(*args, **kwargs)
            except Exception:
                if attempt == 2:
                    raise
        raise RuntimeError("buraya ulaşılamaz")
    return wrapper

Dependency inversion için ABC yerine Protocol. Event-driven bir sistemde KafkaConsumer, InMemoryQueue ve TestStub sınıfların hepsi, ortak bir base class olmadan bir MessageSource Protocol’ünü tatmin edebilir — Python’da ports/adapters (hexagonal architecture) mimarisini dürüst tutan şey tam olarak budur.

Birden fazla kişinin bakımını yaptığı, birkaç bin satırı aşan herhangi bir kod tabanı için CI’da mypy --strict veya pyright pazarlığa açık değildir — statik olarak yakalanan tip hataları, aksi halde distributed bir consumer group’ta sabah 3’te ortaya çıkacak tam olarak o bug sınıfıdır.

1.5 Performans Mühendisliği

Optimize etmeden önce profille. cProfile + snakeviz, ya da production’da process’i yeniden başlatmadan sampling profiling için py-spy (py-spy dump --pid <pid>; py-spy top --pid <pid>).
Veri yapılarının gerçek karmaşıklığını bil. list.pop(0) O(n)‘dir; collections.deque kullan. Membership testleri: set/dict ortalama O(1), list O(n).
Yüksek kardinaliteli objeler (milyonlarca instance) için __slots__ — per-instance __dict__‘i kaldırır, belleği %40–50 azaltır ve attribute erişimini hızlandırır.
Gerçek CPU-bound hot path’ler için Cython / C extension / Rust (PyO3 üzerinden) — ama sadece profiling darboğazın pure Python’da kaçınılmaz olduğunu kanıtladıktan sonra.
Batch’le ve vektörize et. Veri üzerinde satır satır Python döngüleri, neredeyse her zaman NumPy/Polars vektörize operasyonlarla değiştirilebilir, çoğu zaman 10-100x daha hızlı.

from dataclasses import dataclass

@dataclass(slots=True, frozen=True)
class Event:
    id: str
    payload: bytes
    timestamp: float

1.6 Design Pattern’lar, Pythonic Yöntemle

Klasik GoF pattern’ları Python’da genelde çöker çünkü dil zaten first-class fonksiyonlara, closure’lara ve duck typing’e sahiptir:

Strategy pattern → sadece bir fonksiyon/callable pass et.
Singleton → bir modül zaten bir singleton’dır; ya da bir factory fonksiyonda functools.lru_cache(maxsize=None) kullan.
Observer pattern → reactive programming’in (Bölüm 2) elle yazılmış bir observer listesinde bulunmayan backpressure ve composability ile formalize ettiği şey tam olarak budur.
Decorator pattern → Python’ın @decorator sözdizimi dile gömülü olarak bu pattern’in kendisidir.

from functools import lru_cache

@lru_cache(maxsize=None)
def get_settings() -> "Settings":
    return Settings()  # bir kere oluşturulur, tekrar kullanılır — temiz bir singleton

1.7 Derinlemesine: Çekirdek asyncio Primitifleri, Tam Olarak

Bu bölüm var, çünkü şu dokuz kavram — async def, await, event loop, blocking I/O, non-blocking I/O, concurrency, timeout, cancellation, semaphore — çoğu Python mühendisinin zihinsel modelinin production yükü altında tam olarak nerede çöktüğü yerdir. Her biri burada bir principal engineer’ın ihtiyaç duyduğu derinlikte ele alınıyor: “ne yapar” değil, “kaputun altında gerçekte ne dir ve sessizce nerede başarısız olur.”

1.7.1 `async def` — Gerçekte Ne Oluşturur

async def, çağrıldığında çalışan bir fonksiyon tanımlamaz. Bir coroutine fonksiyonu tanımlar — onu çağırmak, gövdenin tek bir satırını bile çalıştırmadan anında bir coroutine objesi döndürür.

async def fetch_data(url: str) -> str:
    print("Bu satır henüz çalışmıyor")
    return "data"

coro = fetch_data("https://example.com")
print(type(coro))          # <class 'coroutine'>
print(coro)                # <coroutine object fetch_data at 0x...>
# "Bu satır henüz çalışmıyor" hiç yazdırılmadı — gövde çalışmadı.

Coroutine objesi, bir şey onu sürene kadar hareketsizdir — ya await edilir, ya asyncio.create_task() içine sarılır, ya da asyncio.run()‘a geçirilir. Bu yüzden fetch_data("url") (yani await‘i unutmak) birinci sınıf gerçek dünya bug’ıdır: Python hiçbir hata fırlatmaz, sessizce kullanılmayan bir coroutine objesi oluşturur ve en iyi ihtimalle bir RuntimeWarning: coroutine was never awaited alırsın — ya da daha kötüsü, coroutine hiç çalışmaz ve programın sessizce hiçbir şey yapmaz.

async def main():
    result = fetch_data("https://example.com")  # BUG: await eksik
    print(result)  # "data" değil, "<coroutine object ...>" yazdırır

Kaputun altında: bir coroutine, generator’larla aynı makine üzerine inşa edilmiştir (PEP 492, async def/await‘i, yield için zaten var olan generator protokolünün üzerine kurdu). İç yapıda, bir coroutine objesinin bir send() metodu vardır; event loop, onu StopIteration fırlatana (dönüş değerini taşıyarak) veya bir şey beklemek için kontrolü geri verene kadar ilerletmek için tekrar tekrar send(None) çağırır.

1.7.2 `await` — Gerçek Askıya Alma Noktası

await, yalnızca bir async def gövdesi içinde yasaldır (başka yerde kullanmak SyntaxErrordır). Tam olarak üç şey yapar:

await edilen objenin __await__() metodunu çağırır ve bir iterator elde eder.
O iterator’ı sürer, iterator her yield ettiğinde kontrolü event loop’a geri verir (gerçek askıya alma budur — kontrol gerçekten loop’a döner, o da başka task’ları çalıştırabilir).
Await edilen operasyonun sonucu hazır olduğunda, tam o noktada çalışmaya devam eder; iterator’ın son StopIteration.value‘su, ifadenin sonucu olur.

class Awaitable:
    def __await__(self):
        print("askıya alınıyor...")
        yield  # kontrolü burada event loop'a geri verir
        print("devam ediliyor!")
        return 42

async def main():
    result = await Awaitable()
    print(result)  # 42

asyncio.run(main())
# Çıktı:
# askıya alınıyor...
# devam ediliyor!
# 42

“Awaitable” nedir? __await__ uygulayan her şey: coroutine’ler, asyncio.Future, asyncio.Task (bir Future alt sınıfı) ve bu protokol üzerine inşa edilmiş kütüphanelerdeki objeler. Düz bir generator, uygun şekilde decorate edilmedikçe awaitable değildir — bu, senkron generator’ları yanlışlıkla async koda karıştırırken “object is not awaitable” TypeError‘larının yaygın bir kaynağıdır.

Kritik zihinsel model: await expr, “bu coroutine’i burada askıya al, event loop’un başka faydalı iş yapmasına izin ver, ve expr‘in bir sonucu olduğunda tam burada devam et” anlamına gelir. “Blokla ve bekle” değildir — bu ayrım, asyncio’nun tek bir thread üzerinde on binlerce eşzamanlı bağlantıya hizmet edebilmesinin tüm sebebidir.

1.7.3 Event Loop — Gerçekte Ne Yapar, Adım Adım

Event loop, tek-thread’li, cooperative bir scheduler’dır. Çekirdek algoritması, gerçekte ne olduğuna basitleştirilmiş hâliyle:

loop.run_forever():
    while durdurulmadı:
        timeout = bir sonraki zamanlanmış callback'e kadar geçen süreyi hesapla()
        events = selector.select(timeout)       # BURADA I/O hazır olmasını bekleyerek bloklanır
        for (fd, event) in events:
            callback = kayıtlı_callback[fd]
            callback()                           # örn. bu socket'i bekleyen bir Task'ı devam ettirir
        run_ready_callbacks()                    # call_soon ile zamanlanmış şeyler
        run_due_scheduled_callbacks()             # call_later ile zamanlanmış şeyler

Event loop’un OS thread’ini gerçekten “blokladığı” tek ve tek yer, selector.select(timeout) içindedir — OS’un I/O hazır olma bildirimini bekler (Linux’ta epoll, BSD/macOS’ta kqueue, Windows’ta proactor loop üzerinden IOCP). Geri kalan her şey saf Python bytecode yürütmesidir, bir seferde bir coroutine adımı, asla gerçekten paralel değil.

import asyncio

async def show_loop_identity():
    loop = asyncio.get_running_loop()
    print(f"Çalıştığı loop: {loop}")
    print(f"Çalışıyor mu: {loop.is_running()}")

asyncio.run(show_loop_identity())

Task zamanlaması, somut olarak: asyncio.create_task(coro), bir coroutine’i bir Task içine sarar, ilk adımını loop.call_soon() ile zamanlar ve anında döner — task şimdi çalışmaya başlamaz, loop’un bir sonraki iterasyonunda çalışmaya başlar. Bu yüzden şu yaygın hata sessizce başarısız olur:

async def main():
    task = asyncio.create_task(slow_operation())
    # main() burada `task`'ı await etmeden dönerse,
    # task uçuş halindeyken garbage-collect edilebilir ve
    # "Task was destroyed but it is pending" uyarısı alırsın.
    return "tamamlandı"

Kural: oluşturulan bir task’ı her zaman ya await et, ya güçlü bir referans tut ve kapsayan scope çıkmadan önce await et, ya da bir TaskGroup kullan — asla bir task’ı fırlatıp await edilmeden scope dışına düşmesine izin verme.

1.7.4 Blocking I/O — Gerçekte Neye Mal Olur

“Blocking” şu anlama gelir: çağıran thread, operasyon tamamlanana kadar işletim sistemi tarafından askıya alınır ve o sırada başka hiçbir şey yapamaz — bu, altta yatan sistem çağrısının bir özelliğidir, Python’ın değil.

import time
import asyncio

async def kotu_handler():
    time.sleep(2)          # BLOCKING: TÜM event loop'u 2 saniye dondurur
    return "tamamlandı"

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(kotu_handler())
        tg.create_task(kotu_handler())
        tg.create_task(kotu_handler())
    # Bunlar eşzamanlı ÇALIŞMAZ — toplam süre ~2 değil ~6 saniyedir,
    # çünkü time.sleep() kontrolü asla loop'a geri vermez.

Bir asyncio uygulamasını sessizce sabote eden klasik bloklayan çağrılar: time.sleep(), requests.get() (senkron requests kütüphanesi), senkron dosya I/O’su (yavaş bir dosya sistemi/network mount’unda open().read()), senkron veritabanı driver’ları (asyncpg değil, psycopg2) ve herhangi bir CPU-bound saf-Python hesaplaması (sıkı bir döngü, büyük bir payload’ı JSON-parse etmek, büyük bir string üzerinde regex).

Çözüm — bir thread pool’a offload et, bu gerçekten paralel çalışır (bloklayan syscall sırasında GIL serbest bırakılır) ve event loop diğer coroutine’lere hizmet vermeye devam eder:

import asyncio

async def iyi_handler():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, time.sleep, 2)  # bir thread'de çalışır
    return "tamamlandı"

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(iyi_handler())
        tg.create_task(iyi_handler())
        tg.create_task(iyi_handler())
    # Şimdi bunlar gerçekten örtüşür — toplam süre ~2 saniyedir.

CPU-bound bloklayan iş için (I/O-bound değil), bir thread pool GIL yüzünden yardımcı olmaz — bunun yerine, process sınırlarını aşmanın IPC/pickling maliyetini kabul ederek loop.run_in_executor(ProcessPoolExecutor(), cpu_agir_fn, arg) kullan.

1.7.5 Non-Blocking I/O — Gerçek Mekanizma

Non-blocking I/O, altta yatan syscall’ın, thread’i askıya almak yerine anında ya veriyle ya da “operasyon bloklanırdı” sinyaliyle (EWOULDBLOCK/EAGAIN) dönmesi anlamına gelir. asyncio’nun transport’ları soketleri non-blocking moda ayarlar (socket.setblocking(False)) ve ne zaman tekrar deneyeceğini bilmek için OS’un hazır olma bildirim API’sini kullanır.

import asyncio

async def non_blocking_fetch(host: str, port: int, request: bytes) -> bytes:
    reader, writer = await asyncio.open_connection(host, port)  # non-blocking bağlantı
    writer.write(request)
    await writer.drain()                    # non-blocking yazma, backpressure'a saygılı
    response = await reader.read(-1)        # non-blocking okuma
    writer.close()
    await writer.wait_closed()
    return response

asyncio.open_connection, StreamReader.read() ve StreamWriter.drain()‘ın hepsi altta selector-tabanlı non-blocking transport üzerine inşa edilmiştir — sözdizimsel olarak (await sayesinde) bloklayan çağrılar gibi görünürler, ama kaputun altında selector’a bir callback kaydederler ve kontrolü loop’a geri verirler, yalnızca OS hazır olduğunu bildirdiğinde devam ederler.

Temel ayrım, tam olarak ifade edilmiş:

	Blocking I/O	Non-blocking I/O
Kim bekler	OS thread’i, kernel tarafından askıya alınır	Hiç kimse — syscall her durumda anında döner
Bu sırada ne olur	O thread’de başka hiçbir şey çalışamaz	Event loop başka coroutine’ler çalıştırır
“Bekleme” nasıl ifade edilir	Thread çağrıdan basitçe dönmez	Bir callback kaydedilir; kontrol anında çağırana döner
Çok sayıda eşzamanlı operasyonun maliyeti	Eşzamanlı operasyon başına bir OS thread’i (pahalı: her biri ~1-8MB stack)	Toplamda bir thread, binlerce eşzamanlı operasyon (ucuz: coroutine objeleri ~KB ölçeğinde)

1.7.6 Concurrency — Tam Olarak, ve Parallelism’den Ayrımı

Concurrency (eşzamanlılık), bir programın yapısının bir özelliğidir: birden fazla mantıksal görev, tam olarak aynı anda çalışıp çalışmadıklarına bakılmaksızın, yürütmeleri iç içe geçmiş şekilde devam eder. Parallelism (paralellik), yürütmenin bir özelliğidir: görevler ayrı çekirdeklerde gerçekten tam olarak aynı anda çalışır.

Concurrency (asyncio, tek thread):
Task A: ----[çalış]----[bekle]----[çalış]----[bitti]
Task B: --[bekle]----[çalış]----[bekle]----[çalış]--[bitti]
        (TEK thread üzerinde iç içe — asla aynı anda iki instruction)

Parallelism (multiprocessing, birden fazla çekirdek):
Core 1: Task A: ----------[sürekli çalış]----------[bitti]
Core 2: Task B: ----------[sürekli çalış]----------[bitti]
        (gerçekten eşzamanlı)

asyncio sana paralellik olmadan concurrency verir — I/O-bound iş yükleri için doğru araç olmasının (görevlerin çoğu zamanı hesaplama yapmak değil beklemek ile geçirdiği yerlerde) ve CPU-bound iş yükleri için yanlış araç olmasının (görevlerin zamanlarını gerçekten CPU kullanarak geçirdiği, tek bir thread’de iç içe geçmenin sıfır hızlanma sağladığı yerlerde — orada gerçek paralellik için multiprocessing gerekir, 1.3.1’deki GIL tartışmasına tabi) tam sebebi budur.

1.7.7 Timeout — Kesin Semantik ve Modern API

Bir timeout, await edilen bir operasyonun bir başarısızlık olarak ele alınmadan önce ne kadar sürebileceğini sınırlar. İki API var; aralarındaki farkı bil:

import asyncio

# Eski API: tek bir awaitable'ı sarar
async def wait_for_ile():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=5.0)
    except TimeoutError:
        print("Zaman aşımı — altta yatan task senin için iptal edildi")

# Modern API (3.11+): birden fazla await boyunca kompoze edilebilen bir context manager
async def timeout_context_ile():
    try:
        async with asyncio.timeout(5.0):
            step1 = await do_step_one()
            step2 = await do_step_two()   # 5 saniyelik bütçe HER İKİ adımı birlikte kapsar
    except TimeoutError:
        print("Blok içinde bir yerde zaman aşımı oldu")

asyncio.timeout() (3.11+) daha iyi varsayılandır çünkü kompoze edilebilir: timeout’ları iç içe geçirebilirsin, deadline’ı dinamik olarak yeniden zamanlayabilirsin (asyncio.timeout_at()), ve timeout’un operasyon tam tamamlandığı anda tetiklendiği durumu (3.11’in yeniden yazımından önce wait_for‘un tarihsel olarak kenar durumlarda daha az zarif ele aldığı gerçek bir race condition) doğru şekilde ele alır.

Timeout’ta gerçekte ne olur, tam olarak: altta yatan task iptal edilir (bkz. 1.7.8) — await noktasında TimeoutError fırlatılır ve coroutine, yok edilmeden önce bir finally bloğunda temizlik yapmak için bir şans alır. Bir timeout, mekanik olarak, saate bağlanmış bir cancellation’dır.

1.7.8 Cancellation — Gerçek Yayılma Mekanizması

asyncio’da cancellation cooperative‘dir, preemptive değil. task.cancel() çağırmak task’ı anında durdurmaz — bir CancelledError‘ın task’ın bir sonraki await noktasında fırlatılmasını zamanlar.

import asyncio

async def worker():
    try:
        print("başlıyor")
        await asyncio.sleep(10)   # cancellation BURADA etkili olur, .cancel()'da değil
        print("iptal edilirse bu satır asla çalışmaz")
    except asyncio.CancelledError:
        print("ölmeden önce temizlik yapılıyor")
        raise   # KRİTİK: özel bir sebebin olmadıkça yeniden fırlat

async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("main() task'ın iptal edildiğini görüyor")

asyncio.run(main())
# Çıktı:
# başlıyor
# ölmeden önce temizlik yapılıyor
# main() task'ın iptal edildiğini görüyor

CancelledError‘ı yeniden fırlatmak neden önemli: onu yutmak (yakalayıp yeniden fırlatmamak), task’ın onu await eden herkese normal şekilde tamamlanmış gibi görünmesini sağlar; bu, TaskGroup‘un ve wait_for‘un cancellation durumunu doğru izleme yeteneğini bozar ve bir Task‘ı, event loop’un dahili muhasebesinin beklemediği bir durumda sıkışmış bırakabilir. CancelledError, özellikle geniş bir except Exception:‘ın onu yanlışlıkla yutmaması için Exception‘dan değil BaseException‘dan miras alır.

Asla yield etmeyen, bekleyen bir await, pratikte iptal edilemez: bir coroutine sıkı bir senkron döngüde takılıysa (hiç await noktası yok), üzerinde .cancel() çağırmanın, bir await‘e ulaşana kadar hiçbir etkisi yoktur. Bu, coroutine’ler içindeki bloklayan çağrıların neden tehlikeli olduğunun bir başka açısıdır — sadece yavaş değildirler, task’ı cancellation ve timeout’lara tamamen duyarsız hale getirirler.

Kritik bir bölümü cancellation’dan korumak (shielding):

async def worker():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        # worker()'ın kendisi tekrar iptal ediliyor olsa bile bu temizlik TAMAMLANMALI
        await asyncio.shield(critical_cleanup())
        raise

asyncio.shield(), dış bir cancellation’ın shield’lenmiş iç awaitable’a yayılmasını önler — ama dikkat, dış task’ın kendisi hâlâ iptal edilmiştir; yalnızca shield’lenen operasyon kısa kesilmekten korunur.

1.7.9 Semaphore — Kesin Mekanik ve Doğru Kullanım

asyncio.Semaphore(n), n‘de başlayan dahili bir sayaç tutar. acquire() onu azaltır (sıfırın altına düşerse bekler); release() onu artırır. Concurrency’yi sınırlamanın standart primitif’idir — mantıksal olarak kaç tanesi zamanlanmış olursa olsun, bir bölümün “içinde” aynı anda kaç coroutine olabileceğini sınırlar.

import asyncio

sem = asyncio.Semaphore(3)  # en fazla 3 eşzamanlı operasyon

async def sinirli_fetch(url: str, session):
    async with sem:               # girişte acquire, çıkışta release — exception'da bile
        print(f"{url} çekiliyor, şu anki aktif sayı <= 3")
        return await session.get(url)

async def main(urls: list[str]):
    async with asyncio.TaskGroup() as tg:
        for url in urls:
            tg.create_task(sinirli_fetch(url, session))
    # 10.000 url olsa bile, aynı anda en fazla 3 istek uçuş halindedir.

Semaphore vs BoundedSemaphore: BoundedSemaphore, release() acquire()‘dan daha fazla çağrılırsa ValueError fırlatır — düz bir Semaphore‘un, sayacın amaçlanan maksimumun üzerine tırmanmasına basitçe izin vererek sessizce gizleyeceği belirli bir bug sınıfını (çift release) yakalayan savunmacı bir varyant.

Neden sabit sayıda task değil de semaphore: semaphore, kaç coroutine var olduğunu (coroutine objeleri hafif olduğu için tam 10.000 olabilir, ucuzca) kaçının eşzamanlı çalıştığından (3’e sınırlı) ayırır — bu, “her şeye fan out et, ama gerçek concurrency’yi kısıtla” için doğru pattern’dir; işi elle 3’lük gruplara bölmenin (slotlar boşken bile grup sınırları arasında gereksiz yere serileştiren) aksine.

Fakir adamın connection pool sınırlayıcısı olarak semaphore:

db_semaphore = asyncio.Semaphore(20)  # DB pool'unun max bağlantı sayısıyla eşleşir

async def query(sql: str):
    async with db_semaphore:
        async with pool.acquire() as conn:
            return await conn.fetch(sql)

Bu, uygulamanın, altta yatan connection pool’un gerçekte hizmet verebileceğinden daha fazla mantıksal eşzamanlı sorgu açmasını önler — bu olmadan, fazla coroutine’ler basitçe connection pool’un kendi dahili bekleme mantığı içinde görünmez şekilde kuyruğa girer, bu da uygulama katmanındaki açık bir semaphore’dan çok daha zor gözlemlenebilir ve akıl yürütülebilir.

1.7.10 Dokuz Kavramı Bir Araya Getirmek

import asyncio
import time

sem = asyncio.Semaphore(5)                     # concurrency'yi sınırla

async def fetch_one(item_id: int, client) -> dict:
    async with sem:                             # semaphore: eşzamanlı işi kısıtla
        try:
            async with asyncio.timeout(3.0):    # timeout: HERHANGİ bir çağrının ne kadar sürebileceğini sınırla
                loop = asyncio.get_running_loop()
                # bloklayan legacy client, event loop'u durdurmasın diye offload edildi:
                response = await loop.run_in_executor(None, client.get_sync, item_id)
                return response
        except TimeoutError:
            return {"item_id": item_id, "error": "timeout"}
        except asyncio.CancelledError:
            # cancellation: temizlik yap, sonra yay — asla yutma
            print(f"fetch_one({item_id}) iptal edildi — temizleniyor")
            raise

async def fetch_all(item_ids: list[int], client) -> list[dict]:
    async with asyncio.TaskGroup() as tg:       # structured concurrency: öksüz task yok
        tasks = [tg.create_task(fetch_one(i, client)) for i in item_ids]
    return [t.result() for t in tasks]

async def main():
    client = LegacyBlockingClient()
    start = time.monotonic()
    results = await fetch_all(list(range(100)), client)
    print(f"{len(results)} item, en fazla 5 eşzamanlı ve her biri 3sn'ye sınırlı "
          f"olarak {time.monotonic() - start:.2f}sn'de çekildi.")

asyncio.run(main())   # event loop: oluşturulur, main() tamamlanana kadar çalışır, sonra yıkılır

Dokuz kavramın her biri burada, var olma sebebi olan işi tam olarak yaparak görünür: async def/await coroutine’leri tanımlar ve sürer; event loop (asyncio.run tarafından örtük olarak oluşturulur) her şeyi zamanlar ve sürer; blocking I/O (legacy client) loop’u durdurmasın diye kasıtlı olarak offload edilir, geri kalanı non-blocking tutar; concurrency (paralellik değil) 100 mantıksal fetch’in tek bir thread’de iç içe geçmesine izin verir; semaphore gerçekte kaçının uçuşta olduğunu sınırlar; timeout her bir çağrının en kötü durumunu sınırlar; ve cancellation doğru şekilde ele alınır (temizlik için yakalanır, sonra yeniden fırlatılır) sessizce yutulmak yerine.

2. Reactive Programming

2.1 Reactive Manifesto, Tam Olarak

Reactive sistemler dört özellikle tanımlanır ve bunlar bir kontrol listesi değil, bir bağımlılık zinciri oluşturur:

Responsive ← (üzerine kurulur) Resilient + Elastic ← (şu yolla sağlanır) Message-Driven

Responsive (yanıt veren): sistem, hem başarı hem başarısızlık koşulları altında zamanında yanıt verir.
Resilient (dayanıklı): sistem, izolasyon, containment ve replikasyon yoluyla arıza karşısında responsive kalır.
Elastic (esnek): sistem, ölçeklendirme (in/out) ve back-pressure sinyali yoluyla değişken yük altında responsive kalır.
Message-Driven (mesaj tabanlı): temel. Asenkron, non-blocking mesaj geçişi, gevşek bağlaşım (loose coupling), izolasyon ve konum şeffaflığı sağlar — resilience ve elasticity’yi mümkün kılan şey ilk etapta budur.

2.2 Reactive Streams: Gerçek Spesifikasyon

Reactive Streams spesifikasyonu (RxJava, Project Reactor, Akka Streams’in ve kavramsal olarak RxPY’nin altında yatan), dört arayüz tanımlar:

Publisher<T>.subscribe(Subscriber<T>)
Subscriber<T>.onSubscribe(Subscription)
Subscriber<T>.onNext(T)
Subscriber<T>.onError(Throwable)
Subscriber<T>.onComplete()
Subscription.request(n: Long)
Subscription.cancel()

Kritik, sık gözden kaçan detay: backpressure pull-based’dir, push-based değil. Bir Subscriber, Publisher‘a şu anda tam olarak kaç item işleyebileceğini söylemek için request(n) çağırır. Bu, naif “producer olabildiğince hızlı push eder” modelini tersine çevirir ve hızlı producer’ların yavaş consumer’ları boğmasını engelleyen tek mekanizmadır — manifesto’nun “elastic” özelliğinin somut uygulaması.

2.3 Python’da Reactive Programming: RxPY

import reactivex as rx
from reactivex import operators as ops
from reactivex.scheduler import ThreadPoolScheduler

pool = ThreadPoolScheduler(max_workers=8)

source = rx.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

source.pipe(
    ops.filter(lambda x: x % 2 == 0),
    ops.map(lambda x: x * x),
    ops.buffer_with_count(2),
    ops.observe_on(pool),
).subscribe(
    on_next=lambda x: print(f"Alındı: {x}"),
    on_error=lambda e: print(f"Hata: {e}"),
    on_completed=lambda: print("Tamamlandı"),
)

Her author-level mühendisin isimlerini değil, semantiklerini ezbere bilmesi gereken operatör aileleri:

Kategori	Operatörler	Semantik
Oluşturma	`of`, `from_iterable`, `interval`, `create`	Bir stream’in nasıl başladığı
Dönüşüm	`map`, `scan`, `buffer`, `window`	Veriyi item-bazlı veya window-bazlı şekillendirmek
Filtreleme	`filter`, `distinct_until_changed`, `debounce`, `throttle_first`	Hacmi/gürültüyü azaltmak
Kombinasyon	`merge`, `zip`, `combine_latest`, `with_latest_from`	Çoklu stream kompozisyonu — bu dördü arasındaki farklar klasik bir mülakat tuzağıdır
Hata yönetimi	`catch`, `retry`, `on_error_resume_next`	Stream’in dışında değil, içinde hata kurtarma
Backpressure	`sample`, `buffer`, `throttle` (RxPY’nin backpressure primitive’leri JVM Rx’ten daha zayıftır — bkz. 2.4)

combine_latest vs zip vs with_latest_from ayrımı, tam olarak:

zip(a, b), yalnızca her iki stream de yeni, eşleşmiş bir item ürettiğinde emit eder (yavaş olanı bekler).
combine_latest(a, b), herhangi bir stream emit ettiğinde, diğerinin en son değerini kullanarak emit eder.
with_latest_from(a, b), yalnızca a emit ettiğinde, o anda b‘nin en son değerini örnekleyerek emit eder (asimetrik — b tek başına asla emisyonu tetiklemez).

2.4 RxPY’nin Dürüst Kısıtlaması

RxPY’nin backpressure hikayesi RxJava/Reactor’unkinden daha zayıftır, çünkü senkron Python iterable’ları üzerinden gerçek pull-based backpressure ile asyncio’nun push-based modeli, ekstra özen olmadan temiz bir şekilde birleşmez. Pratikte, gerçek backpressure’a ihtiyaç duyan production Python sistemleri şunlara dayanır:

asyncio.Queue(maxsize=N) açık put/get ile — sınırlı (bounded) bir queue zaten bir backpressure primitive’idir.
Kafka consumer poll döngüleri (Bölüm 3) — doğal backpressure mekanizması, hazır olana kadar poll()‘u tekrar çağırmamaktır.
Trio’nun memory channel’ları — sınırlı kanallar, gönderen tarafı bloklayarak backpressure’ı zorunlu kılar.

import asyncio

async def producer(queue: asyncio.Queue, n: int):
    for i in range(n):
        await queue.put(i)  # queue doluysa bloklar — bu backpressure'IN KENDİSİ
    await queue.put(None)  # sentinel

async def consumer(queue: asyncio.Queue):
    while (item := await queue.get()) is not None:
        await asyncio.sleep(0.1)  # yavaş işlemeyi simüle et
        queue.task_done()

async def main():
    q = asyncio.Queue(maxsize=10)
    await asyncio.gather(producer(q, 100), consumer(q))

Author-level çıkarım: reactive kütüphaneler (RxPY), event dönüşümlerini deklaratif şekilde kompoze etmek için kullanışlıdır; reactive mimari (message-driven, resilient, elastic sistemler) sadece Rx kütüphanesiyle değil, transport seçimin (Kafka), kuyruklama disiplinin ve backpressure-farkında consumer tasarımınla elde edilir.

2.5 Reactive vs. Async/Await — Hangisi Ne Zaman Kullanılır

Konu	asyncio/await	RxPY / Reactive Streams
Tek asenkron operasyon, tek sonuç	✅ doğal uyum	⚠️ gereğinden fazla
Birden fazla bağımsız event stream’ini kompoze etmek (UI event’leri, sensör verisi, market tick’leri)	⚠️ hantal, manuel fan-in	✅ doğal uyum
Karmaşık temporal operatörler (debounce, sliding window, throttle)	❌ elle yazılmış ve hataya açık	✅ yerleşik, iyi test edilmiş
Backpressure-kritik, yüksek throughput’lu pipeline’lar	⚠️ açık bounded queue gerektirir	⚠️ özen gerektirir (2.4) — genelde Kafka’nın kendi flow control’ü kazanır

3. Kafka ve Event-Driven Mimari

3.1 Çekirdek Mimari, Tam Olarak

Broker: bir Kafka sunucusu; bir cluster, KRaft (Kafka’nın kendi Raft-tabanlı metadata quorum’u, Kafka 4.0/geç 3.x sürümleri itibarıyla ZooKeeper’ın tamamen yerini almıştır) ile koordine edilen broker’lar kümesidir.
Topic: isimlendirilmiş, sadece-ekleme yapılan (append-only) bir log, partition’lara bölünmüştür.
Partition: paralellik, sıralama ve depolamanın gerçek birimi. Sıralama yalnızca partition içinde garantilidir, bir topic’in partition’ları arasında asla değil.
Offset: bir kaydın konumunu tanımlayan, monoton artan partition-başına sıra numarası.
Replication factor: her partition’ın broker’lar arasındaki kopya sayısı. Bir replika leader‘dır (tüm okuma/yazmaları işler); diğerleri ondan replike eden follower’lardır.
ISR (In-Sync Replicas): leader ile tam olarak senkron olan replika alt kümesi. Bir yazma, ancak gerekli sayıda ISR üyesi tarafından onaylandığında (acks ve min.insync.replicas tarafından yönetilir) “commit edilmiş” sayılır.

Topic: orders (3 partition, replication factor 3)

Partition 0: [Leader: Broker1] [Follower: Broker2] [Follower: Broker3]
Partition 1: [Leader: Broker2] [Follower: Broker3] [Follower: Broker1]
Partition 2: [Leader: Broker3] [Follower: Broker1] [Follower: Broker2]

3.2 Producer Semantiği — Tam Olarak

from confluent_kafka import Producer

conf = {
    "bootstrap.servers": "broker1:9092,broker2:9092",
    "acks": "all",                    # tüm in-sync replikaları bekle
    "enable.idempotence": True,       # retry'ları broker seviyesinde dedupe eder
    "max.in.flight.requests.per.connection": 5,  # idempotence açıkken 5'e kadar güvenli
    "compression.type": "zstd",
    "linger.ms": 10,                  # batch penceresi — throughput/latency ödünleşimi
    "batch.size": 65536,
    "retries": 2147483647,            # pratikte sonsuz; idempotence tekrarları önler
}

producer = Producer(conf)

def delivery_report(err, msg):
    if err is not None:
        print(f"Teslimat başarısız: {err}")
    else:
        print(f"Teslim edildi {msg.topic()}[{msg.partition()}]@{msg.offset()}")

producer.produce(
    topic="orders",
    key=str(order.customer_id).encode(),  # aynı key -> aynı partition -> sıralama korunur
    value=order.to_json().encode(),
    on_delivery=delivery_report,
)
producer.flush()

acks semantiği, tam olarak:

acks=0: fire-and-forget. Durability garantisi yok. En hızlı, güvensiz.
acks=1: sadece leader onayı. Leader, follower’lar replike etmeden önce çökerse veri kaybolabilir.
acks=all (ya da -1): leader tüm mevcut ISR üyelerini bekler. RF=3 üzerinde min.insync.replicas=2 ile birleştirildiğinde, sıfır veri kaybıyla bir broker arızasına tolerans sağlar.

Idempotent producer: her producer’a bir PID (producer ID) ve partition başına monoton bir sıra numarası atar, broker’ın tekrarlanan gönderimleri dedupe etmesine izin verir. Bu, retry’lı en-az-bir-kez‘i broker’da tam-olarak-bir-kez’e çevirir, ama tek başına sana uçtan-uca tam-olarak-bir-kez vermez (bkz. 3.5).

Partitioning stratejisi: varsayılan hash-partitioner, murmur2(key) % num_partitions kullanır. Aynı key → her zaman aynı partition → sıralama garantini sağlayan mekanizma budur. Key’leri bilinçli seç: customer-başına sıralama için customer_id, order-başına için order_id — herhangi bir sıralama semantiğine ihtiyacın varsa asla yüksek kardinaliteli rastgele değerler üzerinden partition’lama.

3.3 Consumer Semantiği — Tam Olarak

from confluent_kafka import Consumer, KafkaException

conf = {
    "bootstrap.servers": "broker1:9092,broker2:9092",
    "group.id": "order-processor",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,       # manuel commit = doğruluk kontrolü
    "isolation.level": "read_committed",  # yalnızca commit edilmiş transactional mesajları gör
    "max.poll.interval.ms": 300000,    # rebalance tetiklenmeden önceki zaman bütçesi
}

consumer = Consumer(conf)
consumer.subscribe(["orders"])

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            raise KafkaException(msg.error())

        process_order(msg.value())          # iş mantığı
        consumer.commit(msg, asynchronous=False)  # BAŞARILI işlemden SONRA commit et
finally:
    consumer.close()

Consumer group mekaniği:

Partition’lar, gruptaki consumer’lar arasında dağıtılır — her partition, herhangi bir anda grup içinde tam olarak bir consumer’a aittir.
Partition sayısının ötesinde consumer eklemek, fazlalıkları boşta bırakır. Partition sayısı, consumer paralelliğinin senin katı üst sınırındır.
Rebalancing, üyelik değişikliğinde (katılma/ayrılma/çökme) veya partition sayısı değişikliğinde oluşur. Tarihsel olarak “stop-the-world” (tüm consumer’lar durur), modern cooperative sticky rebalancing (partition.assignment.strategy=cooperative-sticky), yalnızca taşınması gereken belirli partition’ları yeniden atayarak kesintiyi minimize eder.

Offset commit stratejisi — gerçek doğruluk sorusu:

Strateji	Arıza modu
İşlemeden önce commit	En-fazla-bir-kez: commit’ten sonra, işlemeden önce çökme → mesaj kaybolur
İşledikten sonra commit (auto-commit kapalı)	En-az-bir-kez: işledikten sonra, commit’ten önce çökme → mesaj yeniden işlenir
Atomik commit + işleme (transactional outbox / Kafka transaction’ları)	Etkin-olarak-bir-kez: idempotent işleme veya Kafka transaction’ları gerektirir

Consumer’larını iş mantığı seviyesinde idempotent olacak şekilde tasarla (örn. order_id‘ye göre upsert, kör insert değil) — pratikte en-az-bir-kez teslimatı gerçekten güvenli kılan şey budur, ve uçtan uca tam-olarak-bir-kez peşinde koşmaktan çok daha sağlam bir mühendislik varsayılanıdır.

3.4 Schema Yönetimi

Kontrolsüz JSON payload’ları, ölçekte production Kafka olaylarının #1 nedenidir. Uyumluluk modlarını zorlayan Confluent Schema Registry‘yi Avro veya Protobuf ile kullan:

BACKWARD: yeni şema, eski şemayla yazılmış veriyi okuyabilir (önce consumer upgrade’leri için güvenli).
FORWARD: eski şema, yeni şemayla yazılmış veriyi okuyabilir (önce producer upgrade’leri için güvenli).
FULL: her iki yön de geçerlidir.

from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import SerializationContext, MessageField

sr_client = SchemaRegistryClient({"url": "http://schema-registry:8081"})

order_schema_str = """
{
  "type": "record",
  "name": "Order",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "created_at", "type": "long", "logicalType": "timestamp-millis"}
  ]
}
"""

avro_serializer = AvroSerializer(sr_client, order_schema_str)
serialized = avro_serializer(order.__dict__, SerializationContext("orders", MessageField.VALUE))

Kural: şema evrimi yalnızca default değerli opsiyonel alan ekleme veya zaten opsiyonel olan alanları kaldırma şeklinde olmalıdır. Bir alanı asla yerinde yeniden adlandırma veya tipini değiştirme — yeni bir alan ekle ve eskisini deprecate et.

3.5 Exactly-Once Semantics (EOS) — Gerçekte Ne Garanti Ediyor

Kafka’nın transactional API’si (isolation.level=read_committed + transactional producer), read-process-write desenleri için (örn. Kafka Streams, ya da consume-transform-produce döngüleri) Kafka ekosistemi içinde tam-olarak-bir-kez sağlar, keyfi harici sistemlere karşı değil.

from confluent_kafka import Producer, Consumer

producer = Producer({
    "bootstrap.servers": "broker1:9092",
    "transactional.id": "order-enricher-1",
})
producer.init_transactions()

consumer = Consumer({
    "bootstrap.servers": "broker1:9092",
    "group.id": "enricher",
    "isolation.level": "read_committed",
    "enable.auto.commit": False,
})
consumer.subscribe(["raw-orders"])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue

    producer.begin_transaction()
    try:
        enriched = enrich(msg.value())
        producer.produce("enriched-orders", value=enriched)
        # Consumer offset'lerini transaction'IN PARÇASI olarak gönder — kritik bağlantı
        producer.send_offsets_to_transaction(
            consumer.position(consumer.assignment()),
            consumer.consumer_group_metadata(),
        )
        producer.commit_transaction()
    except Exception:
        producer.abort_transaction()
        raise

Neden çalışır: offset commit’i ve produce, aynı atomik transaction‘ın parçası hâline gelir. Ya ikisi de olur ya da hiçbiri — Kafka-to-Kafka pipeline’ları için klasik “işlendi ama commit edilmedi” veya “commit edildi ama işlenmedi” boşluğunu ortadan kaldırır. O döngü içinde Kafka-olmayan bir sisteme (bir veritabanı, bir HTTP çağrısı) yazdığın anda, tekrar transactional outbox pattern’ine veya idempotent yazmalara ihtiyaç duyarsın, çünkü Kafka transaction’ları harici bir sistemin transaction sınırını kapsayamaz.

3.6 Kafka ile Event Sourcing ve CQRS

Event Sourcing: log, gerçeğin kaynağının kendisidir; state, event’leri replay/fold ederek türetilir. Compacted bir topic (cleanup.policy=compact), key başına yalnızca son değeri tutar — event-sourced bir varlığın “mevcut durum” projeksiyonu için mükemmel.
CQRS: write modelini (event ekleme) read model(ler)inden (materialized view’lar, genelde Kafka Streams ile veya Postgres/Elasticsearch’e yazan downstream bir consumer ile inşa edilir) ayırır.

# Bir read-model inşa eden minimal bir "projector" consumer
def project_order_events(consumer, db):
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        event = deserialize(msg.value())
        match event["type"]:
            case "OrderCreated":
                db.upsert_order(event["order_id"], status="created", **event["data"])
            case "OrderShipped":
                db.update_order_status(event["order_id"], status="shipped")
            case "OrderCancelled":
                db.update_order_status(event["order_id"], status="cancelled")
        consumer.commit(msg, asynchronous=False)

Transactional Outbox pattern (dual-write problemini çözer — “hem DB yazması hem Kafka publish’i gerçekten oldu mu?”):

İş satırını ve bir “outbox” event satırını aynı yerel veritabanı transaction’ında yaz.
Ayrı bir süreç (Debezium CDC connector veya polling bir publisher), outbox tablosunu okur ve Kafka’ya publish eder, sonra satırı publish edilmiş olarak işaretler.
Bu, DB ve Kafka arasında dağıtık bir transaction gerektirmeden, event’in ancak ve ancak DB transaction’ı commit edilmişse publish edilmesini garanti eder.

3.7 Faust — Python-Native Stream Processing

JVM yerine Python’da Kafka Streams tarzı bir topoloji için Faust, karşılaştırılabilir bir model sağlar:

import faust

app = faust.App("order-processor", broker="kafka://broker1:9092")

class Order(faust.Record, serializer="json"):
    order_id: str
    customer_id: str
    amount: float

orders_topic = app.topic("orders", value_type=Order)
high_value_topic = app.topic("high-value-orders", value_type=Order)

# bir table = changelog-destekli, partition'lanmış, hataya dayanıklı key-value store
customer_totals = app.Table("customer-totals", default=float)

@app.agent(orders_topic)
async def process(orders):
    async for order in orders:
        customer_totals[order.customer_id] += order.amount
        if order.amount > 10_000:
            await high_value_topic.send(value=order)

Faust table’ları bir changelog topic tarafından desteklenir — state hataya dayanıklıdır ve failover’da otomatik olarak yeniden inşa edilir, Kafka Streams’in KTable‘ıyla aynı temel fikirdir.

4. Distributed Systems Temelleri

4.1 CAP Teoremi, Tam Olarak (ve Neden Sıklıkla Yanlış Uygulanır)

Bir network partition‘ı (P) verildiğinde, bir sistem şunlar arasında seçim yapmalıdır:

Consistency (C): her okuma, en son yazmayı ya da bir hata döndürür.
Availability (A): her istek, en son yazma olduğu garantisi olmadan bir (hata-olmayan) yanıt alır.

Kritik nüans: CAP yalnızca gerçek bir partition sırasında geçerlidir. Partition koşulları dışında, aslında seçim yapmaya zorlanmazsın — bu yüzden PACELC, günlük tasarım için daha kullanışlı bir zihinsel modeldir:

Partition: A ya da C seç. Else (partition yoksa): Latency ya da Consistency seç.

DynamoDB veya Cassandra gibi sistemlerin “ayarlanabilir tutarlılıklı AP” olmasının nedeni budur — quorum parametreleri (R, W, N) aracılığıyla, partition olmasa bile latency/consistency ekseninde nerede durduğunu operasyon-başına seçmene izin verirler.

4.2 Tutarlılık Modelleri — Gerçek Spektrum

En güçlüden en zayıfa:

Linearizability: operasyonlar, çağrı ile yanıt arasındaki bir noktada, tek bir global gerçek-zamanlı sıraya tutarlı biçimde atomik olarak gerçekleşiyormuş gibi görünür. Pahalıdır — yazma başına tipik olarak consensus (Raft/Paxos) gerektirir.
Sequential consistency: tüm operasyonlar, her process’in kendi program sırasıyla tutarlı bir toplam sırada görünür — ama bu sıranın gerçek zamanla eşleşmesi gerekmez.
Causal consistency: nedensel olarak ilişkili operasyonlar (B, A‘nın yazdığı bir değeri okudu) herkes tarafından bu sırada görülür; eşzamanlı (ilişkisiz) operasyonlar farklı gözlemciler tarafından farklı sıralarda görülebilir.
Eventual consistency: yeni yazma olmadığında, tüm replikalar nihayetinde aynı değere yakınsar. Ara dönemde sıralama garantisi yoktur.

Author-level içgörü: pratikte “eventually consistent” sistemlerin çoğu aslında en azından causal consistency‘ye ihtiyaç duyar, aksi halde kullanıcılar bug algılar (“bir yorum yaptım, sonra bir yanıt geldi, ama başka biri yanıttan önce yorumu gördü”). Vector clock’ların / Lamport timestamp’lerinin “AP” sistemlerde bile önemli olmasının sebebi budur.

4.3 Mantıksal Saatler

Lamport timestamp’leri toplam (ama eşzamanlı event’ler için mutlaka nedensel olarak doğru olmayan) bir sıra sağlar:

class LamportClock:
    def __init__(self):
        self.time = 0

    def tick(self) -> int:
        self.time += 1
        return self.time

    def receive(self, other_time: int) -> int:
        self.time = max(self.time, other_time) + 1
        return self.time

Vector clock’lar, event başına O(n) alan maliyetiyle (n = node sayısı) gerçek nedenselliği yakalar:

class VectorClock:
    def __init__(self, node_id: str, nodes: list[str]):
        self.node_id = node_id
        self.clock = {n: 0 for n in nodes}

    def tick(self) -> dict[str, int]:
        self.clock[self.node_id] += 1
        return dict(self.clock)

    def merge(self, other: dict[str, int]):
        for node, t in other.items():
            self.clock[node] = max(self.clock.get(node, 0), t)
        self.tick()

    @staticmethod
    def concurrent(a: dict[str, int], b: dict[str, int]) -> bool:
        """Ne a <= b ne de b <= a ise event'ler eşzamanlıdır (gerçek bir çakışma)."""
        a_leq_b = all(a.get(k, 0) <= b.get(k, 0) for k in set(a) | set(b))
        b_leq_a = all(b.get(k, 0) <= a.get(k, 0) for k in set(a) | set(b))
        return not a_leq_b and not b_leq_a

Bu concurrent() kontrolü, Riak/Dynamo gibi sistemlerin uygulama seviyesinde veya CRDT-tabanlı çözünürlük gerektiren yazma çakışmalarını tam olarak nasıl tespit ettiğidir.

4.4 Consensus: Raft, Somut Olarak

Raft, consensus’u üç anlaşılabilir alt probleme ayrıştırır: leader election, log replication, safety.

Node’lar Follower, Candidate ya da Leader‘dır.
Zaman, her biri en fazla bir leader’a sahip term’lere bölünür.
Bir election timeout içinde (split vote’ları önlemek için rastgeleleştirilmiş) heartbeat duymayan bir follower, candidate olur, term’ini artırır ve oy ister.
Bir candidate, node’ların çoğunluğundan oy aldığında leader olur.
Leader, log’una entry ekler ve follower’lara replike eder; bir entry, çoğunluğa replike edildiğinde commit edilmiş sayılır — bu, Kafka’daki min.insync.replicas‘a doğrudan bağlanan safety çıpasıdır (KRaft’ın kendisi bir Raft implementasyonudur).

# Basitleştirilmiş leader-election çekirdeği (açıklayıcı, production-safe değil)
import random, time, enum

class Role(enum.Enum):
    FOLLOWER = "follower"
    CANDIDATE = "candidate"
    LEADER = "leader"

class RaftNode:
    def __init__(self, node_id: str, peers: list[str]):
        self.node_id = node_id
        self.peers = peers
        self.role = Role.FOLLOWER
        self.current_term = 0
        self.voted_for: str | None = None
        self.last_heartbeat = time.monotonic()
        self.election_timeout = random.uniform(0.15, 0.3)  # rastgele -> split vote'ları önler

    def on_election_timeout(self):
        if time.monotonic() - self.last_heartbeat < self.election_timeout:
            return
        self.role = Role.CANDIDATE
        self.current_term += 1
        self.voted_for = self.node_id
        votes = 1  # kendine oy
        for peer in self.peers:
            if self.request_vote(peer, self.current_term):
                votes += 1
        if votes > (len(self.peers) + 1) // 2:
            self.role = Role.LEADER

    def request_vote(self, peer: str, term: int) -> bool:
        ...  # peer'a RPC; term daha yeniyse ve log en az güncel ise peer oy verir

Bunu anlamak pratikte neden önemli: ister Kafka’nın KRaft controller quorum’unu, ister etcd’yi (Kubernetes tarafından kullanılır), ister Consul’u düşünüyor ol — production’da debug edeceğin arıza modları (network partition sırasında split-brain, okuma sunan bayat leader) doğrudan bu mekaniklere kadar iz sürer.

4.5 Dağıtık Transaction’lar: 2PC vs Saga

Two-Phase Commit (2PC): bir coordinator tüm katılımcılardan PREPARE yapmalarını ister; yalnızca hepsi evet oyu verirse COMMIT gönderir. Atomiklik garanti eder ama bloklayıcıdır — coordinator prepare’den sonra ama commit’ten önce çökerse, katılımcılar kilitleri süresiz tutar. Bu yüzden 2PC modern dağıtık/microservice sistemlerinde nadirdir.

Saga pattern: her biri, sonraki bir adım başarısız olursa geri almak için tanımlı bir compensating action‘a sahip bir dizi yerel transaction.

@dataclass
class SagaStep:
    action: Callable[[], None]
    compensation: Callable[[], None]

class Saga:
    def __init__(self, steps: list[SagaStep]):
        self.steps = steps
        self.completed: list[SagaStep] = []

    def execute(self):
        try:
            for step in self.steps:
                step.action()
                self.completed.append(step)
        except Exception:
            for step in reversed(self.completed):
                step.compensation()  # ters sırada geri al
            raise

saga = Saga([
    SagaStep(reserve_inventory, release_inventory),
    SagaStep(charge_payment, refund_payment),
    SagaStep(schedule_shipment, cancel_shipment),
])
saga.execute()

Choreography vs Orchestration Saga’lar:

Choreography: her servis, bir sonraki servisi tetikleyen event’leri (Kafka üzerinden) publish eder — merkezi bir coordinator yok, ama genel akışı izlemek/debug etmek daha zor.
Orchestration: merkezi bir saga orchestrator’ı her adımı açıkça çağırır ve compensation’ı yönetir — merkezi bir bağlaşım noktası pahasına, akıl yürütmesi ve gözlemlemesi daha kolay.

Ölçekte, saf choreography, saga ~4-5 adımı aştığında genelde durable bir state machine ile orchestration‘a (örn. Temporal.io, AWS Step Functions) yenilir, çünkü “bu saga hangi durumda?” sorusu saf event choreography ile operasyonel olarak cevaplanamaz hale gelir.

4.6 Sharding ve Replikasyon

Sharding stratejileri:

Range-based: basit, verimli range sorguları sağlar, ama hotspot riski taşır (örn. zamana göre sıralı key’lerin hepsi en yeni shard’a çarpması).
Hash-based: eşit dağılım, range sorgularını öldürür, consistent hashing kullanmadıkça node sayısı değişikliklerinde büyük resharding riski.
Consistent hashing: node’lar ve key’ler bir hash ring üzerindeki noktalara eşlenir; bir node eklemek/çıkarmak yalnızca key’lerin ~1/N‘ini yeniden eşler, hepsini değil. Virtual node’lar (fiziksel node başına birden fazla nokta), dengesiz yük dağılımını düzeltir.

import hashlib
import bisect

class ConsistentHashRing:
    def __init__(self, nodes: list[str], vnodes: int = 150):
        self.ring: dict[int, str] = {}
        self.sorted_keys: list[int] = []
        for node in nodes:
            for i in range(vnodes):
                key = self._hash(f"{node}#{i}")
                self.ring[key] = node
        self.sorted_keys = sorted(self.ring)

    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def get_node(self, key: str) -> str:
        if not self.ring:
            raise RuntimeError("Boş ring")
        h = self._hash(key)
        idx = bisect.bisect(self.sorted_keys, h) % len(self.sorted_keys)
        return self.ring[self.sorted_keys[idx]]

Replikasyon:

Leader-follower (primary-replica): basit, ama failover, sync vs async replikasyona bağlı olarak bir kesinti/veri kaybı penceresine sahiptir.
Multi-leader: yazmalar herhangi bir yerde kabul edilir, ama çakışma çözümü gerektirir (last-write-wins, CRDT’ler veya vector-clock tabanlı merge).
Leaderless (Dynamo-tarzı): okuma/yazmalar N replikaya gider, R + W > N quorum parametreleri okuma ve yazma kümeleri arasında örtüşme garanti eder — Cassandra/DynamoDB’nin ayarlanabilir tutarlılığının arkasındaki mekanizma budur.

4.7 Distributed Tracing ve Observability

Bir isteğin bir düzine servise ve arada bir Kafka topic’ine yayıldığı bir sistemde, distributed tracing (OpenTelemetry) opsiyonel bir araç değildir — nedenselliği sonradan yeniden inşa etmenin tek yoludur.

from opentelemetry import trace
from opentelemetry.propagate import inject, extract

tracer = trace.get_tracer(__name__)

# Producer tarafı: trace context'ini Kafka header'larına enjekte et
def produce_with_trace(producer, topic, key, value):
    with tracer.start_as_current_span("produce_order") as span:
        headers = {}
        inject(headers)  # W3C traceparent header'ı burada enjekte edilir
        producer.produce(
            topic, key=key, value=value,
            headers=[(k, v.encode()) for k, v in headers.items()],
        )

# Consumer tarafı: trace'i çıkar ve devam ettir
def consume_with_trace(msg):
    headers = {k: v.decode() for k, v in (msg.headers() or [])}
    ctx = extract(headers)
    with tracer.start_as_current_span("consume_order", context=ctx):
        process_order(msg.value())

Message broker boyunca header propagation olmadan, her trace producer’da ölür ve “bu sipariş neden uçtan uca 4 saniye sürdü” sorusunu cevaplama yeteneğini kaybedersin — Kafka tabanlı observability kurulumlarındaki en yaygın tek boşluk.

5. Resilience Mühendisliği

5.1 Resilience Araç Seti — Beş Pattern, Tam Olarak

5.1.1 Timeout

Her network çağrısının açık bir timeout’a ihtiyacı vardır — güvenli bir varsayılan “sonsuza kadar bekle” diye bir şey yoktur.

import httpx

async def fetch_with_timeout(url: str):
    async with httpx.AsyncClient(timeout=httpx.Timeout(connect=2.0, read=5.0, write=5.0, pool=2.0)) as client:
        return await client.get(url)

Connect timeout‘u (bir TCP bağlantısı bile kurabiliyor muyuz) read timeout’tan (sunucu zamanında yanıt veriyor mu) ayır — bunları tek bir sayıya karıştırmak, production’da gerçekte hangi arıza modunu gördüğünü gizler.

5.1.2 Exponential Backoff ve Jitter ile Retry

Naif retry fırtınaları, kademeli kesintilerin (cascading outage) önde gelen nedenidir: her client aynı sabit aralıkta retry yaptığında, zaten zorlanan bir servise karşı bir “thundering herd” içinde senkronize olurlar.

import random
import asyncio
from typing import TypeVar, Callable, Awaitable

T = TypeVar("T")

async def retry_with_backoff(
    fn: Callable[[], Awaitable[T]],
    max_attempts: int = 5,
    base_delay: float = 0.1,
    max_delay: float = 10.0,
) -> T:
    for attempt in range(max_attempts):
        try:
            return await fn()
        except (ConnectionError, TimeoutError):
            if attempt == max_attempts - 1:
                raise
            # Full jitter (AWS Architecture Blog'un önerdiği formül)
            delay = min(max_delay, base_delay * (2 ** attempt))
            jittered = random.uniform(0, delay)
            await asyncio.sleep(jittered)
    raise RuntimeError("buraya ulaşılamaz")

Yalnızca idempotent operasyonları retry et, ya da bir idempotency key ile idempotent hale getirilmiş operasyonları. Bir dedup key olmadan idempotent-olmayan bir ödeme tahsilatını retry etmek, müşterileri çift faturalandırmaya doğrudan giden bir yoldur.

5.1.3 Circuit Breaker

Bir client’ın, zaten başarısız olan bir servisi tekrar tekrar çağırmasını önler, ona toparlanma zamanı verir ve client’ı boşuna kaynak harcamaktan korur.

import time
import enum
from dataclasses import dataclass, field

class CircuitState(enum.Enum):
    CLOSED = "closed"       # normal işleyiş
    OPEN = "open"           # hızlı başarısız oluyor, downstream'e çağrı yok
    HALF_OPEN = "half_open" # downstream'in toparlanıp toparlanmadığını test ediyor

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout: float = 30.0
    half_open_max_calls: int = 3

    state: CircuitState = field(default=CircuitState.CLOSED)
    failure_count: int = 0
    last_failure_time: float = 0.0
    half_open_calls: int = 0

    def call(self, fn: Callable[[], T]) -> T:
        if self.state == CircuitState.OPEN:
            if time.monotonic() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
            else:
                raise CircuitOpenError("Circuit açık — hızlı başarısız oluyor")

        if self.state == CircuitState.HALF_OPEN and self.half_open_calls >= self.half_open_max_calls:
            raise CircuitOpenError("Half-open çağrı bütçesi tükendi")

        try:
            result = fn()
        except Exception:
            self._on_failure()
            raise
        else:
            self._on_success()
            return result

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.monotonic()
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN  # half-open'da tek bir hata yeniden açar
        elif self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

    def _on_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.half_open_calls += 1
            if self.half_open_calls >= self.half_open_max_calls:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
        else:
            self.failure_count = 0

class CircuitOpenError(Exception):
    pass

Half-open state, tam olarak: küçük, sınırlı sayıda deneme çağrısının geçmesine izin verir. Half-open sırasında tek bir hata, circuit’i anında yeniden açar — bu asimetri (yeniden açmaya hızlı, tamamen kapatmaya temkinli) kasıtlıdır ve marjinal, kararsız koşullar altında open ile closed arasında “flapping"i önleyen tam olarak budur.

5.1.4 Bulkhead İzolasyonu

Gemi bölmelendirmesinden adını alır — bir yavaş/başarısız bağımlılığın, ilgisiz çağrıların ihtiyaç duyduğu kaynakları tüketememesi için her downstream bağımlılık başına kaynak havuzlarını (thread pool’lar, connection pool’lar, semaphore’lar) izole et.

import asyncio

class Bulkhead:
    def __init__(self, max_concurrent: int, max_queue: int):
        self._sem = asyncio.Semaphore(max_concurrent)
        self._queue_limiter = asyncio.Semaphore(max_concurrent + max_queue)

    async def execute(self, fn: Callable[[], Awaitable[T]]) -> T:
        if not self._queue_limiter.locked() and self._queue_limiter._value <= 0:
            raise BulkheadFullError("Bulkhead kuyruğu tükendi")
        async with self._queue_limiter:
            async with self._sem:
                return await fn()

class BulkheadFullError(Exception):
    pass

# Downstream bağımlılık başına bir bulkhead — tüm mesele bu
payment_bulkhead = Bulkhead(max_concurrent=10, max_queue=20)
inventory_bulkhead = Bulkhead(max_concurrent=20, max_queue=40)

Bulkhead’ler olmadan, tek bir yavaş downstream bağımlılık, paylaşılan bir thread/connection pool’unun %100’ünü tüketebilir ve sağlıklı bağımlılıklara giden istekleri aç bırakabilir — “kademeli arıza” başlıklı postmortem’lerdeki en yaygın kök neden etiketi budur.

5.1.5 Rate Limiting

Senin sistemini aşırı yüklenmekten korur (aşırı yüklenmiş bir downstream‘den caller’ları koruyan circuit breaker’ların aksine).

import time
import threading

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate            # saniyede eklenen token
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = time.monotonic()
        self._lock = threading.Lock()

    def allow(self, cost: int = 1) -> bool:
        with self._lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last_refill = now
            if self.tokens >= cost:
                self.tokens -= cost
                return True
            return False

Token bucket vs. sliding window vs. fixed window: token bucket, uzun vadeli ortalama oranı zorlarken capacity‘ye kadar kontrollü patlamalara izin verir — genelde doğru varsayılan. Fixed window en basitidir ama pencere sınırlarında 2x patlamaya izin verir (bilinmesi değerli, nadiren kullanılması değerli). Sliding window log kesindir ama key başına O(n) bellek gerektirir; sliding window counter bunu O(1)‘de yaklaşık olarak sağlar.

5.2 Pattern’ları Birleştirmek — Gerçek Production Yığını

Bu pattern’lar alternatif değildir; dayanıklı bir çağrı yolu hepsini şu sırayla uygulanmış şekilde kompoze eder:

Rate Limiter (kendini koru)
  → Bulkhead (bu bağımlılığın kaynak havuzunu izole et)
    → Circuit Breaker (bu bağımlılık sağlıksızsa hızlı başarısız ol)
      → Timeout (herhangi bir çağrının ne kadar sürebileceğini sınırla)
        → Retry (backoff+jitter ile) (geçici arızadan toparlan)
          → [gerçek network çağrısı]

async def resilient_call(fn: Callable[[], Awaitable[T]], breaker: CircuitBreaker, bulkhead: Bulkhead) -> T:
    async def with_timeout():
        return await asyncio.wait_for(fn(), timeout=5.0)

    async def with_retry():
        return await retry_with_backoff(with_timeout, max_attempts=3)

    async def with_breaker():
        return breaker.call(lambda: asyncio.get_event_loop().run_until_complete(with_retry()))
        # (açıklayıcı — gerçek async kodda, async-native bir breaker implementasyonu kullan)

    return await bulkhead.execute(with_retry)

Kritik bir sıralama kuralı: retry, circuit breaker’ın hata sayımının dışında değil, içinde olmalıdır — aksi halde her retry denemesi, mantıksal çağrı başına breaker’ın hata sayacını bağımsız olarak birden çok kez tetikler, bu da gerçek downstream sağlığına göre çok daha agresif bir şekilde açılmasına neden olur.

5.3 Zarafetle Bozulma (Graceful Degradation) ve Fallback’ler

Resilience yalnızca başarısız olmamak değildir — faydalı şekilde başarısız olmakla ilgilidir.

async def get_recommendations(user_id: str) -> list[Recommendation]:
    try:
        return await ml_recommendation_service.get(user_id, timeout=0.3)
    except (TimeoutError, CircuitOpenError):
        # Fallback: daha ucuz, cache'lenmiş, "yeterince iyi" bir yanıta düş
        return await get_popular_items_cached()

Fallback stratejilerini bozulma maliyetine göre sırala:

Bayat cache’i sun (en iyisi: neredeyse tam işlevsellik, biraz eski veri).
Daha ucuz/basit hesaplanmış bir sonuç sun (kısmi işlevsellik).
Statik bir varsayılan sun (minimal işlevsellik, her zaman kullanılabilir).
Tüm isteği değil, o özelliği başarısız kıl (blast radius’u izole et — örn. tüm sayfayı 500’lemek yerine, sayfayı öneri widget’ı olmadan render et).

5.4 Chaos Engineering

Sistemini enjekte edilmiş arıza altında doğrulamadan dayanıklı olduğunu bilemezsin — chaos testi olmayan resilience pattern’ları doğrulanmamış varsayımlardır.

İlkeler (Principles of Chaos Engineering’den):

Steady state‘i ölçülebilir bir çıktı olarak tanımla (dahili metrikler değil — başarılı checkout oranı gibi iş/kullanıcı-yönelimli metrikler).
Steady state’in hem kontrol hem deneysel grupta geçerli olduğunu hipotez et.
Gerçek dünya event’leri enjekte et: broker/node arızası, latency enjeksiyonu, network partition, kaynak tükenmesi, clock skew.
Hipotezi çürütmeye çalış — amaç, zayıflıkları production’da seni bulmadan önce bulmaktır.
Blast radius’u minimize et: staging’de başla, sonra production trafiğinin ufak bir yüzdesi, güven arttıkça artır.

# Test ortamında chaos testi için minimal bir latency/failure enjeksiyon wrapper'ı
import random

class ChaosMiddleware:
    def __init__(self, failure_rate: float = 0.0, latency_ms: tuple[int, int] = (0, 0)):
        self.failure_rate = failure_rate
        self.latency_ms = latency_ms

    async def __call__(self, fn: Callable[[], Awaitable[T]]) -> T:
        if self.latency_ms != (0, 0):
            await asyncio.sleep(random.uniform(*self.latency_ms) / 1000)
        if random.random() < self.failure_rate:
            raise ConnectionError("Chaos ile enjekte edilmiş arıza")
        return await fn()

Production-grade araçlar: Kubernetes-native fault injection için Chaos Mesh ve LitmusChaos; yönetilen chaos-as-a-service için Gremlin; özellikle Kafka için, load test sırasında broker’ları/partition leader’larını öldürmek, çoğu ekibin asla çalıştırmadığı en yüksek değerli tek chaos deneyidir.

5.5 Resilience Ön Koşulu Olarak Observability

Gözlemleyemediğin bir sisteme resilience inşa edemezsin. Üç sütun, ve her birinin gerçekte ne için olduğu:

Metrikler (Prometheus/Grafana): toplu trendler — “hata oranı yükseliyor mu,” “p99 latency bozuluyor mu.” Depolaması ucuz, tasarım gereği kayıplı (istek-başına detay yok).
Log’lar (structured, trace ID ile ilişkilendirilmiş): bir şeyin yanlış gittiğini bildiğinde belirli bir istek için “tam olarak ne oldu” detayı.
Trace’ler (OpenTelemetry, Bölüm 4.7): “dağıtık çağrı grafiğinde nerede yanlış gitti” — servisler-arası nedenselliği yeniden inşa eden tek sütun.

Herhangi bir servis için golden signal’lar (Google SRE kitabı): Latency, Traffic, Errors, Saturation. Production’daki bir Kafka consumer’ı için her dashboard’un en az şunlara ihtiyacı vardır: consumer lag (partition başına), işleme latency’si, hata/DLQ oranı ve rebalance sıklığı — özellikle rebalance sıklığı, çünkü sık rebalancing’in kendisi bir resilience kırmızı bayrağıdır (genelde yavaş işleme nedeniyle aşılan max.poll.interval.ms‘den kaynaklanır, bu da orijinal sorunu daha da kötüleştirir).

5.6 Dead Letter Queue’lar ve Zehirli Mesaj Yönetimi

Tek bir hatalı biçimlendirilmiş mesaj, tüm bir consumer group’unu asla durduramamalıdır.

def process_with_dlq(consumer, producer, dlq_topic: str, max_retries: int = 3):
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue

        retry_count = get_retry_count(msg.headers())
        try:
            process_order(msg.value())
            consumer.commit(msg, asynchronous=False)
        except PoisonPillError as e:
            if retry_count >= max_retries:
                producer.produce(
                    dlq_topic,
                    key=msg.key(),
                    value=msg.value(),
                    headers=[
                        ("original_topic", msg.topic().encode()),
                        ("original_partition", str(msg.partition()).encode()),
                        ("original_offset", str(msg.offset()).encode()),
                        ("error", str(e).encode()),
                        ("retry_count", str(retry_count).encode()),
                    ],
                )
                producer.flush()
                consumer.commit(msg, asynchronous=False)  # zehirli mesajı geç
            else:
                increment_retry_and_requeue(msg)

DLQ header’larında tam kaynak bilgisini (original_topic, original_partition, original_offset, hata detayı) koru — bu olmadan, DLQ mesajları kurtarılabilir bir kuyruk değil, yeniden üretilemez, debug edilemez bir mezarlık haline gelir.

6. Hepsini Birleştirmek: Referans Mimari

Python’da dayanıklı, reactive, event-driven bir sipariş işleme pipeline’ı:

[API Gateway] --HTTP--> [Order Service]
                              |
                    (1. Bir transaction içinde yerel DB'ye yaz)
                    (2. Aynı transaction'da outbox satırı yaz)
                              |
                    [Debezium CDC / Outbox Publisher]
                              |
                              v
                    Kafka Topic: orders.created (3 partition, RF=3, key=customer_id)
                         /              \
                        /                \
        [Inventory Consumer Group]   [Notification Consumer Group]
         (idempotent upsert,          (idempotent, rate-limited,
          inventory DB'ye circuit      bulkhead-izole edilmiş
          breaker, DLQ)                email/SMS provider çağrıları)
                |
                v
    Kafka Topic: orders.inventory-reserved
                |
                v
     [Payment Saga Orchestrator] --çağırır--> [Payment Service]
        (payment çağrısında circuit             (order_id başına
         breaker + retry+jitter +                idempotency key)
         timeout + bulkhead)
                |
        başarı -> orders.completed
        başarısızlık -> compensating action'lar -> orders.cancelled

Bu mimarinin kodladığı tasarım kararları:

Transactional outbox (3.6), order DB’si ile Kafka arasındaki dual-write problemini ortadan kaldırır.
customer_id‘ye göre partition’lama (3.2), genel paralellikten ödün vermeden customer-başına event sıralamasını korur.
Her consumer group bağımsız olarak ölçeklenebilir, partition sayısıyla sınırlıdır (3.3) — inventory ve notification işleme bağımsız olarak ölçeklenir.
Her consumer’da idempotent işleme (3.3, 5.1.2), en-az-bir-kez teslimatı güvenli kılar, tam uçtan-uca tam-olarak-bir-kez ihtiyacını atlar.
Çoklu servis payment/inventory/shipment transaction’ı için 2PC değil, Saga orchestration (4.5).
Her harici çağrı tam resilience yığınına sarılmış (5.2): rate limiter → bulkhead → circuit breaker → timeout → retry.
Her consumer sınırında DLQ’lar (5.6), zehirli mesajların işlemeyi durdurmasını önler.
Kafka boyunca yayılan distributed tracing header’ları (4.7), tüm sipariş yaşam döngüsünün — HTTP isteğinden üç consumer group’a kadar — tek bir trace olarak yeniden inşa edilebilir olmasını sağlar.

7. İleri Okuma

Designing Data-Intensive Applications — Martin Kleppmann (mühendisler için tek en iyi distributed systems kitabı)
Kafka: The Definitive Guide — Neha Narkhede, Gwen Shapira, Todd Palino
Reactive Manifesto — reactivemanifesto.org
Reactive Streams spesifikasyonu — reactive-streams.org
Google SRE Kitabı — sre.google/books
Raft makalesi: “In Search of an Understandable Consensus Algorithm” — Ongaro & Ousterhout
AWS Architecture Blog: “Exponential Backoff and Jitter”
Principles of Chaos Engineering — principlesofchaos.org
PEP 703 (GIL’i opsiyonel yapmak), PEP 654 (exception group’lar), PEP 492/PEP 525 (async generator’lar)

Bu doküman, ekosistemin 2026’nın erken-orta dönemindeki durumunu yansıtır. Kafka tamamen KRaft’a geçmiştir (ZooKeeper kaldırıldı); Python’ın free-threaded build’i hâlâ deneyseldir; sürüme özgü detayları değişebilir olarak ele al ve production kararlarından önce güncel release notlarıyla doğrula.