Ich habe Kafka Stream-Anwendung mit 1.0.0 Kafka Stream API. Ich habe Einzelmakler 0.10.2.0 Kafka und einzelnes Thema mit einer einzelnen Partition. Alle konfigurierbaren Parameter sind identisch mit Ausnahme von Producer request.timeout.ms. Ich konfigurierte Producer request.timeout.ms mit 5 Minuten, um das Problem Kafka Streams program is throwing exceptions when producing zu beheben.Wie man die Leistung verbessert, um von kafka zu lesen und an kafka mit kafka weiterzuleiten Stream-Anwendung
In meiner Stream-Anwendung lese ich die Ereignisse von Kafka, verarbeite sie und leite sie zu einem anderen Thema des gleichen Kafkas weiter.
Nach der Berechnung der Statistiken habe ich beobachtet, dass die Verarbeitung dauert 5% der Zeit und verbleibenden 95% der Zeit dauert zum Lesen & Schreiben.
Obwohl ich Dutzende von Millionen von Ereignissen in Kafka, einige Male Kafka Umfrage gibt einzelne Stelle von Datensätzen zurück und manchmal Kafka Umfrage gibt tausend Datensätze.
Manchmal dauert Context Forward mehr Zeit, um weniger Datensätze an Kafka zu senden, und manchmal dauert Context Forward weniger Zeit, um mehr Datensätze an Kafka zu senden.
Ich versuchte, die Leseleistung zu erhöhen, indem ich max.poll.records, poll.ms Werte erhöhte. Aber kein Glück.
Wie kann ich die Leistung beim Lesen und Weiterleiten verbessern? Wie würde Kafka Umfrage und Forward funktionieren? Welche Parameter haben zur Verbesserung der Leistung beigetragen?
Im Folgenden sind einige wichtige Hersteller Config-Parameter in meiner Anwendung.
acks = 1
batch.size = 16384
buffer.memory = 33554432
compression.type = none
connections.max.idle.ms = 540000
enable.idempotence = false
linger.ms = 100
max.block.ms = 60000
max.in.flight.requests.per.connection = 5
max.request.size = 1048576
metadata.max.age.ms = 300000
receive.buffer.bytes = 32768
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 240000
retries = 10
retry.backoff.ms = 100
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
transaction.timeout.ms = 60000
transactional.id = null
Im Folgenden sind einige wichtige Verbraucher-Konfigurationsparameter in meiner Anwendung:
auto.commit.interval.ms = 5000
auto.offset.reset = earliest
check.crcs = true
connections.max.idle.ms = 540000
enable.auto.commit = false
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
heartbeat.interval.ms = 3000
internal.leave.group.on.close = false
isolation.level = read_uncommitted
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 2147483647
max.poll.records = 10000
metadata.max.age.ms = 300000
receive.buffer.bytes = 65536
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 305000
retry.backoff.ms = 100
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
session.timeout.ms = 10000
einige wichtige Stream Konfigurationsparameter in meiner Anwendung Im Folgenden sind:
application.server =
buffered.records.per.partition = 1000
cache.max.bytes.buffering = 10485760
commit.interval.ms = 30000
connections.max.idle.ms = 540000
key.serde = null
metadata.max.age.ms = 300000
num.standby.replicas = 0
num.stream.threads = 1
poll.ms = 1000
processing.guarantee = at_least_once
receive.buffer.bytes = 32768
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
replication.factor = 1
request.timeout.ms = 40000
retry.backoff.ms = 100
rocksdb.config.setter = null
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
state.cleanup.delay.ms = 600000
timestamp.extractor = null
value.serde = null
windowstore.changelog.additional.retention.ms = 86400000
zookeeper.connect =