Wie kann ich mit Kafka (über 15 MB) große Nachrichten senden?

118

Ich sende String-Nachrichten mit der Java Producer API an Kafka V. 0.8. Wenn die Nachrichtengröße ca. 15 MB beträgt, erhalte ich eine MessageSizeTooLargeException. Ich habe versucht, message.max.bytesauf 40 MB einzustellen , aber ich bekomme immer noch die Ausnahme. Kleine Nachrichten funktionierten ohne Probleme.

(Die Ausnahme erscheint im Hersteller, ich habe keinen Verbraucher in dieser Anwendung.)

Was kann ich tun, um diese Ausnahme zu beseitigen?

Mein Beispiel Produzent Konfiguration

private ProducerConfig kafkaConfig() {
    Properties props = new Properties();
    props.put("metadata.broker.list", BROKERS);
    props.put("serializer.class", "kafka.serializer.StringEncoder");
    props.put("request.required.acks", "1");
    props.put("message.max.bytes", "" + 1024 * 1024 * 40);
    return new ProducerConfig(props);
}

Fehlerprotokoll:

4709 [main] WARN  kafka.producer.async.DefaultEventHandler  - Produce request with correlation id 214 failed due to [datasift,0]: kafka.common.MessageSizeTooLargeException
4869 [main] WARN  kafka.producer.async.DefaultEventHandler  - Produce request with    correlation id 217 failed due to [datasift,0]: kafka.common.MessageSizeTooLargeException
5035 [main] WARN  kafka.producer.async.DefaultEventHandler  - Produce request with   correlation id 220 failed due to [datasift,0]: kafka.common.MessageSizeTooLargeException
5198 [main] WARN  kafka.producer.async.DefaultEventHandler  - Produce request with correlation id 223 failed due to [datasift,0]: kafka.common.MessageSizeTooLargeException
5305 [main] ERROR kafka.producer.async.DefaultEventHandler  - Failed to send requests for topics datasift with correlation ids in [213,224]

kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
at kafka.producer.async.DefaultEventHandler.handle(Unknown Source)
at kafka.producer.Producer.send(Unknown Source)
at kafka.javaapi.producer.Producer.send(Unknown Source)
Sonson123
quelle
5
Mein erster Instinkt wäre, Sie zu bitten, diese große Nachricht in mehrere kleinere aufzuteilen: - / Ich vermute, dass dies aus irgendeinem Grund nicht möglich ist, aber Sie möchten es vielleicht trotzdem überdenken: Riesige Nachrichten bedeuten normalerweise, dass ein Designfehler vorliegt irgendwo sollte das wirklich behoben werden.
Aaron Digulla
1
Danke, aber es würde meine Logik viel komplexer machen. Warum ist es eine schlechte Idee, Kafka für Nachrichten mit etwa 15 MB zu verwenden? Ist 1 MB die maximale Beschränkung der Nachrichtengröße, die verwendet werden kann? Ich habe in der Kafka-Dokumentation nicht viel über die Beschränkung der Nachrichtengröße herausgefunden.
Sonson123
2
Dies hat nichts mit Kafka oder einem anderen Nachrichtenverarbeitungssystem zu tun. Meine Argumentation: Wenn mit Ihrer 15-MB-Datei etwas schief geht, ist das anschließende Aufräumen des Chaos sehr teuer. Deshalb teile ich normalerweise große Dateien in viele kleinere Jobs auf (die dann normalerweise auch parallel ausgeführt werden können).
Aaron Digulla
Haben Sie eine Komprimierung verwendet?
Könnten

Antworten:

181

Sie müssen drei (oder vier) Eigenschaften anpassen:

  • Verbraucherseite: fetch.message.max.bytes- Hiermit wird die größte Größe einer Nachricht bestimmt, die vom Verbraucher abgerufen werden kann.
  • Brokerseite: replica.fetch.max.bytes- Dadurch können die Replikate in den Brokern Nachrichten innerhalb des Clusters senden und sicherstellen, dass die Nachrichten korrekt repliziert werden. Wenn dies zu klein ist, wird die Nachricht niemals repliziert, und daher wird der Verbraucher die Nachricht niemals sehen, da die Nachricht niemals festgeschrieben (vollständig repliziert) wird.
  • Brokerseite: message.max.bytes- Dies ist die größte Größe der Nachricht, die der Broker von einem Produzenten empfangen kann.
  • Brokerseite (pro Thema): max.message.bytes- Dies ist die größte Größe der Nachricht, die der Broker an das Thema anhängen darf. Diese Größe wird vor der Komprimierung validiert. (Standardmäßig Broker message.max.bytes.)

Ich habe den schwierigen Weg zu Nummer 2 herausgefunden - Sie erhalten KEINE Ausnahmen, Nachrichten oder Warnungen von Kafka. Denken Sie also daran, wenn Sie große Nachrichten senden.

lachender Mann
quelle
3
Ok, Sie und user2720864 waren korrekt. Ich hatte das nur message.max.bytesim Quellcode eingestellt. Aber ich muss diese Werte in der Konfiguration des Kafka-Servers einstellen config/server.properties. Jetzt funktionieren auch größere Nachrichten :).
Sonson123
3
Gibt es bekannte Nachteile, wenn diese Werte zu hoch eingestellt werden?
Ivan Balashov
7
Ja. Auf der Verbraucherseite weisen Sie fetch.message.max.bytesJEDER Partition Speicher zu. Dies bedeutet, dass bei Verwendung einer großen Anzahl für die fetch.message.max.bytesKombination mit einer großen Anzahl von Partitionen viel Speicherplatz verbraucht wird. Da der Replikationsprozess zwischen den Brokern auch ein spezialisierter Verbraucher ist, wird dadurch auch Speicher auf den Brokern verbraucht.
Laughing_man
3
Beachten Sie, dass es auch eine max.message.bytesKonfiguration pro Thema gibt, die niedriger sein kann als die des Brokers message.max.bytes.
Peter Davis
1
Laut offiziellem Dokument /.*fetch.*bytes/scheinen die Parameter auf der Verbraucherseite und diejenigen, die die Replikation zwischen Brokern betreffen , keine harten Grenzen zu sein: "Dies ist kein absolutes Maximum, wenn [...] dieser Wert größer ist als der Rekordstapel noch zurückgegeben werden, um sicherzustellen, dass Fortschritte erzielt werden können. "
Bluu
56

Für Kafka 0.10 und den neuen Verbraucher sind geringfügige Änderungen erforderlich, verglichen mit der Antwort von smile_man :

  • Broker: Keine Änderungen, Sie müssen noch Eigenschaften message.max.bytesund erhöhen replica.fetch.max.bytes. message.max.bytesmuss gleich oder kleiner (*) sein als replica.fetch.max.bytes.
  • Produzent: Erhöhen Sie max.request.size, um die größere Nachricht zu senden.
  • Verbraucher: Erhöhen Sie sich max.partition.fetch.bytes, um größere Nachrichten zu erhalten.

(*) Lesen Sie die Kommentare, um mehr über message.max.bytes<= zu erfahrenreplica.fetch.max.bytes

Sascha Vetter
quelle
2
Wissen Sie, warum message.max.byteskleiner sein muss als replica.fetch.max.bytes?
Kostas
2
" replica.fetch.max.bytes (Standard: 1 MB) - Maximale Datengröße, die ein Broker replizieren kann. Diese muss größer sein als message.max.bytes. Andernfalls akzeptiert ein Broker Nachrichten und kann sie nicht replizieren potenzieller Datenverlust. " Quelle: Umgang mit großen Nachrichten-Kafka
Sascha Vetter
2
Vielen Dank, dass Sie sich mit einem Link bei mir gemeldet haben. Dies scheint dem zu entsprechen, was der Cloudera-Leitfaden ebenfalls vorschlägt. Beide sind jedoch falsch - beachten Sie, dass sie keinen technischen Grund dafür bieten, warumreplica.fetch.max.bytes sie unbedingt größer sein sollten message.max.bytes. Ein Confluent-Mitarbeiter hat heute früher bestätigt, was ich vermutet habe: dass die beiden Mengen tatsächlich gleich sein können.
Kostas
2
Gibt es Updates zu message.max.bytes<replica.fetch.max.bytesoder message.max.bytes=replica.fetch.max.bytes@Kostas?
Sascha Vetter
2
Ja, sie können gleich sein: mail-archive.com/[email protected]/msg25494.html (Ismael arbeitet für Confluent)
Kostas
13

Sie müssen die folgenden Eigenschaften überschreiben:

Broker-Konfigurationen ($ KAFKA_HOME / config / server.properties)

  • replica.fetch.max.bytes
  • message.max.bytes

Consumer-Konfigurationen ($ KAFKA_HOME / config / consumer.properties)
Dieser Schritt hat bei mir nicht funktioniert. Ich füge es der Consumer-App hinzu und es hat gut funktioniert

  • fetch.message.max.bytes

Starten Sie den Server neu.

Weitere Informationen finden Sie in dieser Dokumentation: http://kafka.apache.org/08/configuration.html

user2550587
quelle
1
Für den Befehlszeilen-Consumer muss das Flag --fetch-size = <bytes> verwendet werden. Die Datei consumer.properties (kafka 0.8.1) scheint nicht gelesen zu werden. Ich würde auch empfehlen, die Komprimierung von der Herstellerseite mit der Option compress.codec zu aktivieren.
Ziggy Eunicien
Ziggys Kommentar hat bei mir funktioniert kafka 0.8.1.1. Danke dir!
James
Könnte es sein, dass fetch.message.max.bytes in ConsumerConfig durch max.partition.fetch.bytes ersetzt wird?
s_bei
12

Die Idee ist, dass die gleiche Größe der Nachricht vom Kafka-Produzenten an den Kafka-Broker gesendet und dann von Kafka Consumer empfangen wird, d. H.

Kafka Produzent -> Kafka Broker -> Kafka Consumer

Angenommen, wenn 15 MB Nachricht gesendet werden sollen, müssen der Produzent , der Broker und der Verbraucher , alle drei, synchron sein.

Kafka Producer sendet 15 MB -> Kafka Broker erlaubt / speichert 15 MB -> Kafka Consumer erhält 15 MB

Die Einstellung sollte daher sein:

a) auf Broker:

message.max.bytes=15728640 
replica.fetch.max.bytes=15728640

b) zum Verbraucher:

fetch.message.max.bytes=15728640
Ravi
quelle
2
Könnte es sein, dass fetch.message.max.bytes in ConsumerConfig durch max.partition.fetch.bytes ersetzt wird?
s_bei
7

Eine wichtige Sache, an die Sie sich erinnern sollten, ist, dass dieses message.max.bytesAttribut mit dem Eigentum des Verbrauchers synchron sein muss fetch.message.max.bytes. Die Abrufgröße muss mindestens so groß sein wie die maximale Nachrichtengröße. Andernfalls kann es vorkommen, dass Produzenten Nachrichten senden können, die größer sind, als der Verbraucher verbrauchen / abrufen kann. Es könnte sich lohnen, einen Blick darauf zu werfen.
Welche Version von Kafka verwenden Sie? Geben Sie auch einige weitere Details an, die Sie erhalten. Gibt es etwas wie ... payload size of xxxx larger than 1000000im Protokoll auftauchen?

user2720864
quelle
1
Ich habe meine Frage mit weiteren Informationen aktualisiert: Kafka Version 2.8.0-0.8.0; Jetzt brauche ich nur noch den Produzenten.
Sonson123
6

Die Antwort von @laughing_man ist ziemlich genau. Trotzdem wollte ich eine Empfehlung geben, die ich von Kafka-Experte Stephane Maarek aus Quora gelernt habe .

Kafka ist nicht für große Nachrichten gedacht.

Ihre API sollte Cloud-Speicher (Ex AWS S3) verwenden und einfach eine Referenz von S3 an Kafka oder einen Nachrichtenbroker senden. Sie müssen einen Ort finden, an dem Ihre Daten gespeichert werden können. Vielleicht handelt es sich um ein Netzwerklaufwerk, vielleicht um was auch immer, aber es sollte kein Nachrichtenbroker sein.

Nun, wenn Sie nicht mit der oben genannten Lösung gehen möchten

Die Meldung max Größe ist 1 MB (die Einstellung in Ihrem Broker genannt wird message.max.bytes) Apache Kafka . Wenn Sie es wirklich dringend benötigen, können Sie diese Größe erhöhen und sicherstellen, dass die Netzwerkpuffer für Ihre Produzenten und Verbraucher erhöht werden.

Wenn Sie sich wirklich für die Aufteilung Ihrer Nachricht interessieren, stellen Sie sicher, dass jede Aufteilung der Nachricht genau denselben Schlüssel hat, damit sie auf dieselbe Partition übertragen wird, und Ihr Nachrichteninhalt sollte eine „Teil-ID“ melden, damit Ihr Verbraucher die Nachricht vollständig rekonstruieren kann .

Sie können die Komprimierung auch untersuchen, wenn Ihre Nachricht textbasiert ist (gzip, snappy, lz4-Komprimierung), wodurch die Datengröße möglicherweise verringert wird, jedoch nicht auf magische Weise.

Auch hier müssen Sie ein externes System verwenden, um diese Daten zu speichern und einfach einen externen Verweis auf Kafka zu senden. Das ist eine sehr verbreitete Architektur, mit der Sie sich einverstanden erklären sollten.

Denken Sie daran, dass Kafka nur dann am besten funktioniert, wenn die Nachrichten eine große Menge, aber keine große Größe haben.

Quelle: https://www.quora.com/How-do-I-send-Large-messages-80-MB-in-Kafka

Bhanu Hoysala
quelle
4
Vielleicht möchten Sie beachten, dass "Ihre" Empfehlung eine fast wörtliche Kopie der Quora-Empfehlung von Stéphane Maarek unter quora.com/How-do-I-send-Large-messages-80-MB-in-Kafka
Mike ist
Kafka arbeitet mit großen Nachrichten, absolut kein Problem. Die Intro-Seite auf der Kafka-Homepage verweist sogar auf ein Speichersystem.
calloc_org
3

Für Benutzer von landoop kafka: Sie können die Konfigurationswerte in den Umgebungsvariablen wie folgt übergeben:

docker run -d --rm -p 2181:2181 -p 3030:3030 -p 8081-8083:8081-8083  -p 9581-9585:9581-9585 -p 9092:9092
 -e KAFKA_TOPIC_MAX_MESSAGE_BYTES=15728640 -e KAFKA_REPLICA_FETCH_MAX_BYTES=15728640  landoop/fast-data-dev:latest `

Und wenn Sie rdkafka verwenden, übergeben Sie die message.max.bytes in der Produzenten-Konfiguration wie folgt:

  const producer = new Kafka.Producer({
        'metadata.broker.list': 'localhost:9092',
        'message.max.bytes': '15728640',
        'dr_cb': true
    });

Ebenso für den Verbraucher,

  const kafkaConf = {
   "group.id": "librd-test",
   "fetch.message.max.bytes":"15728640",
   ... .. }                                                                                                                                                                                                                                                      
Informant
quelle