Merge "Improve scalability of Kafka"

ansible/roles/kafka/defaults/main.yml

@@ -22,6 +22,7 @@ kafka_cluster_name: "kolla_kafka"
 kafka_log_dir: "/var/log/kolla/kafka"
 kafka_heap_opts: "-Xmx1G -Xms1G"
 kafka_zookeeper: "{% for host in groups['zookeeper'] %}{{ hostvars[host]['ansible_' + hostvars[host]['api_interface']]['ipv4']['address'] }}:{{ zookeeper_client_port }}{% if not loop.last %},{% endif %}{% endfor %}"
+kafka_broker_count: "{{ groups['kafka'] | length }}"
 
 ####################
 # Docker

ansible/roles/kafka/templates/kafka.server.properties.j2

@@ -5,11 +5,12 @@ socket.send.buffer.bytes=102400
 socket.receive.buffer.bytes=102400
 socket.request.max.bytes=104857600
 log.dirs=/var/lib/kafka/data
-num.partitions=1
+min.insync.replicas={{ kafka_broker_count if kafka_broker_count|int < 3 else 2 }}
+num.partitions=30
 num.recovery.threads.per.data.dir=1
-offsets.topic.replication.factor=3
-transaction.state.log.replication.factor=3
-transaction.state.log.min.isr=3
+offsets.topic.replication.factor={{ kafka_broker_count if kafka_broker_count|int < 3 else 3 }}
+transaction.state.log.replication.factor={{ kafka_broker_count if kafka_broker_count|int < 3 else 3 }}
+transaction.state.log.min.isr={{ kafka_broker_count if kafka_broker_count|int < 3 else 2 }}
 log.retention.hours=168
 log.segment.bytes=1073741824
 log.retention.check.interval.ms=300000