# Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # see kafka.server.KafkaConfig for additional details and defaults ############################# Server Basics ############################# #每一个broker在集群中的唯一表示，要求是正数。当该服务器的IP地址发生改变时，broker.id没有变化，则不会影响consumers的消息情况 broker.id=106 #这就是说，这条命令其实并不执行删除动作，仅仅是在zookeeper上标记该topic要被删除而已，同时也提醒用户一定要提前打开delete.topic.enable开关，否则删除动作是不会执行的。 delete.topic.enable=true #是否允许自动创建topic，若是false，就需要通过命令创建topic auto.create.topics.enable=false ############################# Socket Server Settings ############################# # The address the socket server listens on. It will get the value returned from # java.net.InetAddress.getCanonicalHostName() if not configured. # FORMAT: # listeners = listener_name://host_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 #Socket服务器侦听的地址。如果没有配置，它将获得从Java.NET.InAddio.GETCANONICALITHAMEMENE（）返回的值 listeners=PLAINTEXT://node106.yinzhengjie.org.cn:9092 #broker server服务端口 port=9092 #broker的主机地址，若是设置了，那么会绑定到这个地址上，若是没有，会绑定到所有的接口上，并将其中之一发送到ZK，一般不设置 host.name=node106.yinzhengjie.org.cn # Hostname and port the broker will advertise to producers and consumers. If not set, # it uses the value for "listeners" if configured. Otherwise, it will use the value # returned from java.net.InetAddress.getCanonicalHostName(). #kafka 0.9.x以后的版本新增了advertised.listeners配置,kafka 0.9.x以后的版本不要使用 advertised.host.name 和 advertised.host.port 已经deprecated.如果配置的话，它使用 "listeners" 的值。否则，它将使用从java.net.InetAddress.getCanonicalHostName()返回的值。 #advertised.listeners=PLAINTEXT://your.host.name:9092 #将侦听器(listener)名称映射到安全协议，默认情况下它们是相同的。有关详细信息，请参阅配置文档。 #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL #处理网络请求的最大线程数 num.network.threads=30 #处理磁盘I/O的线程数 num.io.threads=30 #套接字服务器使用的发送缓冲区（SOYSNDBUF) socket.send.buffer.bytes=5242880 #套接字服务器使用的接收缓冲区（SOYRCVBUF） socket.receive.buffer.bytes=5242880 #套接字服务器将接受的请求的最大大小（对OOM的保护） socket.request.max.bytes=104857600 #I/O线程等待队列中的最大的请求数，超过这个数量，network线程就不会再接收一个新的请求。应该是一种自我保护机制。 queued.max.requests=1000 ############################# Log Basics ############################# #日志存放目录，多个目录使用逗号分割,如果你有多块磁盘，建议配置成多个目录，从而达到I/O的效率的提升。 log.dirs=/home/data/kafka/logs,/home/data/kafka/logs2,/home/data/kafka/logs3 #每个topic的分区个数，若是在topic创建时候没有指定的话会被topic创建时的指定参数覆盖 num.partitions=20 #在启动时恢复日志和关闭时刷盘日志时每个数据目录的线程的数量，默认1 num.recovery.threads.per.data.dir=30 # 默认副本数 default.replication.factor=2 #服务器接受单个消息的最大大小，即消息体的最大大小，单位是字节 message.max.bytes=104857600 # 自动负载均衡,如果设为true，复制控制器会周期性的自动尝试，为所有的broker的每个partition平衡leadership，为更优先(preferred)的replica分配leadership。 # auto.leader.rebalance.enable=false ############################# Log Flush Policy ############################# #在强制fsync一个partition的log文件之前暂存的消息数量。调低这个值会更频繁的sync数据到磁盘，影响性能。通常建议人家使用replication来确保持久性，而不是依靠单机上的fsync，但是这可以带来更多的可靠性，默认10000。 #log.flush.interval.messages=10000 #2次fsync调用之间最大的时间间隔，单位为ms。即使log.flush.interval.messages没有达到，只要这个时间到了也需要调用fsync。默认3000ms. #log.flush.interval.ms=10000 ############################# Log Retention Policy ############################# # 日志保存时间 (hours|minutes)，默认为7天（168小时）。超过这个时间会根据policy处理数据。bytes和minutes无论哪个先达到都会触发。 log.retention.hours=168 #日志数据存储的最大字节数。超过这个时间会根据policy处理数据。 #log.retention.bytes=1073741824 #控制日志segment文件的大小，超出该大小则追加到一个新的日志segment文件中（-1表示没有限制） log.segment.bytes=536870912 # 当达到下面时间，会强制新建一个segment #log.roll.hours = 24*7 # 日志片段文件的检查周期，查看它们是否达到了删除策略的设置（log.retention.hours或log.retention.bytes） log.retention.check.interval.ms=600000 #是否开启压缩 #log.cleaner.enable=false #日志清理策略选择有：delete和compact主要针对过期数据的处理，或是日志文件达到限制的额度，会被 topic创建时的指定参数覆盖 #log.cleanup.policy=delete # 日志压缩运行的线程数 #log.cleaner.threads=2 # 压缩的日志保留的最长时间 #log.cleaner.delete.retention.ms=3600000 ############################# Zookeeper ############################# #zookeeper集群的地址，可以是多个，多个之间用逗号分割. zookeeper.connect=node106.yinzhengjie.org.cn:2181,node107.yinzhengjie.org.cn:2181,node108.yinzhengjie.org.cn:2181/kafka01 #ZooKeeper的最大超时时间，就是心跳的间隔，若是没有反映，那么认为已经死了，不易过大 zookeeper.session.timeout.ms=180000 #指定多久消费者更新offset到zookeeper中。注意offset更新时基于time而不是每次获得的消息。一旦在更新zookeeper发生异常并重启，将可能拿到已拿到过的消息,连接zk的超时时间 zookeeper.connection.timeout.ms=6000 #请求的最大大小为字节,请求的最大字节数。这也是对最大记录尺寸的有效覆盖。注意：server具有自己对消息记录尺寸的覆盖，这些尺寸和这个设置不同。此项设置将会限制producer每次批量发送请求的数目，以防发出巨量的请求。 max.request.size=104857600 #每次fetch请求中，针对每次fetch消息的最大字节数。这些字节将会督导用于每个partition的内存中，因此，此设置将会控制consumer所使用的memory大小。这个fetch请求尺寸必须至少和server允许的最大消息尺寸相等，否则，producer可能发送的消息尺寸大于consumer所能消耗的尺寸。 fetch.message.max.bytes=104857600 #ZooKeeper集群中leader和follower之间的同步时间,换句话说：一个ZK follower能落后leader多久。 #zookeeper.sync.time.ms=2000 ############################# Replica Basics ############################# # leader接收follower的"fetch请求"的超时时间,默认是10秒。 # replica.lag.time.max.ms=30000 # 如果relicas落后太多,将会认为此partition relicas已经失效。而一般情况下,因为网络延迟等原因,总会导致replicas中消息同步滞后。如果消息严重滞后,leader将认为此relicas网络延迟较大或者消息吞吐能力有限。在broker数量较少,或者网络不足的环境中,建议提高此值.follower落后于leader的最大message数,这个参数是broker全局的。设置太大 了，影响真正“落后”follower的移除;设置的太小了，导致follower的频繁进出。无法给定一个合适的replica.lag.max.messages的值,因此不推荐使用，据说新版本的Kafka移除了这个参数。 #replica.lag.max.messages=4000 # follower与leader之间的socket超时时间 #replica.socket.timeout.ms=30000 # follower每次fetch数据的最大尺寸 replica.fetch.max.bytes=104857600 # follower的fetch请求超时重发时间 replica.fetch.wait.max.ms=2000 # fetch的最小数据尺寸 #replica.fetch.min.bytes=1 #0.11.0.0版本开始unclean.leader.election.enable参数的默认值由原来的true改为false，可以关闭unclean leader election，也就是不在ISR(IN-Sync Replica)列表中的replica，不会被提升为新的leader partition。kafka集群的持久化力大于可用性，如果ISR中没有其它的replica，会导致这个partition不能读写。 unclean.leader.election.enable=false # follower中开启的fetcher线程数, 同步速度与系统负载均衡 num.replica.fetchers=5 # partition leader与replicas之间通讯时,socket的超时时间 #controller.socket.timeout.ms=30000 # partition leader与replicas数据同步时,消息的队列尺寸. #controller.message.queue.size=10 #指定将使用哪个版本的 inter-broker 协议。 在所有经纪人升级到新版本之后，这通常会受到冲击。升级时要设置 #inter.broker.protocol.version=0.10.1 #指定broker将用于将消息添加到日志文件的消息格式版本。 该值应该是有效的ApiVersion。 一些例子是：0.8.2，0.9.0.0，0.10.0。 通过设置特定的消息格式版本，用户保证磁盘上的所有现有消息都小于或等于指定的版本。 不正确地设置这个值将导致使用旧版本的用户出错，因为他们将接收到他们不理解的格式的消息。 #log.message.format.version=0.10.1