Kafka configuration tuning
Kafka configuration tuning Red Hat Streams for Apache Kafka 2.9 Use Kafka configuration properties to optimize the streaming of data
num.partitions:1 num.partitions=1 broker.id = 1 log.dirs = /var/lib/kafka zookeeper.connect = zoo1.my-domain.com:2181,zoo2.my-domain.com:2181,zoo3.my-domain.com:2181 listeners = internal-1://:9092 authorizer.class.name = kafka.security.auth.SimpleAclAuthorizer ssl.truststore.location = /path/to/truststore.jks ssl.truststore.password = 123456 ssl.client.auth = required num.partitions=1 default.replication.factor=3 offsets.topic.replication.factor=3 transaction.state.log.replication.factor=3 transaction.state.log.min.isr=2 log.retention.hours=168 log.segment.bytes=1073741824 log.retention.check.interval.ms=300000 num.network.threads=3 num.io.threads=8 num.recovery.threads.per.data.dir=1 socket.send.buffer.bytes=102400 socket.receive.buffer.bytes=102400 socket.request.max.bytes=104857600 group.initial.rebalance.delay.ms=0 zookeeper.connection.timeout.ms=6000 num.partitions=1 auto.create.topics.enable=false default.replication.factor=3 min.insync.replicas=2 replica.fetch.max.bytes=1048576 auto.create.topics.enable=false delete.topic.enable=true transaction.state.log.replication.factor=3 transaction.state.log.min.isr=2 offsets.topic.num.partitions=50 offsets.topic.replication.factor=3 num.network.threads=3 1 queued.max.requests=500 2 num.io.threads=8 3 num.recovery.threads.per.data.dir=4 4 replica.socket.receive.buffer.bytes=65536 socket.request.max.bytes=104857600 socket.send.buffer.bytes=1048576 socket.receive.buffer.bytes=1048576 log.segment.bytes=1073741824 log.roll.ms=604800000 log.cleanup.policy=delete log.cleaner.enable=true log.retention.ms=1680000 log.retention.bytes=1073741824 log.segment.delete.delay.ms=60000 log.retention.check.interval.ms=300000 log.cleaner.delete.retention.ms=86400000 log.cleaner.backoff.ms=15000 log.cleaner.dedupe.buffer.size=134217728 log.cleaner.io.buffer.load.factor=0.9 log.cleaner.threads=8 log.cleaner.io.max.bytes.per.second=1.7976931348623157E308 log.flush.scheduler.interval.ms=2000 log.flush.interval.ms=50000 log.flush.interval.messages=100000 replica.lag.time.max.ms=30000 # auto.leader.rebalance.enable=true leader.imbalance.check.interval.seconds=300 leader.imbalance.per.broker.percentage=10 # unclean.leader.election.enable=false group.initial.rebalance.delay.ms=3000 bootstrap.servers=localhost:9092 1 key.deserializer=org.apache.kafka.common.serialization.StringDeserializer 2 value.deserializer=org.apache.kafka.common.serialization.StringDeserializer 3 client.id=my-client 4 group.id=my-group-id 5 group.id=my-group-id 1 fetch.max.wait.ms=500 1 fetch.min.bytes=16384 2 NUMBER-OF-BROKERS * fetch.max.bytes and NUMBER-OF-PARTITIONS * max.partition.fetch.bytes fetch.max.bytes=52428800 1 max.partition.fetch.bytes=1048576 2 enable.auto.commit=false 1 enable.auto.commit=false isolation.level=read_committed 1 max.poll.records=100 1 session.timeout.ms=30000 2 heartbeat.interval.ms=5000 3 auto.offset.reset=earliest 1 group.instance.id=<unique_id> max.poll.interval.ms=300000 max.poll.records=500 session.timeout.ms=30000 heartbeat.interval.ms=5000 partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor bootstrap.servers=localhost:9092 1 key.serializer=org.apache.kafka.common.serialization.StringSerializer 2 value.serializer=org.apache.kafka.common.serialization.StringSerializer 3 client.id=my-client 4 compression.type=gzip 5 acks=all 1 delivery.timeout.ms=120000 2 enable.idempotence=true 1 max.in.flight.requests.per.connection=5 2 acks=all 3 retries=2147483647 4 enable.idempotence=false 1 max.in.flight.requests.per.connection=1 2 retries=2147483647 enable.idempotence=true max.in.flight.requests.per.connection=5 acks=all retries=2147483647 transactional.id= UNIQUE-ID 1 transaction.timeout.ms=900000 2 linger.ms=100 1 batch.size=16384 2 buffer.memory=33554432 3 partitioner.class=my-custom-partitioner 1 apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect-cluster annotations: strimzi.io/use-connector-resources: "true" spec: replicas: 3 config: offset.flush.timeout.ms: 10000 # resources: requests: cpu: "1" memory: 2Gi limits: cpu: "2" memory: 2Gi # apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: name: my-source-connector labels: strimzi.io/cluster: my-connect-cluster spec: class: org.apache.kafka.connect.file.FileStreamSourceConnector tasksMax: 2 config: producer.override.batch.size: 327680 producer.override.linger.ms: 100 # apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: name: my-sink-connector labels: strimzi.io/cluster: my-connect-cluster spec: class: org.apache.kafka.connect.file.FileStreamSinkConnector tasksMax: 2 config: consumer.fetch.max.bytes: 52428800 consumer.max.partition.fetch.bytes: 1048576 consumer.max.poll.records: 500 # curl -X POST http://my-connect-cluster-connect-api:8083/connectors -H 'Content-Type: application/json' -d '{ "name": "my-source-connector", "config": { "connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector", "file": "/opt/kafka/LICENSE", "topic":"my-topic", "tasksMax": "4", "type": "source" "producer.override.batch.size": 327680 "producer.override.linger.ms": 100 } }' apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaMirrorMaker2 metadata: name: my-mirror-maker2 spec: version: 3.9.0 replicas: 1 connectCluster: "my-cluster-target" clusters: - alias: "my-cluster-source" bootstrapServers: my-cluster-source-kafka-bootstrap:9092 - alias: "my-cluster-target" config: offset.flush.timeout.ms: 10000 bootstrapServers: my-cluster-target-kafka-bootstrap:9092 mirrors: - sourceCluster: "my-cluster-source" targetCluster: "my-cluster-target" sourceConnector: tasksMax: 2 config: producer.override.batch.size: 327680 producer.override.linger.ms: 100 consumer.fetch.max.bytes: 52428800 consumer.max.partition.fetch.bytes: 1048576 consumer.max.poll.records: 500 # resources: requests: cpu: "1" memory: Gi limits: cpu: "2" memory: 4Gi message.max.bytes: 10000000 replica.fetch.max.bytes: 10485760 batch.size: 327680 max.request.size: 10000000 fetch.max.bytes: 10000000 max.partition.fetch.bytes: 10485760 apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaBridge metadata: name: my-bridge spec: # producer: config: batch.size: 327680 max.request.size: 10000000 consumer: config: fetch.max.bytes: 10000000 max.partition.fetch.bytes: 10485760 # apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: name: my-source-connector labels: strimzi.io/cluster: my-connect-cluster spec: # config: producer.override.batch.size: 327680 producer.override.max.request.size: 10000000 # apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: name: my-sink-connector labels: strimzi.io/cluster: my-connect-cluster spec: # config: consumer.fetch.max.bytes: 10000000 consumer.max.partition.fetch.bytes: 10485760 # apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaMirrorMaker2 metadata: name: my-mirror-maker2 spec: # mirrors: - sourceCluster: "my-cluster-source" targetCluster: "my-cluster-target" sourceConnector: tasksMax: 2 config: producer.override.batch.size: 327680 producer.override.max.request.size: 10000000 consumer.fetch.max.bytes: 10000000 consumer.max.partition.fetch.bytes: 10485760 # dnf install <package_name> dnf install <path_to_download_package>