Błędy producenta Kafka w Node.js: rozwiązania i optymalizacja

Problemy i rozwiązania przy pracy z producentem Kafka w Node.js

Wybór partycji dla wiadomości to krytyczny etap wpływający na rozkład obciążenia między konsumentami. Nieprawidłowa logika partycjonowania może prowadzić do dysbalansu i utraty skalowalności.

Częstą błędem jest użycie funkcji partycjonowania opartej na dzieleniu całkowitym wartości klucza wiadomości (key). Jeśli klucz nie jest liczbą lub nie może zostać poprawnie przekonwertowany na integer (np. MongoDB ObjectID), wszystkie wiadomości będą kierowane do jednej partycji.

const producer = await kafka.producer(topic, {
  createPartitioner: function () {
    return ({ partitionMetadata, message }: PartitionerArgs) => {
      return partitionMetadata.length
        ? partitionMetadata[Number(message.key) % partitionMetadata.length]?.partitionId || 0
        : 0;
    };
  },
});

W tym przypadku konsumenty w grupie nie będą mogły skutecznie rozdzielić obciążenia, ponieważ wszystkie dane skupią się w jednej partycji.

Rozwiązanie:

• Weryfikuj typ i format klucza partycji na poziomie producenta.
• Skonfiguruj metryki do śledzenia dysbalansu rozmiarów partycji.
• Używaj bardziej niezawodnych algorytmów partycjonowania, np. haszowanie ciągu klucza.

Publikacja w nieistniejącej partycji

Numeracja partycji w temacie zaczyna się od 0. Jeśli producent próbuje opublikować wiadomość w partycji o indeksie przekraczającym dostępne (np. partycja 1 dla tematu z jedną partycją), KafkaJS nie rzuci błędu. Metoda send() zwróci pusty tablicę, a wiadomość faktycznie nie zostanie zapisana.

const result = await producer.send({
  topic: 'user_events',
  messages: [
    { key: '2', value: 'user_created', partition: 1 }
  ]
});
console.log(result); // []

Rozwiązanie:

• Twórz tematy i partycje za pomocą plików konfiguracyjnych podczas wdrożenia, unikając błędów ręcznych.
• Weryfikuj odpowiedź metody send() na poziomie aplikacji. Jeśli wynik jest pusty lub zawiera błąd (np. errorCode różny od 0), należy podjąć działania: rzucić wyjątek, zapisać do logu, opublikować w zapasowym temacie lub wysłać metrykę.
• Regularnie sprawdzaj metadane tematów (listę dostępnych partycji) przed wysyłką.

Nieefektywne wykorzystanie producentów

Tworzenie nowego egzemplarza producenta (kafka.producer()) dla każdego żądania lub operacji to częsty błąd prowadzący do znacznych kosztów. Każdy nowy producent nawiązuje ciężkie połączenia sieciowe z brokerami Kafka.

app.post('/send-message', async (request) => {
  const { topic, message } = request.body;
  const producer = kafka.producer(); // Nowy egzemplarz na każde żądanie
  // ...
});

Rozwiązanie:

• Używaj wzorca Singleton lub puli producentów. Stwórz jeden egzemplarz producenta i używaj go do wszystkich operacji publikacji (producer.send()). Zwiększy to przepustowość i zmniejszy obciążenie brokerów.
• Dla architektur wielowątkowych lub mikroserwisowych rozważ użycie wspólnego klienta Kafka z udostępnionym producentem.

Problemy sieciowe i strategia ponownych prób

Sieć między producentem a klastrem Kafka może być niestabilna. Aby zwiększyć niezawodność dostarczania wiadomości, należy:

• Skonfigurować obsługę zdarzenia request_timeout do śledzenia problemów sieciowych.

producer.on('producer.network.request_timeout', (error) => {
  console.error({
    code: 'producer_network_timeout',
    message: error.message,
    data: { stack: error.stack },
  });
})

• Skonfigurować mechanizm ponownych prób (retry) podczas tworzenia producenta. Domyślnie liczba prób wynosi 5.

kafka.producer({
  retry: { retries: 10 }
});

Uwaga: zwiększenie liczby prób bez odpowiedniej strategii (np. bez zwiększania odstępów między próbami) może prowadzić do obciążenia typu DDoS dla brokerów. Używaj konfiguracji z rosnącymi odstępami (exponential backoff).

Obsługa dużych wiadomości

Kafka nie została zaprojektowana do przesyłania wiadomości większych niż 1 MB. Publikacja dużych danych (5–10 MB) prowadzi do poważnych problemów:

• Częsta rotacja segmentów: przy małym maksymalnym rozmiarze segmentu (np. 512 MB) duże wiadomości powodują szybką rotację, co zwiększa obciążenie systemu.
• Wysoki ruch sieciowy: replikacja dużych wiadomości do wszystkich In-Sync Replicas (ISR) zwiększa obciążenie sieci.
• Obniżona przepustowość: bufor wiadomości producenta rośnie, zwiększają się ryzyka utraty danych oraz czas oczekiwania na potwierdzenie przy acks=-1.
• Zużycie zasobów: wymagane jest więcej RAM dla os page cache na brokerach i dla JVM Kafka, wzrasta obciążenie CPU z powodu działania zbieracza śmieci (GC).

Zalecenia dotyczące rozmiaru wiadomości:

• Użyj wzoru: message.max.bytes = średnia wielkość wiadomości * 2, ale mniejsza niż 1 MB.
• Do przesyłania dużych plików (wideo, audio, dokumenty) używaj zewnętrznych magazynów (np. S3). W wiadomości Kafka przesyłaj tylko odnośnik do obiektu.
• Jeśli konieczne jest przesłanie dużych JSON/XML bez zewnętrznego magazynu, prawdopodobnie wiadomość zawiera nadmiarowe dane dla różnych konsumentów. Rozważ podział tematu na kilka specjalistycznych, każdy z minimalnym niezbędnym zestawem pól.
• Alternatywna, choć bardziej skomplikowana metoda — podział dużej wiadomości na bloki i scalenie po stronie konsumenta.

Co jest ważne

• Partycjonowanie: niepoprawne obliczanie partycji prowadzi do dysbalansu obciążenia i utraty skalowalności. Weryfikacja klucza i monitorowanie metryk są niezbędne.
• Istnienie partycji: publikacja w nieistniejącej partycji nie powoduje błędu w KafkaJS, ale wiadomość nie zostaje zapisana. Weryfikacja odpowiedzi send() jest krytyczna.
• Efektywność producenta: tworzenie nowego producenta na każde żądanie to kosztowna operacja. Używaj wzorca Singleton lub puli.
• Sieć i ponowne próby: konfiguracja obsługi timeoutów sieciowych i strategii ponownych prób zwiększa niezawodność dostarczania.
• Rozmiar wiadomości: przesyłanie danych większych niż 1 MB negatywnie wpływa na wydajność klastra. Używaj zewnętrznych magazynów dla dużych plików i optymalizuj strukturę wiadomości.

— Editorial Team