Jak ładować miliony rekordów na sekundę do ClickHouse przy użyciu klienta Java

Próba wstrzyknięcia miliona zdarzeń na sekundę do bazy analitycznej przy użyciu standardowego podejścia opartego na klasycznym sterowniku JDBC zazwyczaj kończy się gwałtownym załamaniem wydajności. Wątki aplikacji blokują się na puli połączeń, zużycie pamięci w maszynie wirtualnej Java (JVM) szybuje w górę przez ciągłą alokację obiektów pośrednich, a serwer ClickHouse odpowiada lawiną błędów DB::Exception: Too many parts in all data in table. ClickHouse to silnik kolumnowy zoptymalizowany pod kątem liniowego skanowania olbrzymich wolumenów danych, a nie transakcyjnego przetwarzania pojedynczych krotek. Osiągnięcie przepustowości na poziomie od jednego do kilku milionów rekordów na sekundę z poziomu aplikacji w Javie wymaga odejścia od tradycyjnych wzorców na rzecz inżynierii niskopoziomowej: bezpośredniej serializacji binarnej, buforowania zoptymalizowanego pod kątem mechanizmu Garbage Collector oraz świadomego projektowania topologii sieciowej.

Architektura MergeTree a fizyka zapisu w ClickHouse

Zanim w kodzie Javy zostanie zainicjowane jakiekolwiek połączenie sieciowe, konieczne jest zrozumienie fizycznych ograniczeń silnika MergeTree. ClickHouse nie modyfikuje istniejących danych w miejscu i nie zapisuje pojedynczych wierszy w strukturach blokowych tak jak relacyjne bazy OLTP. Każda operacja INSERT tworzy na dysku nowy, niezależny fragment danych, nazywany partem (part). Fragment ten zawiera posortowane dane kolumnowe, skompresowane algorytmem LZ4 lub ZSTD, wraz z indeksami rzadkimi (sparse primary index) oraz plikami znaków kontrolnych.

W tle działania bazy nieustannie pracuje proces scalania (merge), który łączy mniejsze party w większe, eliminując duplikaty lub agregując dane zgodnie z definicją silnika (np. ReplacingMergeTree, SummingMergeTree). Jeśli aplikacja kliencka wysyła setki małych zapytań na sekundę, tempo powstawania nowych partów drastycznie przewyższa możliwości wątków odpowiedzialnych za scalanie. System w ramach samoobrony zaczyna najpierw opóźniać odpowiedzi (sleep), a ostatecznie odrzuca kolejne zapytania, rzucając błędem uniemożliwiającym dalszy zapis.

Fizyczna struktura partów i problem fragmentacji dyskowej

Tworzenie zbyt małych partów skutkuje degradacją podsystemu I/O na poziomie systemu operacyjnego. Zamiast sekwencyjnego zapisu dużych strumieni danych o rozmiarach rzędu 50–100 MB, kontroler dyskowy zostaje zasypany operacjami synchronicznego tworzenia tysięcy mikroskopijnych plików metadanych i kolumn. Każdy part dla tabeli posiadającej kilkadziesiąt kolumn oznacza wygenerowanie przynajmniej kilkudziesięciu plików na dysku. Przy dużej skali prowadzi to do wyczerpania tablicy i-węzłów (inodes) w systemie plików ext4 lub XFS oraz do zablokowania mechanizmu kompresji, który nie jest w stanie uzyskać optymalnego stopnia upakowania danych przy małych próbkach.

Punkt kontrolny: Minimalny wolumen pojedynczego zapisu (batch) trafiającego do silnika MergeTree nie powinien być mniejszy niż 10 000 wierszy, przy czym docelowy, optymalny przedział dla masowych pipeline’ów analitycznych mieści się w granicach od 100 000 do 250 000 wierszy lub około 50–100 MB surowych danych w jednym żądaniu.

Błąd „Too many parts”: punkt krytyczny niewłaściwego zapisu

Pojawienie się wyjątku Too many parts in all data in table to definitywny sygnał ostrzegawczy świadczący o architektonicznym błędzie po stronie producenta danych. ClickHouse posiada domyślny limit liczby aktywnych, niescalonych partów w ramach pojedynczej partycji (parametr parts_to_throw_insert, standardowo ustawiony na 300). Zanim limit ten zostanie osiągnięty, silnik aktywuje mechanizm dławienia (backpressure) przy parametrze parts_to_delay_insert (domyślnie 150), sztucznie wprowadzając opóźnienia w obsłudze żądania HTTP lub TCP.

Aplikacja w Javie, która nie monitoruje tych opóźnień, gwałtownie degraduje swoją przepustowość: pule połączeń zostają zablokowane na oczekiwaniu na odpowiedź serwera, co kaskadowo prowadzi do wyczerpania pamięci podręcznej JVM. Podnoszenie wartości tych parametrów po stronie konfiguracji bazy bez modyfikacji wzorca zapisu w Javie jest błędem krytycznym – jedynie odsuwa w czasie moment awarii serwera bazodanowego z powodu wyczerpania pamięci RAM lub przeciążenia procesora ciągłymi próbami scalania milionów małych plików.

Klient-side batching kontra mechanizm async_insert na serwerze

Nowsze wersje ClickHouse oferują opcję async_insert = 1, która pozwala serwerowi zbierać drobne zapytania w pamięci podręcznej i scalać je w duże bloki przed fizycznym zrzutem na dysk. Choć rozwiązanie to jest przydatne w architekturach mikroserwisowych, gdzie setki rozproszonych aplikacji generują sporadyczne zdarzenia, przy celowaniu w przepustowość na poziomie milionów rekordów na sekundę poleganie na serwerowym async_insert staje się wąskim gardłem.

Serwer ClickHouse musi w takim scenariuszu alokować dodatkowe zasoby CPU na parsowanie milionów oddzielnych żądań HTTP lub pakietów TCP, ekstrakcję nagłówków i zarządzanie stanem kolejki w pamięci RAM. W profesjonalnym scenariuszu Big Data wyłączną odpowiedzialność za agregację i pakowanie danych musi przejąć warstwa aplikacji Java. Mechanizm async_insert na serwerze powinien być traktowany jedynie jako pas bezpieczeństwa, a nie podstawowy filar architektury ingestu.

Jeśli podsystem I/O bazy notuje tysiące operacji zapisu na sekundę przy relatywnie niskim transferze MB/s, oznacza to wadliwą granulację paczek po stronie klienta Java.

Wybór klienta Java: oficjalny ClickHouse Java Client V2 czy JDBC?

Decyzja o doborze biblioteki klienckiej rzutuje bezpośrednio na profil zużycia zasobów procesora i pamięci aplikacji generującej obciążenie. Tradycyjne środowiska korporacyjne domyślnie sięgają po specyfikację JDBC. W kontekście ClickHouse i obciążeń rzędu milionów wierszy na sekundę wybór ten wiąże się jednak z potężnym narzutem wydajnościowym.

Specyfikacja JDBC zmusza aplikację do opakowywania każdego prostego typu w dedykowany obiekt Javy, co przy próbie załadowania 2 000 000 wierszy o strukturze zawierającej 15 kolumn generuje 30 milionów krótkotrwałych instancji obiektów na sekundę. W takim środowisku odśmiecacz pamięci (Garbage Collector), niezależnie od tego, czy używany jest G1, czy ZGC, zaczyna konsumować znaczną część czasu procesora na czyszczenie sterty, wywołując mikro-pauzy, które destabilizują ciągłość strumienia sieciowego.

Nowy, oficjalny ClickHouse Java Client V2 rozwiązuje ten problem fundamentalnie. Zaprojektowany od podstaw jako lekki klient HTTP/TCP operuje na strumieniach surowych bajtów i pozwala na bezpośrednie przekazywanie danych bez konieczności tworzenia warstwy abstrakcji dla każdego rekordu. W połączeniu z kompresją LZ4 realizowaną po stronie procesów JVM stanowi on jedyny racjonalny wybór dla systemów klasy Ultra-High Throughput.

Format ma znaczenie: przewaga RowBinary i Native nad formatami tekstowymi

Wielu inżynierów popełnia błąd, wybierając format JSONEachRow lub CSV z uwagi na łatwość ich debugowania. Weryfikacja surowego tekstu w narzędziach diagnostycznych jest wygodna, ale z perspektywy wydajnościowej formaty tekstowe są katastrofalne. Generowanie tekstu w Javie wymaga konwersji liczb zmiennoprzecinkowych, dat i 64-bitowych liczb całkowitych na reprezentację znakową UTF-8. Po drugiej stronie serwer ClickHouse musi ten strumień sparsować, walidując każdy przecinek, cudzysłów i znak ucieczki.

Operacje parsowania tekstu w pełni obciążają rdzenie CPU serwera bazy, zanim dane dotkną podsystemu pamięci masowej. Zastosowanie formatu RowBinary lub RowBinaryWithNamesAndTypes eliminuje ten narzut niemal do zera. W formacie tym liczby całkowite zapisywane są bezpośrednio w porządku Little-Endian, ciągi znaków poprzedzone są zmiennej długości nagłówkiem rozmiaru (LEB128), a wartości logiczne zajmują dokładnie jeden bajt. ClickHouse pobiera taki strumień i bezpośrednio mapuje go do bloków pamięci kolumnowej bez konieczności wykonywania złożonych operacji leksykalnych.

W skrajnie wymagających scenariuszach optymalnym rozwiązaniem staje się format Native. W przeciwieństwie do RowBinary, który serializuje dane wiersz po wierszu, format Native operuje bezpośrednio na blokach kolumnowych. Wymaga to jednak od aplikacji w Javie zorganizowania bufora w pamięci w układzie wektorowym (przechowywanie tablic dla całych kolumn zamiast obiektów reprezentujących rekordy). Przesłanie tak przygotowanego bloku do ClickHouse sprawia, że silnik bazy wykonuje jedynie prosty transfer DMA na dysk z pominięciem jakiejkolwiek wewnętrznej transpozycji z wierszy na kolumny.

Projektowanie bezalokacyjnego potoku w pamięci JVM

Osiągnięcie stabilnego transferu na poziomie kilku milionów zdarzeń na sekundę wymaga podejścia zbliżonego do inżynierii systemów high-frequency trading (HFT). Pamięć JVM nie może być traktowana jako nieskończony zasób, a tworzenie obiektów reprezentujących pojedyncze zdarzenia biznesowe na ścieżce zapisu musi zostać zredukowane do absolutnego minimum.

Klasyczny wzorzec z kolejką LinkedBlockingQueue<Event> natychmiast uderza w barierę blokowania synchronizatorów wątkowych i ciągłej alokacji węzłów pamięci. Zamiast tego bufor aplikacyjny powinien bazować na pierścieniu buforowym o stałym rozmiarze (wzorzec LMAX Disruptor) lub na podwójnie buforowanych strukturach operujących na bezpośrednich buforach pamięci poza stertą (off-heap DirectByteBuffer). W takim układzie wątki odbierające dane ze źródeł (np. klastra Apache Kafka) nie tworzą nowych struktur, lecz bezpośrednio serializują pola binarne do prealokowanego bufora bajtowego.

Gdy bieżący bufor osiągnie próg 100 000 rekordów lub upłynie określony czas (np. 200 ms), referencja do niego jest przekazywana do wątku I/O odpowiedzialnego za wykonanie żądania sieciowego, a wątki robocze natychmiast przełączają się na zapasowy, wyczyszczony bufor. Proces ten nie generuje ani jednego nowego obiektu na stercie Javy podlegającego czyszczeniu przez GC, co stabilizuje czasy odpowiedzi aplikacji i chroni przed zjawiskiem tzw. tail latency, która mogłaby doprowadzić do przepełnienia buforów wejściowych.

Implementacja produkcyjna z wykorzystaniem ClickHouseClient V2

Poniższa implementacja demonstruje skonfigurowanie oficjalnego klienta Java do pracy z bezpośrednim strumieniowaniem binarnym z włączoną kompresją LZ4 w locie. Wykorzystanie interfejsu strumieniowego pozwala aplikacji przekazać strumień wyjściowy bezpośrednio do warstwy transportowej HTTP bez materializacji całego bloku w pamięci jako pojedynczej tablicy bajtów.

package com.example.clickhouse.ingest;

import com.clickhouse.client.api.Client;
import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatWriter;
import com.clickhouse.data.ClickHouseFormat;
import com.clickhouse.data.format.BinaryStreamUtils;

import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.concurrent.CompletableFuture;

public class HighThroughputIngestionPipeline {

    private final Client client;
    private final String insertQuery;

    public HighThroughputIngestionPipeline(String host, int port, String database, String table) {
        this.client = new Client.Builder()
                .addEndpoint("http://" + host + ":" + port)
                .setUsername("default")
                .setPassword("secret")
                .setDefaultDatabase(database)
                .compressClientRequest(true) // Kompresja LZ4 strumienia wychodzącego z poziomu JVM
                .setSocketTimeout(30_000)
                .setConnectionTimeout(5_000)
                .setMaxConnections(32)
                .build();

        this.insertQuery = "INSERT INTO " + table + " SETTINGS async_insert=0 FORMAT RowBinary";
    }

    public CompletableFuture<Void> streamBatch(EventIterator events, int batchSize) {
        return CompletableFuture.runAsync(() -> {
            try {
                client.loadData(insertQuery, outputStream -> {
                    writeRowBinaryData(events, outputStream, batchSize);
                }, ClickHouseFormat.RowBinary).get();
            } catch (Exception e) {
                throw new RuntimeException("Krytyczny blad przesylu danych do ClickHouse", e);
            }
        });
    }

    private void writeRowBinaryData(EventIterator events, OutputStream out, int count) {
        try {
            int written = 0;
            while (events.hasNext() && written < count) {
                Event event = events.next();

                // 1. UInt64 - Identyfikator zdarzenia
                BinaryStreamUtils.writeInt64(out, event.getId());

                // 2. DateTime64(3) - Precyzyjny znacznik czasu jako int64 milisekund
                BinaryStreamUtils.writeInt64(out, event.getTimestampEpochMillis());

                // 3. LowCardinality(String) - Zapisywany jak zwykly String (LEB128 dlugosc + bajty UTF-8)
                byte[] tenantBytes = event.getTenantId().getBytes(StandardCharsets.UTF_8);
                BinaryStreamUtils.writeVarInt(out, tenantBytes.length);
                out.write(tenantBytes);

                // 4. Float64 - Pomiar metryki
                BinaryStreamUtils.writeFloat64(out, event.getValue());

                // 5. UInt8 - Flaga bitowa (symulacja boolean)
                BinaryStreamUtils.writeInt8(out, (byte) (event.isValid() ? 1 : 0));

                written++;
            }
            out.flush();
        } catch (Exception e) {
            throw new RuntimeException("Blad serializacji binarnej RowBinary", e);
        }
    }
}

Kluczowym elementem powyższej konfiguracji jest dyrektywa compressClientRequest(true). Przesyłanie niezaszyfrowanych, surowych bloków binarnych przez sieć 10GbE lub 25GbE przy wolumenach rzędu setek megabajtów na sekundę potrafi szybko nasycić przepustowość interfejsu sieciowego lub pasma chmurowego (np. AWS EBS/Network limits). Zastosowanie algorytmu LZ4, który charakteryzuje się asymetrycznie niskim kosztem kompresji po stronie CPU w stosunku do stopnia redukcji objętości, pozwala zredukować ruch sieciowy od 3 do nawet 8 razy przy marginalnym narzucie na wątki JVM.

Omijanie Distributed tables: bezpośredni routing do węzłów klastra

W klastrach ClickHouse składających się z wielu maszyn powszechnym wzorcem jest korzystanie z tabel opartych na silniku Distributed. Tabela tego typu nie przechowuje danych fizycznie, lecz pełni rolę wewnętrznego proxy: przyjmuje zapytanie, oblicza funkcję skrótu na kluczu shardingowym i przesyła dane sieciowo do właściwego węzła zawierającego docelową tabelę MergeTree.

Przy próbie osiągnięcia ekstremalnej przepustowości zapisywanie danych do tabeli rozproszonej staje się barierą nie do przebicia. Węzeł pośredniczący musi odebrać cały strumień, zdekompresować go, podzielić w pamięci na pakiety przeznaczone dla każdego z shardów w klastrze, ponownie skompresować i wysłać kolejnymi połączeniami sieciowymi. W rezultacie obciążenie sieciowe ulega podwojeniu (double network hop), a wątki jednego węzła koordynującego ulegają wysyceniu.

W architekturze zorientowanej na miliony zdarzeń to aplikacja kliencka w Javie musi posiadać pełną wiedzę o topologii klastra. Warstwa ingestu powinna pobrać listę aktywnych instancji z tabeli systemowej system.clusters, a następnie samodzielnie realizować podział na partycje przy użyciu tego samego algorytmu, który zdefiniowano w bazie (np. murmurHash3_64 lub cityHash64). Następnie aplikacja utrzymuje niezależne pule buforów dla każdego fizycznego noda i streamuje dane bezpośrednio do lokalnych tabel MergeTree, całkowicie odciążając serwery ClickHouse z roli routerów sieciowych.

Konfiguracja jądra systemu i JVM pod kątem ekstremalnego I/O

Nawet doskonale napisany kod w Javie nie zdoła nasycić bazy danymi, jeśli środowisko uruchomieniowe JVM oraz stos sieciowy systemu Linux będą działać na domyślnych parametrach. Po stronie maszyny wirtualnej krytyczne jest odizolowanie wątków serializujących od operacji odśmiecania pamięci poprzez konfigurację nowoczesnego garbage collectora o niskich czasach zatrzymań:

-XX:+UseZGC -XX:+ZGenerational -XX:SoftMax

HeapSize=12g -Xms16g -Xmx16g -XX:+AlwaysPreTouch

Wprowadzenie generacyjnego ZGC (dostępnego od Java 21) niemal całkowicie eliminuje mikroprzestoje aplikacji, utrzymując pauzy na poziomie submilisekundowym nawet przy dynamicznym przepływie gigabajtów danych poza główną stertę. Opcja -XX:+AlwaysPreTouch zmusza z kolei system operacyjny do fizycznego zmapowania stron pamięci RAM już w trakcie inicjalizacji procesu, co zapobiega powstawaniu opóźnień typu page fault w momentach gwałtownego wzrostu obciążenia buforów I/O.

Równie istotna jest rekonfiguracja warstwy sieciowej Linuksa na serwerach aplikacyjnych oraz maszynach bazodanowych. Domyślne bufory gniazd TCP w większości dystrybucji zostały zaprojektowane z myślą o uniwersalnych usługach webowych, a nie o ciągłym tłoczeniu strumieni o przepustowości rzędu wielu gigabitów. Optymalizacja wymaga zwiększenia rozmiarów kolejek wejściowych i wyjściowych oraz dostrojenia mechanizmu autotuningu TCP:

# Zwiększenie maksymalnego rozmiaru buforów TCP (odczyt/zapis)
sysctl -w net.core.rmem_max=16777216
sysctl -w net.core.wmem_max=16777216
sysctl -w net.ipv4.tcp_rmem="4096 87380 16777216"
sysctl -w net.ipv4.tcp_wmem="4096 65536 16777216"

# Wydłużenie kolejki oczekujących połączeń oraz pakietów sieciowych
sysctl -w net.core.somaxconn=4096
sysctl -w net.core.netdev_max_backlog=10000

# Włączenie ochrony przed wyczerpaniem portów efemerycznych przy setkach równoległych połączeń
sysctl -w net.ipv4.tcp_tw_reuse=1

Bez tych modyfikacji wątki klienta Java zaczynają okresowo blokować się na operacjach wywołania write() na poziomie deskryptorów gniazd, co wywołuje zjawisko backpressure propagujące się wstecz aż do źródeł danych w klastrze Apache Kafka. Stabilny potok wymaga, aby wąskie gardło znajdowało się wyłącznie w przepustowości kontrolerów NVMe po stronie ClickHouse, a nie w narzucie jądra systemu operacyjnego na obsługę pakietów TCP.

Optymalizacja silnika MergeTree pod kątem ciągłego, zmasowanego zapisu

Aplikacja ingestująca może działać bezbłędnie, lecz jeśli konfiguracja tabeli docelowej nie uwzględnia specyfiki zmasowanego zapisu, silnik bazodanowy prędzej czy później odmówi przyjmowania kolejnych paczek. ClickHouse zapisuje każdy zrzucony blok jako niezależny fragment na dysku (part), który następnie w tle łączony jest z innymi fragmentami w ramach procesu background merge. Zbyt duża liczba małych partów generuje błąd Too many parts in all data parts in table i wstrzymuje operacje INSERT.

Aby zapobiec temu zjawisku, struktura docelowa musi unikać nadmiernej fragmentacji klucza partycjonowania. Klasycznym błędem jest partycjonowanie danych po zbyt precyzyjnym wymiarze, na przykład godzinowo: PARTITION BY toYYYYMMDDhh(timestamp). W scenariuszu wielomilionowego strumienia prowadzi to do sytuacji, w której pojedynczy batch z klienta Java zawiera rekordy rozproszone po wielu godzinach (np. z powodu opóźnionych zdarzeń), zmuszając ClickHouse do tworzenia dziesiątek partów w ramach jednego zapytania. Bezpiecznym standardem dla danych strumieniowych o wysokim natężeniu jest partycjonowanie miesięczne lub co najwyżej dzienne:

CREATE TABLE telemetry.events_local ON CLUSTER production
(
    id UInt64,
    timestamp DateTime64(3, 'UTC') CODEC(DoubleDelta, LZ4),
    tenant_id LowCardinality(String),
    value Float64 CODEC(Gorilla, LZ4),
    is_valid UInt8 CODEC(T64, ZSTD(1))
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
PRIMARY KEY (tenant_id, timestamp)
ORDER BY (tenant_id, timestamp, id)
SETTINGS 
    index_granularity = 8192,
    parts_to_delay_insert = 150,
    parts_to_throw_insert = 300,
    max_parts_in_total = 10000;

Warto zwrócić szczególną uwagę na zdefiniowane kodeki kompresji dla poszczególnych kolumn. Domyślny algorytm LZ4 jest wysoce wydajny, ale zastosowanie specjalizowanych kodeków pozwala ClickHouse zapisać dane w jeszcze bardziej zwartej formie niemal bez narzutu na CPU. Kodek DoubleDelta dla rosnących monotonicznie znaczników czasu redukuje niemal wszystkie wartości do różnic rzędu kilkunastu bitów, natomiast algorytm Gorilla idealnie kompresuje zmiennoprzecinkowe szeregi czasowe.

Zwiększenie limitów parts_to_delay_insert i parts_to_throw_insert w parametrach tabeli daje silnikowi bufor bezpieczeństwa w momentach nagłych skoków natężenia ruchu. Daje to procesom kompresującym i łączącym w tle dodatkowy czas na zredukowanie liczby partów bez natychmiastowego blokowania wątków ingestujących aplikacji klienckiej.

Odporność na awarie i semantyka Exactly-Once bez transakcji dwufazowych

Brak klasycznych transakcji ACID w bazach analitycznych wymusza zaprojektowanie odporności na błędy w samej aplikacji. Standardowe mechanizmy ponawiania żądań sieciowych w przypadku zerwania połączenia TCP mogą doprowadzić do powielenia milionów zdarzeń, jeśli serwer ClickHouse zdążył zapisać blok na dysku przed wystąpieniem błędu sieciowego po stronie klienta.

ClickHouse rozwiązuje ten problem natywnie poprzez wbudowany mechanizm deduplikacji bloków. Silnik bazy tworzy skrót kryptograficzny (hash) zawartości całego zapisanego bloku i rejestruje go w pamięci lub w Apache ZooKeeper/Keeperze na określony czas (kontrolowany parametrem replicated_deduplication_window). Jeżeli aplikacja Java napotka błąd I/O podczas transmisji batcha, może bezpiecznie ponowić dokładnie tę samą paczkę danych. Serwer rozpozna identyczny skrót zawartości i zignoruje nadmiarowy zapis, zwracając status sukcesu bez duplikowania wierszy na dysku.

Aby ten mechanizm działał w sposób w pełni deterministyczny przy strumieniowaniu z partycjonowanych systemów takich jak Apache Kafka, aplikacja kliencka musi zagwarantować, że ponawiana paczka danych zawiera dokładnie te same rekordy i w tej samej kolejności. Wymaga to powiązania zatwierdzania offsetów consumera z faktycznym potwierdzeniem zapisu bloku przez sterownik ClickHouse:

// Pseudokod deterministycznego zatwierdzania stanu bufora
public void processPartitionStream(KafkaConsumer<byte[], byte[]> consumer) {
    while (running) {
        ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(100));
        if (records.isEmpty()) continue;

        DirectMemoryBuffer buffer = serializeToNativeMemory(records);
        
        boolean inserted = false;
        int attempts = 0;
        
        while (!inserted && attempts < MAX_RETRIES) {
            try {
                // Przesyłamy ten sam, niezmieniony bufor z danymi
                client.loadData(insertQuery, buffer.asInputStream(), ClickHouseFormat.RowBinary).get();
                inserted = true;
            } catch (Exception e) {
                attempts++;
                logger.warn("Błąd wysyłki bloku, ponawianie próby: {}", attempts, e);
                backoff(attempts);
            }
        }

        if (inserted) {
            // Dopiero po uzyskaniu pewności zapisu commitujemy offsety do brokera
            consumer.commitSync();
            buffer.reset();
        } else {
            handlePoisonPill(buffer);
        }
    }
}

Taka architektura eliminuje potrzebę stosowania kosztownych rozproszonych koordynatorów transakcji, przenosząc ciężar atomowości na poziom pojedynczego bloku w bazie danych. Dzięki temu całe środowisko zachowuje maksymalną wydajność I/O, a ewentualne awarie sieciowe czy restarty pojedynczych instancji klastra ClickHouse nie prowadzą do degradacji spójności analitycznego zbioru danych.