当前位置:   article > 正文

kafka1.0.0配置属性Broker、Topic、Producer、Consumer、Connect、Streams、AdminClient_attempting to use idempotence with a broker which

attempting to use idempotence with a broker which does not support the requi

http://kafka.apachecn.org/documentation.html#configuration

3. 配置

Kafkas使用property文件格式的键值对来配置程序。这些键值对配置既可以来自property文件也可以来程序内部。

3.1 Broker 配置

核心基础配置如下:

  • broker.id
  • log.dirs
  • zookeeper.connect

Topic-level配置及其默认值在下面有更详尽的讨论。

名称描述类型默认值有效值重要性
zookeeper.connectZookeeper主机地址string  
advertised.host.name不建议:仅在未设置`advertised.listeners` 或 `listeners`时使用。用`advertised.listeners`替换。 主机名发布到zookeeper供客户端使用。在IaaS环境,这可能需要与broker绑定不通的端口。如果未设置,将使用`host.name`的值(如果已经配置)。否则,他将使用java.net.InetAddress.getCanonicalHostName()返回的值。stringnull 
advertised.listeners监听器发布到ZooKeeper供客户端使用,如果与`listeners`配置不同。在IaaS环境,这可能需要与broker绑定不通的接口。如果没有设置,将使用`listeners`的配置。与`listeners`不同的是,配置0.0.0.0元地址是无效的。stringnull 
advertised.port不建议:仅在未设置“advertised.listeners”或“listeners”时使用。使用`advertised.listeners`代替。 这个端口发布到ZooKeeper供客户端使用。在IaaS环境,这可能需要与broker绑定不通的端口。如果没有设置,它将绑定和broker相同的端口。intnull 
auto.create.topics.enable是否允许在服务器上自动创建topicbooleantrue 
auto.leader.rebalance.enable是否允许leader平衡。后台线程会定期检查并触发leader平衡。booleantrue 
background.threads用于处理各种后台任务的线程数量int10[1,...]
broker.id用于服务的broker id。如果没设置,将生存一个唯一broker id。为了避免ZooKeeper生成的id和用户配置的broker id相冲突,生成的id将在reserved.broker.max.id的值基础上加1。int-1 
compression.type为特点的topic指定一个最终压缩类型。此配置接受的标准压缩编码方式有('gzip', 'snappy', 'lz4')。此外还有'uncompressed'相当于不压缩;'producer'意味着压缩类型由'producer'决定。stringproducer 
delete.topic.enable是否允许删除topic。如果关闭此配置,通过管理工具删除topic将不再生效。booleantrue 
host.name不建议: 仅在未设置`listeners`时使用。使用`listeners`来代替。 如果设置了broker主机名,则他只会当定到这个地址。如果没设置,将绑定到所有接口。string"" 
leader.imbalance.check.interval.seconds由控制器触发分区重新平衡检查的频率设置long300 
leader.imbalance.per.broker.percentage每个broker允许的不平衡的leader的百分比,如果高于这个比值将触发leader进行平衡。这个值用百分比来指定。int10 
listeners监听器列表 - 使用逗号分隔URI列表和监听器名称。如果侦听器名称不是安全协议,则还必须设置listener.security.protocol.map。指定主机名为0.0.0.0来绑定到所有接口。留空则绑定到默认接口上。合法监听器列表的示例:PLAINTEXT:// myhost:9092,SSL://:9091 CLIENT://0.0.0.0:9092,REPLICATION:// localhost:9093stringnull 
log.dir保存日志数据的目录(对log.dirs属性的补充)string/tmp/kafka-logs 
log.dirs保存日志数据的目录,如果未设置将使用log.dir的配置。stringnull 
log.flush.interval.messages在将消息刷新到磁盘之前,在日志分区上累积的消息数量。long9223372036854775807[1,...]
log.flush.interval.ms在刷新到磁盘之前,任何topic中的消息保留在内存中的最长时间(以毫秒为单位)。如果未设置,则使用log.flush.scheduler.interval.ms中的值。longnull 
log.flush.offset.checkpoint.interval.ms日志恢复点的最后一次持久化刷新记录的频率int60000[0,...]
log.flush.scheduler.interval.ms日志刷新器检查是否需要将所有日志刷新到磁盘的频率(以毫秒为单位)long9223372036854775807 
log.flush.start.offset.checkpoint.interval.ms我们更新日志持久化记录开始offset的频率int60000[0,...]
log.retention.bytes日志删除的大小阈值long-1 
log.retention.hours日志删除的时间阈值(小时为单位)int168 
log.retention.minutes日志删除的时间阈值(分钟为单位),如果未设置,将使用log.retention.hours的值intnull 
log.retention.ms日志删除的时间阈值(毫秒为单位),如果未设置,将使用log.retention.minutes的值longnull 
log.roll.hours新日志段轮转时间间隔(小时为单位),次要配置为log.roll.msint168[1,...]
log.roll.jitter.hours从logrolltimemillis(以小时计)中减去的最大抖动,次要配置log.roll.jitter.msint0[0,...]
log.roll.jitter.ms从logrolltimemillis(以毫秒计)中减去的最大抖动,如果未设置,则使用log.roll.jitter.hours的配置longnull 
log.roll.ms新日志段轮转时间间隔(毫秒为单位),如果未设置,则使用log.roll.hours配置longnull 
log.segment.bytes单个日志段文件最大大小int1073741824[14,...]
log.segment.delete.delay.ms从文件系统中删除一个日志段文件前的保留时间long60000[0,...]
message.max.bytes

kafka允许的最大的一个批次的消息大小。 如果这个数字增加,且有0.10.2版本以下的consumer,那么consumer的提取大小也必须增加,以便他们可以取得这么大的记录批次。 在最新的消息格式版本中,记录总是被组合到一个批次以提高效率。 在以前的消息格式版本中,未压缩的记录不会分组到批次中,并且此限制仅适用于该情况下的单个记录。

可以使用topic设置`max.message.bytes`来设置每个topic。 max.message.bytes.

int1000012[0,...]
min.insync.replicas当producer将ack设置为“全部”(或“-1”)时,min.insync.replicas指定了被认为写入成功的最小副本数。如果这个最小值不能满足,那么producer将会引发一个异常(NotEnoughReplicas或NotEnoughReplicasAfterAppend)。当一起使用时,min.insync.replicas和acks允许您强制更大的耐久性保证。 一个经典的情况是创建一个复本数为3的topic,将min.insync.replicas设置为2,并且producer使用“all”选项。 这将确保如果大多数副本没有写入producer则抛出异常。int1[1,...]
num.io.threads服务器用于处理请求的线程数,可能包括磁盘I/Oint8[1,...]
num.network.threads服务器用于从接收网络请求并发送网络响应的线程数int3[1,...]
num.recovery.threads.per.data.dir每个数据目录,用于启动时日志恢复和关闭时刷新的线程数int1[1,...]
num.replica.fetchers从源broker复制消息的拉取器的线程数。增加这个值可以增加follow broker的I/O并行度。int1 
offset.metadata.max.bytes与offset提交相关联的元数据条目的最大大小int4096 
offsets.commit.required.acks在offset提交可以接受之前,需要设置acks的数目,一般不需要更改,默认值为-1。short-1 
offsets.commit.timeout.msoffset提交将延迟到topic所有副本收到提交或超时。这与producer请求超时类似。int5000[1,...]
offsets.load.buffer.size每次从offset段文件往缓存加载时,批量读取的数据大小int5242880[1,...]
offsets.retention.check.interval.ms检查失效offset的频率long600000[1,...]
offsets.retention.minutes超过这个保留期限未提交的offset将被丢弃int1440[1,...]
offsets.topic.compression.codec用于offsets topic的压缩编解码器 - 压缩可用于实现“原子”提交int0 
offsets.topic.num.partitionsOffsets topic的分区数量(部署后不应更改)int50[1,...]
offsets.topic.replication.factoroffset topic的副本数(设置的越大,可用性越高)。内部topic创建将失败,直到集群大小满足此副本数要求。short3[1,...]
offsets.topic.segment.bytes为了便于更快的日志压缩和缓存加载,offset topic段字节应该保持相对较小int104857600[1,...]
port不建议: 仅在未设置“listener”时使用。使用`listeners`来代替。端口用来来监听和接受连接int9092 
queued.max.requests网络线程阻塞前队列允许的最大请求数int500[1,...]
quota.consumer.default不建议:仅在动态默认配额未配置或在zookeeper中使用。任何由clientid区分开来的consumer,如果它每秒产生的字节数多于这个值,就会受到限制long9223372036854775807[1,...]
quota.producer.default不建议:仅在动态默认配额未配置或在zookeeper中使用。任何由clientid区分开来的producer,如果它每秒产生的字节数多于这个值,就会受到限制long9223372036854775807[1,...]
replica.fetch.min.bytes复制数据过程中,replica收到的每个fetch响应,期望的最小的字节数,如果没有收到足够的字节数,就会等待更多的数据,直到达到replicaMaxWaitTimeMs(复制数据超时时间)int1 
replica.fetch.wait.max.ms副本follow同leader之间通信的最大等待时间,失败了会重试。 此值始终应始终小于replica.lag.time.max.ms,以防止针对低吞吐量topic频繁收缩ISRint500 
replica.high.watermark.checkpoint.interval.mshigh watermark被保存到磁盘的频率,用来标记日后恢复点/td>long5000 
replica.lag.time.max.ms如果一个follower在这个时间内没有发送fetch请求或消费leader日志到结束的offset,leader将从ISR中移除这个follower,并认为这个follower已经挂了long10000 
replica.socket.receive.buffer.bytessocket接收网络请求的缓存大小int65536 
replica.socket.timeout.ms副本复制数据过程中,发送网络请求的socket超时时间。这个值应该大于replica.fetch.wait.max.ms的值int30000 
request.timeout.ms该配置控制客户端等待请求响应的最长时间。如果在超时之前未收到响应,则客户端将在必要时重新发送请求,如果重试仍然失败,则请求失败。int30000 
socket.receive.buffer.bytes服务端用来处理socket连接的SO_RCVBUFF缓冲大小。如果值为-1,则使用系统默认值。int102400 
socket.request.max.bytessocket请求的最大大小,这是为了防止server跑光内存,不能大于Java堆的大小。int104857600[1,...]
socket.send.buffer.bytes服务端用来处理socket连接的SO_SNDBUF缓冲大小。如果值为-1,则使用系统默认值。int102400 
transaction.max.timeout.ms事务允许的最大超时时间。如果客户请求的事务超时,那么broker将在InitProducerIdRequest中返回一错误。 这样可以防止客户超时时间过长,从而阻碍consumers读取事务中包含的topic。int900000[1,...]
transaction.state.log.load.buffer.size将producer ID和事务加载到高速缓存中时,从事务日志段(the transaction log segments)中批量读取的大小。int5242880[1,...]
transaction.state.log.min.isr覆盖事务topic的min.insync.replicas配置int2[1,...]
transaction.state.log.num.partitions事务topic的分区数(部署后不应该修改)int50[1,...]
transaction.state.log.replication.factor事务topic的副本数(设置的越大,可用性越高)。内部topic在集群数满足副本数之前,将会一直创建失败。short3[1,...]
transaction.state.log.segment.bytes事务topic段应保持相对较小,以便于更快的日志压缩和缓存负载。int104857600[1,...]
transactional.id.expiration.ms事务协调器在未收到任何事务状态更新之前,主动设置producer的事务标识为过期之前将等待的最长时间(以毫秒为单位)int604800000[1,...]
unclean.leader.election.enable指定副本是否能够不再ISR中被选举为leader,即使这样可能会丢数据booleanfalse 
zookeeper.connection.timeout.ms与ZK server建立连接的超时时间,没有配置就使用zookeeper.session.timeout.msintnull 
zookeeper.session.timeout.msZooKeeper的session的超时时间int6000 
zookeeper.set.aclZooKeeper客户端连接是否设置ACL安全y安装booleanfalse 
broker.id.generation.enable是否允许服务器自动生成broker.id。如果允许则产生的值会交由reserved.broker.max.id审核booleantrue 
broker.rackbroker的机架位置。 这将在机架感知副本分配中用于容错。例如:RACK1,us-east-1stringnull 
connections.max.idle.ms连接空闲超时:服务器socket处理线程空闲超时关闭时间long600000 
controlled.shutdown.enable是否允许服务器关闭broker服务booleantrue 
controlled.shutdown.max.retries当发生失败故障时,由于各种原因导致关闭服务的次数int3 
controlled.shutdown.retry.backoff.ms在每次重试关闭之前,系统需要时间从上次故障状态(控制器故障切换,副本延迟等)中恢复。 这个配置决定了重试之前等待的时间。long5000 
controller.socket.timeout.ms控制器到broker通道的socket超时时间int30000 
default.replication.factor自动创建topic时的默认副本个数int1 
delete.records.purgatory.purge.interval.requests删除purgatory中请求的清理间隔时间(purgatory:broker对于无法立即处理的请求,将会放在purgatory中,当请求完成后,并不会立即清除,还会继续在purgatory中占用资源,直到下一次delete.records.purgatory.purge.interval.requests)int1 
fetch.purgatory.purge.interval.requests提取purgatory中请求的间隔时间int1000 
group.initial.rebalance.delay.ms在执行第一次重新平衡之前,group协调器将等待更多consumer加入group的时间。延迟时间越长意味着重新平衡的工作可能越小,但是等待处理开始的时间增加。int3000 
group.max.session.timeout.msconsumer注册允许的最大会话超时时间。超时时间越短,处理心跳越频繁从而使故障检测更快,但会导致broker被抢占更多的资源。int300000 medium
group.min.session.timeout.msconsumer注册允许的最小会话超时时间。超时时间越短,处理心跳越频繁从而使故障检测更快,但会导致broker被抢占更多的资源。int6000 
inter.broker.listener.namebroker间通讯的监听器名称。如果未设置,则侦听器名称由security.inter.broker.protocol定义。 同时设置此项和security.inter.broker.protocol属性是错误的,只设置一个。stringnull 
inter.broker.protocol.version指定使用哪个版本的 inter-broker 协议。 在所有broker升级到新版本之后,这通常会有冲突。一些有效的例子是:0.8.0, 0.8.1, 0.8.1.1, 0.8.2, 0.8.2.0, 0.8.2.1, 0.9.0.0, 0.9.0.1,详情可以检查apiversion的完整列表string1.0-IV0 
log.cleaner.backoff.ms检查log是否需要清除的时间间隔。long15000[0,...]
log.cleaner.dedupe.buffer.size日志去重清理线程所需要的内存long134217728 
log.cleaner.delete.retention.ms日志记录保留时间long86400000 
log.cleaner.enable在服务器上启用日志清理器进程。如果任何topic都使用cleanup.policy = compact,包括内部topic offset,则建议开启。如果被禁用的话,这些topic将不会被压缩,而且会不断增长。booleantrue 
log.cleaner.io.buffer.load.factor日志清理器去重的缓存负载数。完全重复数据的缓存比例可以改变。数值越高,清理的越多,但会导致更多的hash冲突double0.9 
log.cleaner.io.buffer.size所有清理线程的日志清理I/O缓存区所需要的内存int524288[0,...]
log.cleaner.io.max.bytes.per.second日志清理器受到的大小限制数,因此它的I/O读写总和将小于平均值double1.7976931348623157E308 
log.cleaner.min.cleanable.ratio日志中脏数据清理比例double0.5 
log.cleaner.min.compaction.lag.ms消息在日志中保持未压缩的最短时间。 仅适用于正在压缩的日志。long0 
log.cleaner.threads用于日志清理的后台线程的数量int1[0,...]
log.cleanup.policy超出保留窗口期的日志段的默认清理策略。用逗号隔开有效策略列表。有效策略:“delete”和“compact”listdelete[compact, delete]
log.index.interval.bytes添加offset索引字段大小间隔(设置越大,代表扫描速度越快,但是也更耗内存)int4096[0,...]
log.index.size.max.bytesoffset索引的最大字节数int10485760[4,...]
log.message.format.version指定broker用于将消息附加到日志的消息格式版本。应该是一个有效的apiversion值。例如:0.8.2,0.9.0.0,0.10.0,详情去看apiversion。通过设置特定的消息格式版本,用户得保证磁盘上的所有现有消息的版本小于或等于指定的版本。不正确地设置这个值会导致旧版本的用户出错,因为他们将接收到他们无法处理的格式消息。string1.0-IV0 
log.message.timestamp.difference.max.msbroker收到消息时的时间戳和消息中指定的时间戳之间允许的最大差异。当log.message.timestamp.type=CreateTime,如果时间差超过这个阈值,消息将被拒绝。如果log.message.timestamp.type = logappendtime,则该配置将被忽略。允许的最大时间戳差值,不应大于log.retention.ms,以避免不必要的频繁日志滚动。long9223372036854775807 
log.message.timestamp.type定义消息中的时间戳是消息创建时间还是日志追加时间。 该值应该是“createtime”或“logappendtime”。stringCreateTime[CreateTime, LogAppendTime]
log.preallocate创建新的日志段前是否应该预先分配文件?如果你在windows上使用kafka,你可能需要打开个这个选项booleanfalse 
log.retention.check.interval.ms日志清理器检查是否有日志符合删除的频率(以毫秒为单位)long300000[1,...]
max.connections.per.ip每个IP允许的最大连接数int2147483647[1,...]
max.connections.per.ip.overrides每个IP或主机名将覆盖默认的最大连接数string"" 
num.partitions每个topic的默认日志分区数int1[1,...]
principal.builder.class实现kafkaprincipalbuilder接口类的全名,该接口用于构建授权期间使用的kafkaprincipal对象。此配置还支持以前已弃用的用于ssl客户端身份验证的principalbuilder接口。如果未定义主体构建器,则默认采用所使用的安全协议。对于ssl身份验证,如果提供了一个主体名称,主体名称将是客户端证书的专有名称;否则,如果不需要客户端身份验证,则主体名称将是匿名的。对于sasl身份验证,如果使用gssapi,则将使用由sasl.kerberos.principal.to.local.rules定义的规则来生成主体,而使用其他机制的sasl身份验证ID。若果用明文,委托人将是匿名的。classnull 
producer.purgatory.purge.interval.requestsproducer请求purgatory的清除间隔(请求数量)int1000 
queued.max.request.bytes在不再读取请求之前队列的字节数long-1 
replica.fetch.backoff.ms当拉取分区发生错误时,睡眠的时间。int1000[0,...]
replica.fetch.max.bytes尝试提取每个分区的消息的字节数。这并不是绝对最大值,如果第一个非空分区的第一个批量记录大于这个值,那么批处理仍将被执行并返回,以确保进度可以正常进行下去。broker接受的最大批量记录大小通过message.max.bytes(broker配置)或max.message.bytes(topic配置)进行配置。int1048576[0,...]medium
replica.fetch.response.max.bytes预计整个获取响应的最大字节数。记录被批量取回时,如果取第一个非空分区的第一个批量记录大于此值,记录的批处理仍将被执行并返回以确保可以进行下去。因此,这不是绝对的最大值。 broker接受的最大批量记录大小通过message.max.bytes(broker配置)或max.message.bytes(topic配置)进行配置。int10485760[0,...]
reserved.broker.max.id可以用于broker.id的最大数量int1000[0,...]
sasl.enabled.mechanismskafka服务器中启用的sasl机制的列表。 该列表可能包含安全提供程序可用的任何机制。默认情况下只有gssapi是启用的。listGSSAPI 
sasl.kerberos.kinit.cmdKerberos kinit 命令路径。string/usr/bin/kinit 
sasl.kerberos.min.time.before.relogin登录线程在尝试刷新间隔内的休眠时间。long60000 
sasl.kerberos.principal.to.local.rules主体名称到简称映射的规则列表(通常是操作系统用户名)。按顺序,使用与principal名称匹配的第一个规则将其映射到简称。列表中的任何后续规则都将被忽略。 默认情况下,{username} / {hostname} @ {realm}形式的主体名称映射到{username}。 有关格式的更多细节,请参阅安全授权和acls。 请注意,如果由principal.builder.class配置提供了kafkaprincipalbuilder的扩展,则忽略此配置。listDEFAULT 
sasl.kerberos.service.namekafka运行的kerberos的主体名称。 这可以在kafka的JAAS配置或在kafka的配置中定义。stringnull 
sasl.kerberos.ticket.renew.jitter添加到更新时间的随机抖动的百分比double0.05 
sasl.kerberos.ticket.renew.window.factor登录线程将休眠,直到从上次刷新到ticket的到期的时间到达(指定窗口因子),在此期间它将尝试更新ticket。double0.8 
sasl.mechanism.inter.broker.protocolSASL机制,用于broker之间的通讯,默认是GSSAPI。stringGSSAPI 
security.inter.broker.protocolbroker之间的安全通讯协议,有效值有:PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL。同时设置此配置和inter.broker.listener.name属性会出错stringPLAINTEXT 
ssl.cipher.suites密码套件列表。 这是一种用于使用tls或ssl网络协议来协商网络连接的安全设置的认证,加密,mac和密钥交换算法的命名组合。 默认情况下,所有可用的密码套件都受支持。listnull 
ssl.client.auth配置请求客户端的broker认证。常见的设置:
  • ssl.client.auth=required如果设置需要客户端认证。
  • ssl.client.auth=requested客户端认证可选,不同于requested,客户端可选择不提供自身的身份验证信息。
  • ssl.client.auth=none 不需要客户端身份认证。
stringnone[required, requested, none]
ssl.enabled.protocols已启用的SSL连接协议列表。listTLSv1.2,TLSv1.1,TLSv1 
ssl.key.password秘钥库文件中的私钥密码。对客户端是可选的。passwordnull 
ssl.keymanager.algorithm用于SSL连接的密钥管理工厂算法。默认值是为Java虚拟机配置的密钥管理器工厂算法。stringSunX509 
ssl.keystore.location密钥仓库文件的位置。客户端可选,并可用于客户端的双向认证。stringnull 
ssl.keystore.password密钥仓库文件的仓库密码。客户端可选,只有ssl.keystore.location配置了才需要。passwordnull 
ssl.keystore.type密钥仓库文件的格式。客户端可选。stringJKS 
ssl.protocol用于生成SSLContext,默认是TLS,适用于大多数情况。允许使用最新的JVM,LS, TLSv1.1 和TLSv1.2。 SSL,SSLv2和SSLv3 老的JVM也可能支持,但由于有已知的安全漏洞,不建议使用。stringTLS  
ssl.provider用于SSL连接的安全提供程序的名称。默认值由JVM的安全程序提供。stringnull 
ssl.trustmanager.algorithm信任管理工厂用于SSL连接的算法。默认为Java虚拟机配置的信任算法。stringPKIX 
ssl.truststore.location信任文件的存储位置。stringnull 
ssl.truststore.password信任存储文件的密码。 如果密码未设置,则仍然可以访问信任库,但完整性检查将被禁用。passwordnull 
ssl.truststore.type信任存储文件的文件格式。stringJKS 
alter.config.policy.class.name应该用于验证的alter configs策略类。 该类应该实现org.apache.kafka.server.policy.alterconfigpolicy接口。classnull 
authorizer.class.name用于认证授权的程序类string"" 
create.topic.policy.class.name用于验证的创建topic策略类。 该类应该实现org.apache.kafka.server.policy.createtopicpolicy接口。classnull 
listener.security.protocol.map侦听器名称和安全协议之间的映射。必须定义为相同的安全协议可用于多个端口或IP。例如,即使两者都需要ssl,内部和外部流量也可以分开。具体的说,用户可以定义名字为INTERNAL和EXTERNAL的侦听器,这个属性为:internal:ssl,external:ssl。 如图所示,键和值由冒号分隔,映射条目以逗号分隔。 每个监听者名字只能在映射表上出现一次。 通过向配置名称添加规范化前缀(侦听器名称小写),可以为每个侦听器配置不同的安全性(ssl和sasl)设置。 例如,为内部监听器设置不同的密钥仓库,将会设置名称为“listener.name.internal.ssl.keystore.location”的配置。 如果没有设置侦听器名称的配置,配置将回退到通用配置(即`ssl.keystore.location`)。stringPLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL 
metric.reporters度量报告的类列表,通过实现MetricReporter接口,允许插入新度量标准类。JmxReporter包含注册JVM统计。list"" 
metrics.num.samples维持计算度量的样本数int2[1,...]
metrics.recording.level指标的最高记录级别stringINFO 
metrics.sample.window.ms计算度量样本的时间窗口long30000[1,...]
quota.window.num在内存中保留客户端限额的样本数int11[1,...]
quota.window.size.seconds每个客户端限额的样本时间跨度int1[1,...]
replication.quota.window.num在内存中保留副本限额的样本数int11[1,...]
replication.quota.window.size.seconds每个副本限额样本数的时间跨度int1[1,...]
ssl.endpoint.identification.algorithm端点身份标识算法,使用服务器证书验证服务器主机名stringnull 
ssl.secure.random.implementation用于SSL加密操作的SecureRandom PRNG实现stringnull 
transaction.abort.timed.out.transaction.cleanup.interval.ms回滚已超时的事务的时间间隔int60000[1,...]
transaction.remove.expired.transaction.cleanup.interval.ms删除由于transactional.id.expiration.ms传递过程而过期的事务的时间间隔int3600000[1,...]low
zookeeper.sync.time.msZK follower同步可落后leader多久/td>int2000 

More details about broker configuration can be found in the scala class kafka.server.KafkaConfig.

3.2 Topic级别配置

与Topic相关的配置既包含服务器默认值,也包含可选的每个Topic覆盖值。 如果没有给出每个Topic的配置,那么服务器默认值就会被使用。 通过提供一个或多个 --config 选项,可以在创建Topic时设置覆盖值。 本示例使用自定义的最大消息大小和刷新率创建了一个名为 my-topic 的topic:

1

2

> bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic my-topic --partitions 1

    --replication-factor 1 --config max.message.bytes=64000 --config flush.messages=1

也可以在使用alter configs命令稍后更改或设置覆盖值. 本示例重置my-topic的最大消息的大小:

1

2

> bin/kafka-configs.sh --zookeeper localhost:2181 --entity-type topics --entity-name my-topic

    --alter --add-config max.message.bytes=128000

您可以执行如下操作来检查topic设置的覆盖值

1

> bin/kafka-configs.sh --zookeeper localhost:2181 --entity-type topics --entity-name my-topic --describe

您可以执行如下操作来删除一个覆盖值

1

> bin/kafka-configs.sh --zookeeper localhost:2181  --entity-type topics --entity-name my-topic --alter --delete-config max.message.bytes

以下是Topic级别配置。 “服务器默认属性”列是该属性的默认配置。 一个Topic如果没有给出一个明确的覆盖值,相应的服务器默认配置将会生效。

名称描述类型默认值有效值服务器默认属性重要性
cleanup.policy该配置项可以是 "delete" 或 "compact"。 它指定在旧日志段上使用的保留策略。 默认策略 ("delete") 将在达到保留时间或大小限制时丢弃旧段。 "compact" 设置将启用该topic的日志压缩 。listdelete[compact, delete]log.cleanup.policymedium
compression.type为给定的topic指定最终压缩类型。这个配置接受标准的压缩编解码器 ('gzip', 'snappy', lz4) 。它为'uncompressed'时意味着不压缩,当为'producer'时,这意味着保留producer设置的原始压缩编解码器。stringproducer[uncompressed, snappy, lz4, gzip, producer]compression.typemedium
delete.retention.ms保留 日志压缩 topics的删除墓碑标记的时间。此设置还对consumer从偏移量0开始时必须完成读取的时间进行限制,以确保它们获得最后阶段的有效快照(否则,在完成扫描之前可能会收集到删除墓碑)。long86400000[0,...]log.cleaner.delete.retention.msmedium
file.delete.delay.ms删除文件系统上的一个文件之前所需等待的时间。long60000[0,...]log.segment.delete.delay.msmedium
flush.messages这个设置允许指定一个时间间隔n,每隔n个消息我们会强制把数据fsync到log。例如,如果设置为1,我们会在每条消息之后同步。如果是5,我们会在每五个消息之后进行fsync。一般来说,我们建议您不要设置它,而是通过使用replication机制来持久化数据,和允许更高效的操作系统后台刷新功能。这个设置可以针对每个topic的情况自定义 (请参阅 topic的配置部分).long9223372036854775807[0,...]log.flush.interval.messagesmedium
flush.ms这个设置允许指定一个时间间隔,每隔一段时间我们将强制把数据fsync到log。例如,如果这个设置为1000,我们将在1000 ms后执行fsync。一般来说,我们建议您不要设置它,而是通过使用replication机制来持久化数据,和允许更高效的操作系统后台刷新功能。long9223372036854775807[0,...]log.flush.interval.msmedium
follower.replication.throttled.replicas应该在follower侧限制日志复制的副本列表。该列表应以[PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:...的形式描述一组副本,或者也可以使用通配符“*”来限制该topic的所有副本。list""[partitionId],[brokerId]:[partitionId],[brokerId]:...follower.replication.throttled.replicasmedium
index.interval.bytes此设置控制Kafka向其偏移索引添加索引条目的频率。默认设置确保我们大约每4096个字节索引一条消息。更多的索引允许读取更接近日志中的确切位置,但这会使索引更大。您可能不需要改变该值。int4096[0,...]log.index.interval.bytesmedium
leader.replication.throttled.replicas应该在leader侧限制日志复制的副本列表。该列表应以[PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:...的形式描述一组副本,或者也可以使用通配符“*”来限制该topic的所有副本。list""[partitionId],[brokerId]:[partitionId],[brokerId]:...leader.replication.throttled.replicasmedium
max.message.bytes

Kafka允许的最大记录批次大小。如果这个参数被增加了且consumers是早于0.10.2版本,那么consumers的fetch size必须增加到该值,以便他们可以取得这么大的记录批次。

在最新的消息格式版本中,记录总是分组成多个批次以提高效率。在以前的消息格式版本中,未压缩的记录不会分组到多个批次,并且限制在该情况下只能应用单条记录。

int1000012[0,...]message.max.bytesmedium
message.format.version指定broker将用于将消息附加到日志的消息格式版本。该值应该是有效的ApiVersion。如:0.8.2,0.9.0.0,0.10.0,查看ApiVersion获取更多细节。通过设置特定的消息格式版本,用户将发现磁盘上的所有现有消息都小于或等于指定的版本。不正确地设置此值将导致旧版本的使用者中断,因为他们将收到他们不理解的格式的消息。string1.0-IV0 log.message.format.versionmedium
message.timestamp.difference.max.msbroker接收消息时所允许的时间戳与消息中指定的时间戳之间的最大差异。如果message.timestamp.type=CreateTime,则如果时间戳的差异超过此阈值,则将拒绝消息。如果message.timestamp.type=LogAppendTime,则忽略此配置。long9223372036854775807[0,...]log.message.timestamp.difference.max.msmedium
message.timestamp.type定义消息中的时间戳是消息创建时间还是日志附加时间。值应该是“CreateTime”或“LogAppendTime”stringCreateTime log.message.timestamp.typemedium
min.cleanable.dirty.ratio此配置控制日志compaction程序尝试清理日志的频率(假设启用了log compaction )。默认情况下,我们将避免清除超过50%的日志已经合并的日志。这个比率限制了重复在日志中浪费的最大空间(最多为50%,日志中最多有50%可能是重复的)。一个更高的比率将意味着更少,更高效的清理,但将意味着在日志中浪费更多的空间。double0.5[0,...,1]log.cleaner.min.cleanable.ratiomedium
min.compaction.lag.ms消息在日志中保持未压缩的最短时间。仅适用于被合并的日志。long0[0,...]log.cleaner.min.compaction.lag.msmedium
min.insync.replicas当producer将ack设置为“all”(或“-1”)时,此配置指定必须确认写入才能被认为成功的副本的最小数量。如果这个最小值无法满足,那么producer将引发一个异常(NotEnough Replicas或NotEnough ReplicasAfterAppend)。
当使用时,min.insync.Copicas和ack允许您执行更好的持久化保证。一个典型的场景是创建一个复制因子为3的topic,将min.insync.Copicas设置为2,并生成带有“All”的ack。这将确保如果大多数副本没有接收到写,则producer将引发异常。
int1[1,...]min.insync.replicasmedium
preallocate如果在创建新的日志段时应该预先分配磁盘上的文件,则为True。booleanfalse log.preallocatemedium
retention.bytes如果使用“delete”保留策略,此配置控制分区(由日志段组成)在放弃旧日志段以释放空间之前的最大大小。默认情况下,没有大小限制,只有时间限制。由于此限制是在分区级别强制执行的,因此,将其乘以分区数,计算出topic保留值,以字节为单位。long-1 log.retention.bytesmedium
retention.ms如果使用“delete”保留策略,此配置控制保留日志的最长时间,然后将旧日志段丢弃以释放空间。这代表了用户读取数据的速度的SLA。long604800000 log.retention.msmedium
segment.bytes此配置控制日志的段文件大小。保留和清理总是一次完成一个文件,所以更大的段大小意味着更少的文件,但对保留的粒度控制更少。int1073741824[14,...]log.segment.bytesmedium
segment.index.bytes此配置控制将偏移量映射到文件位置的索引大小。我们预先分配这个索引文件并且只在日志滚动后收缩它。您通常不需要更改此设置。int10485760[0,...]log.index.size.max.bytesmedium
segment.jitter.ms从预定的分段滚动时间减去最大随机抖动,以避免段滚动产生惊群效应。long0[0,...]log.roll.jitter.msmedium
segment.ms这个配置控制在一段时间后,Kafka将强制日志滚动,即使段文件没有满,以确保保留空间可以删除或合并旧数据。long604800000[0,...]log.roll.msmedium
unclean.leader.election.enable指示是否启用不在ISR集合中的副本选为领导者作为最后的手段,即使这样做可能导致数据丢失。booleanfalse unclean.leader.election.enablemedium

3.3 Producer 配置

以下是JAVA生产者的配置:

NAMEDESCRIPTIONTYPEDEFAULTVALID VALUESIMPORTANCE
bootstrap.servers这是一个用于建立初始连接到kafka集群的"主机/端口对"配置列表。不论这个参数配置了哪些服务器来初始化连接,客户端都是会均衡地与集群中的所有服务器建立连接。—配置的服务器清单仅用于初始化连接,以便找到集群中的所有服务器。配置格式: host1:port1,host2:port2,.... 由于这些主机是用于初始化连接,以获得整个集群(集群是会动态变化的),因此这个配置清单不需要包含整个集群的服务器。(当然,为了避免单节点风险,这个清单最好配置多台主机)。list  high
key.serializer关键字的序列化类,实现以下接口: org.apache.kafka.common.serialization.Serializer 接口。class  high
value.serializer值的序列化类,实现以下接口: org.apache.kafka.common.serialization.Serializer 接口。class  high
acks此配置是 Producer 在确认一个请求发送完成之前需要收到的反馈信息的数量。 这个参数是为了保证发送请求的可靠性。以下配置方式是允许的:
  • acks=0 如果设置为0,则 producer 不会等待服务器的反馈。该消息会被立刻添加到 socket buffer 中并认为已经发送完成。在这种情况下,服务器是否收到请求是没法保证的,并且参数retries也不会生效(因为客户端无法获得失败信息)。每个记录返回的 offset 总是被设置为-1。
  • acks=1 如果设置为1,leader节点会将记录写入本地日志,并且在所有 follower 节点反馈之前就先确认成功。在这种情况下,如果 leader 节点在接收记录之后,并且在 follower 节点复制数据完成之前产生错误,则这条记录会丢失。
  • acks=all 如果设置为all,这就意味着 leader 节点会等待所有同步中的副本确认之后再确认这条记录是否发送完成。只要至少有一个同步副本存在,记录就不会丢失。这种方式是对请求传递的最有效保证。acks=-1与acks=all是等效的。
string1[all, -1, 0, 1]high
buffer.memoryProducer 用来缓冲等待被发送到服务器的记录的总字节数。如果记录发送的速度比发送到服务器的速度快, Producer 就会阻塞,如果阻塞的时间超过 max.block.ms 配置的时长,则会抛出一个异常。

这个配置与 Producer 的可用总内存有一定的对应关系,但并不是完全等价的关系,因为 Producer 的可用内存并不是全部都用来缓存。一些额外的内存可能会用于压缩(如果启用了压缩),以及维护正在运行的请求。

long33554432[0,...]high
compression.typeProducer 生成数据时可使用的压缩类型。默认值是none(即不压缩)。可配置的压缩类型包括:nonegzipsnappy, 或者 lz4 。压缩是针对批处理的所有数据,所以批处理的效果也会影响压缩比(更多的批处理意味着更好的压缩)。stringnone high
retries若设置大于0的值,则客户端会将发送失败的记录重新发送,尽管这些记录有可能是暂时性的错误。请注意,这种 retry 与客户端收到错误信息之后重新发送记录并无区别。允许 retries 并且没有设置max.in.flight.requests.per.connection 为1时,记录的顺序可能会被改变。比如:当两个批次都被发送到同一个 partition ,第一个批次发生错误并发生 retries 而第二个批次已经成功,则第二个批次的记录就会先于第一个批次出现。int0[0,...,2147483647]high
ssl.key.passwordkey store 文件中私钥的密码。这对于客户端来说是可选的。passwordnull high
ssl.keystore.locationkey store 文件的位置。这对于客户端来说是可选的,可用于客户端的双向身份验证。stringnull high
ssl.keystore.passwordkey store 文件的密码。这对于客户端是可选的,只有配置了 ssl.keystore.location 才需要配置该选项。passwordnull high
ssl.truststore.locationtrust store 文件的位置。stringnull high
ssl.truststore.passwordtrust store 文件的密码。如果一个密码没有设置到 trust store ,这个密码仍然是可用的,但是完整性检查是禁用的。passwordnull high
batch.size当将多个记录被发送到同一个分区时, Producer 将尝试将记录组合到更少的请求中。这有助于提升客户端和服务器端的性能。这个配置控制一个批次的默认大小(以字节为单位)。

当记录的大小超过了配置的字节数, Producer 将不再尝试往批次增加记录。

发送到 broker 的请求会包含多个批次的数据,每个批次对应一个 partition 的可用数据

小的 batch.size 将减少批处理,并且可能会降低吞吐量(如果 batch.size = 0的话将完全禁用批处理)。 很大的 batch.size 可能造成内存浪费,因为我们一般会在 batch.size 的基础上分配一部分缓存以应付额外的记录。

int16384[0,...]medium
client.id发出请求时传递给服务器的 ID 字符串。这样做的目的是为了在服务端的请求日志中能够通过逻辑应用名称来跟踪请求的来源,而不是只能通过IP和端口号跟进。string"" medium
connections.max.idle.ms在此配置指定的毫秒数之后,关闭空闲连接。long540000 medium
linger.msproducer 会将两个请求发送时间间隔内到达的记录合并到一个单独的批处理请求中。通常只有当记录到达的速度超过了发送的速度时才会出现这种情况。然而,在某些场景下,即使处于可接受的负载下,客户端也希望能减少请求的数量。这个设置是通过添加少量的人为延迟来实现的—即,与其立即发送记录, producer 将等待给定的延迟时间,以便将在等待过程中到达的其他记录能合并到本批次的处理中。这可以认为是与 TCP 中的 Nagle 算法类似。这个设置为批处理的延迟提供了上限:一旦我们接受到记录超过了分区的 batch.size ,Producer 会忽略这个参数,立刻发送数据。但是如果累积的字节数少于 batch.size ,那么我们将在指定的时间内“逗留”(linger),以等待更多的记录出现。这个设置默认为0(即没有延迟)。例如:如果设置linger.ms=5 ,则发送的请求会减少并降低部分负载,但同时会增加5毫秒的延迟。long0[0,...]medium
max.block.ms该配置控制 KafkaProducer.send()KafkaProducer.partitionsFor() 允许被阻塞的时长。这些方法可能因为缓冲区满了或者元数据不可用而被阻塞。用户提供的序列化程序或分区程序的阻塞将不会被计算到这个超时。long60000[0,...]medium
max.request.size请求的最大字节数。这个设置将限制 Producer 在单个请求中发送的记录批量的数量,以避免发送巨大的请求。这实际上也等同于批次的最大记录数的限制。请注意,服务器对批次的大小有自己的限制,这可能与此不同。int1048576[0,...]medium
partitioner.class指定计算分区的类,实现 org.apache.kafka.clients.producer.Partitioner 接口。classorg.apache.kafka.clients.producer.internals.DefaultPartitioner medium
receive.buffer.bytes定义读取数据时 TCP 接收缓冲区(SO_RCVBUF)的大小,如果设置为-1,则使用系统默认值。int32768[-1,...]medium
request.timeout.ms客户端等待请求响应的最大时长。如果超时未收到响应,则客户端将在必要时重新发送请求,如果重试的次数达到允许的最大重试次数,则请求失败。这个参数应该比 replica.lag.time.max.ms (Broker 的一个参数)更大,以降低由于不必要的重试而导致的消息重复的可能性。int30000[0,...]medium
sasl.jaas.configSASL 连接使用的 JAAS 登陆上下文参数,以 JAAS 配置文件的格式进行配置。 JAAS 配置文件格式可参考这里。值的格式: ' (=)*;'passwordnull medium
sasl.kerberos.service.nameKafka 运行时的 Kerberos 主体名称。可以在 Kafka 的 JAAS 配置文件或者 Kafka 的配置文件中配置。stringnull medium
sasl.mechanism用于客户端连接的 SASL 机制。可以是任意安全可靠的机制。默认是 GSSAPI 机制。stringGSSAPI medium
security.protocol与 brokers 通讯的协议。可配置的值有: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.stringPLAINTEXT medium
send.buffer.bytes定义发送数据时的 TCP 发送缓冲区(SO_SNDBUF)的大小。如果设置为-1,则使用系统默认值。int131072[-1,...]medium
ssl.enabled.protocols可用于 SSL 连接的协议列表。listTLSv1.2,TLSv1.1,TLSv1 medium
ssl.keystore.typekey store 文件的文件格类型。这对于客户端来说是可选的。stringJKS medium
ssl.protocol用于生成SSLContext的SSL协议。默认设置是TLS,大多数情况下不会有问题。在最近的jvm版本中,允许的值是TLS、tlsv1.1和TLSv1.2。在旧的jvm中可能会支持SSL、SSLv2和SSLv3,但是由于存在已知的安全漏洞,因此不建议使用。stringTLS medium
ssl.provider用于 SSL 连接security provider 。默认值是当前 JVM 版本的默认 security provider 。stringnull medium
ssl.truststore.typetrust store 的文件类型。stringJKS medium
enable.idempotence当设置为true时, Producer 将确保每个消息在 Stream 中只写入一个副本。如果为false,由于 Broker 故障导致 Producer 进行重试之类的情况可能会导致消息重复写入到 Stream 中。请注意,启用幂等性需要确保 max.in.flight.requests.per.connection小于或等于5,retries 大于等于0,并且ack必须设置为all 。如果这些值不是由用户明确设置的,那么将自动选择合适的值。如果设置了不兼容的值,则将抛出一个ConfigException的异常。booleanfalse low
interceptor.classes配置 interceptor 类的列表。实现 org.apache.kafka.clients.producer.ProducerInterceptor接口之后可以拦截(并可能改变)那些 Producer 还没有发送到 kafka 集群的记录。默认情况下,没有 interceptor 。listnull low
max.in.flight.requests.per.connection在发生阻塞之前,客户端的一个连接上允许出现未确认请求的最大数量。注意,如果这个设置大于1,并且有失败的发送,则消息可能会由于重试而导致重新排序(如果重试是启用的话)。int5[1,...]low
metadata.max.age.ms刷新元数据的时间间隔,单位毫秒。即使没有发现任何分区的 leadership 发生变更也会强制刷新以便能主动发现新的 Broker 或者新的分区。long300000[0,...]low
metric.reporters用于指标监控报表的类清单。实现org.apache.kafka.common.metrics.MetricsReporter接口之后允许插入能够通知新的创建度量的类。JmxReporter 总是包含在注册的 JMX 统计信息中。list"" low
metrics.num.samples计算 metrics 所需要维持的样本数量。int2[1,...]low
metrics.recording.levelmetrics 的最高纪录级别。stringINFO[INFO, DEBUG]low
metrics.sample.window.ms计算 metrics 样本的时间窗口。long30000[0,...]low
reconnect.backoff.max.ms当重新连接到一台多次连接失败的 Broker 时允许等待的最大毫秒数。如果配置该参数,则每台主机的 backoff 将呈指数级增长直到达到配置的最大值。当统计到 backoff 在增长,系统会增加20%的随机波动以避免大量的连接失败。long1000[0,...]low
reconnect.backoff.ms在尝试重新连接到给定的主机之前,需要等待的基本时间。这避免了在一个紧凑的循环中反复连接到同一个主机。这个 backoff 机制应用于所有客户端尝试连接到 Broker 的请求。long50[0,...]low
retry.backoff.ms在尝试将一个失败的请求重试到给定的 topic 分区之前需要等待的时间。这避免在某些失败场景下在紧凑的循环中重复发送请求。long100[0,...]low
sasl.kerberos.kinit.cmdKerberos kinit 命令的路径。string/usr/bin/kinit low
sasl.kerberos.min.time.before.relogin重新尝试登陆之前,登录线程的休眠时间。long60000 low
sasl.kerberos.ticket.renew.jitter随机抖动增加到更新时间的百分比。double0.05 low
sasl.kerberos.ticket.renew.window.factor登录线程将持续休眠直到上一次刷新到 ticket 的过期时间窗口,在此时间窗口它将尝试更新 ticket 。double0.8 low
ssl.cipher.suites密码套件列表。密码套件是利用 TLS 或 SSL 网络协议来实现网络连接的安全设置,是一个涵盖认证,加密,MAC和密钥交换算法的组合。默认情况下,支持所有可用的密码套件。listnull low
ssl.endpoint.identification.algorithm使用服务器证书验证服务器主机名的 endpoint 识别算法。stringnull low
ssl.keymanager.algorithmkey manager factory 用于 SSL 连接的算法。默认值是Java虚拟机配置的 key manager factory 算法。stringSunX509 low
ssl.secure.random.implementation用于 SSL 加密操作的 SecureRandom PRNG 实现。stringnull low
ssl.trustmanager.algorithmtrust manager factory 用于SSL连接的算法。默认值是Java虚拟机配置的 trust manager factory 算法。stringPKIX low
transaction.timeout.ms主动中止进行中的事务之前,事务协调器等待 Producer 更新事务状态的最长时间(以毫秒为单位)。如果此值大于 Broker 中的 max.transaction.timeout.ms 设置的时长,则请求将失败并提示"InvalidTransactionTimeout"错误。int60000 low
transactional.id用于事务交付的 TransactionalId。 这使跨越多个生产者会话的可靠性语义成为可能,因为它可以保证客户在开始任何新的事务之前,使用相同的 TransactionalId 的事务都已经完成。 如果没有提供 TransactionalId ,则 Producer 被限制为幂等递送。 请注意,如果配置了 TransactionalId,则必须启用 enable.idempotence 。 缺省值为空,这意味着无法使用事务。stringnullnon-empty stringlow

如果对老的Scala版本的 Producer 配置感兴趣,请点击 这里.

3.4 Consumer Configs

In 0.9.0.0 we introduced the new Java consumer as a replacement for the older Scala-based simple and high-level consumers. The configs for both new and old consumers are described below.

3.4.1 New Consumer Configs

Below is the configuration for the new consumer:

NAMEDESCRIPTIONTYPEDEFAULTVALID VALUESIMPORTANCE
bootstrap.serversA list of host/port pairs to use for establishing the initial connection to the Kafka cluster. The client will make use of all servers irrespective of which servers are specified here for bootstrapping—this list only impacts the initial hosts used to discover the full set of servers. This list should be in the form host1:port1,host2:port2,.... Since these servers are just used for the initial connection to discover the full cluster membership (which may change dynamically), this list need not contain the full set of servers (you may want more than one, though, in case a server is down).list  high
key.deserializerDeserializer class for key that implements the org.apache.kafka.common.serialization.Deserializer interface.class  high
value.deserializerDeserializer class for value that implements the org.apache.kafka.common.serialization.Deserializer interface.class  high
fetch.min.bytesThe minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request. The default setting of 1 byte means that fetch requests are answered as soon as a single byte of data is available or the fetch request times out waiting for data to arrive. Setting this to something greater than 1 will cause the server to wait for larger amounts of data to accumulate which can improve server throughput a bit at the cost of some additional latency.int1[0,...]high
group.idA unique string that identifies the consumer group this consumer belongs to. This property is required if the consumer uses either the group management functionality by using subscribe(topic) or the Kafka-based offset management strategy.string"" high
heartbeat.interval.msThe expected time between heartbeats to the consumer coordinator when using Kafka's group management facilities. Heartbeats are used to ensure that the consumer's session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances.int3000 high
max.partition.fetch.bytesThe maximum amount of data per-partition the server will return. Records are fetched in batches by the consumer. If the first record batch in the first non-empty partition of the fetch is larger than this limit, the batch will still be returned to ensure that the consumer can make progress. The maximum record batch size accepted by the broker is defined via message.max.bytes (broker config) or max.message.bytes (topic config). See fetch.max.bytes for limiting the consumer request size.int1048576[0,...]high
session.timeout.msThe timeout used to detect consumer failures when using Kafka's group management facility. The consumer sends periodic heartbeats to indicate its liveness to the broker. If no heartbeats are received by the broker before the expiration of this session timeout, then the broker will remove this consumer from the group and initiate a rebalance. Note that the value must be in the allowable range as configured in the broker configuration by group.min.session.timeout.ms and group.max.session.timeout.ms.int10000 high
ssl.key.passwordThe password of the private key in the key store file. This is optional for client.passwordnull high
ssl.keystore.locationThe location of the key store file. This is optional for client and can be used for two-way authentication for client.stringnull high
ssl.keystore.passwordThe store password for the key store file. This is optional for client and only needed if ssl.keystore.location is configured.passwordnull high
ssl.truststore.locationThe location of the trust store file.stringnull high
ssl.truststore.passwordThe password for the trust store file. If a password is not set access to the truststore is still available, but integrity checking is disabled.passwordnull high
auto.offset.resetWhat to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server (e.g. because that data has been deleted):
  • earliest: automatically reset the offset to the earliest offset
  • latest: automatically reset the offset to the latest offset
  • none: throw exception to the consumer if no previous offset is found for the consumer's group
  • anything else: throw exception to the consumer.
stringlatest[latest, earliest, none]medium
connections.max.idle.msClose idle connections after the number of milliseconds specified by this config.long540000 medium
enable.auto.commitIf true the consumer's offset will be periodically committed in the background.booleantrue medium
exclude.internal.topicsWhether records from internal topics (such as offsets) should be exposed to the consumer. If set to true the only way to receive records from an internal topic is subscribing to it.booleantrue medium
fetch.max.bytesThe maximum amount of data the server should return for a fetch request. Records are fetched in batches by the consumer, and if the first record batch in the first non-empty partition of the fetch is larger than this value, the record batch will still be returned to ensure that the consumer can make progress. As such, this is not a absolute maximum. The maximum record batch size accepted by the broker is defined via message.max.bytes (broker config) or max.message.bytes (topic config). Note that the consumer performs multiple fetches in parallel.int52428800[0,...]medium
isolation.level

Controls how to read messages written transactionally. If set to read_committed, consumer.poll() will only return transactional messages which have been committed. If set to read_uncommitted' (the default), consumer.poll() will return all messages, even transactional messages which have been aborted. Non-transactional messages will be returned unconditionally in either mode.

Messages will always be returned in offset order. Hence, in read_committed mode, consumer.poll() will only return messages up to the last stable offset (LSO), which is the one less than the offset of the first open transaction. In particular any messages appearing after messages belonging to ongoing transactions will be withheld until the relevant transaction has been completed. As a result, read_committed consumers will not be able to read up to the high watermark when there are in flight transactions.

Further, when in read_committed the seekToEnd method will return the LSO

stringread_uncommitted[read_committed, read_uncommitted]medium
max.poll.interval.msThe maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before expiration of this timeout, then the consumer is considered failed and the group will rebalance in order to reassign the partitions to another member.int300000[1,...]medium
max.poll.recordsThe maximum number of records returned in a single call to poll().int500[1,...]medium
partition.assignment.strategyThe class name of the partition assignment strategy that the client will use to distribute partition ownership amongst consumer instances when group management is usedlistclass org.apache.kafka.clients.consumer.RangeAssignor medium
receive.buffer.bytesThe size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default will be used.int65536[-1,...]medium
request.timeout.msThe configuration controls the maximum amount of time the client will wait for the response of a request. If the response is not received before the timeout elapses the client will resend the request if necessary or fail the request if retries are exhausted.int305000[0,...]medium
sasl.jaas.configJAAS login context parameters for SASL connections in the format used by JAAS configuration files. JAAS configuration file format is described here. The format for the value is: ' (=)*;'passwordnull medium
sasl.kerberos.service.nameThe Kerberos principal name that Kafka runs as. This can be defined either in Kafka's JAAS config or in Kafka's config.stringnull medium
sasl.mechanismSASL mechanism used for client connections. This may be any mechanism for which a security provider is available. GSSAPI is the default mechanism.stringGSSAPI medium
security.protocolProtocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.stringPLAINTEXT medium
send.buffer.bytesThe size of the TCP send buffer (SO_SNDBUF) to use when sending data. If the value is -1, the OS default will be used.int131072[-1,...]medium
ssl.enabled.protocolsThe list of protocols enabled for SSL connections.listTLSv1.2,TLSv1.1,TLSv1 medium
ssl.keystore.typeThe file format of the key store file. This is optional for client.stringJKS medium
ssl.protocolThe SSL protocol used to generate the SSLContext. Default setting is TLS, which is fine for most cases. Allowed values in recent JVMs are TLS, TLSv1.1 and TLSv1.2. SSL, SSLv2 and SSLv3 may be supported in older JVMs, but their usage is discouraged due to known security vulnerabilities.stringTLS medium
ssl.providerThe name of the security provider used for SSL connections. Default value is the default security provider of the JVM.stringnull medium
ssl.truststore.typeThe file format of the trust store file.stringJKS medium
auto.commit.interval.msThe frequency in milliseconds that the consumer offsets are auto-committed to Kafka if enable.auto.commit is set to true.int5000[0,...]low
check.crcsAutomatically check the CRC32 of the records consumed. This ensures no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so it may be disabled in cases seeking extreme performance.booleantrue low
client.idAn id string to pass to the server when making requests. The purpose of this is to be able to track the source of requests beyond just ip/port by allowing a logical application name to be included in server-side request logging.string"" low
fetch.max.wait.msThe maximum amount of time the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy the requirement given by fetch.min.bytes.int500[0,...]low
interceptor.classesA list of classes to use as interceptors. Implementing the org.apache.kafka.clients.consumer.ConsumerInterceptor interface allows you to intercept (and possibly mutate) records received by the consumer. By default, there are no interceptors.listnull low
metadata.max.age.msThe period of time in milliseconds after which we force a refresh of metadata even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions.long300000[0,...]low
metric.reportersA list of classes to use as metrics reporters. Implementing the org.apache.kafka.common.metrics.MetricsReporter interface allows plugging in classes that will be notified of new metric creation. The JmxReporter is always included to register JMX statistics.list"" low
metrics.num.samplesThe number of samples maintained to compute metrics.int2[1,...]low
metrics.recording.levelThe highest recording level for metrics.stringINFO[INFO, DEBUG]low
metrics.sample.window.msThe window of time a metrics sample is computed over.long30000[0,...]low
reconnect.backoff.max.msThe maximum amount of time in milliseconds to wait when reconnecting to a broker that has repeatedly failed to connect. If provided, the backoff per host will increase exponentially for each consecutive connection failure, up to this maximum. After calculating the backoff increase, 20% random jitter is added to avoid connection storms.long1000[0,...]low
reconnect.backoff.msThe base amount of time to wait before attempting to reconnect to a given host. This avoids repeatedly connecting to a host in a tight loop. This backoff applies to all connection attempts by the client to a broker.long50[0,...]low
retry.backoff.msThe amount of time to wait before attempting to retry a failed request to a given topic partition. This avoids repeatedly sending requests in a tight loop under some failure scenarios.long100[0,...]low
sasl.kerberos.kinit.cmdKerberos kinit command path.string/usr/bin/kinit low
sasl.kerberos.min.time.before.reloginLogin thread sleep time between refresh attempts.long60000 low
sasl.kerberos.ticket.renew.jitterPercentage of random jitter added to the renewal time.double0.05 low
sasl.kerberos.ticket.renew.window.factorLogin thread will sleep until the specified window factor of time from last refresh to ticket's expiry has been reached, at which time it will try to renew the ticket.double0.8 low
ssl.cipher.suitesA list of cipher suites. This is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol. By default all the available cipher suites are supported.listnull low
ssl.endpoint.identification.algorithmThe endpoint identification algorithm to validate server hostname using server certificate.stringnull low
ssl.keymanager.algorithmThe algorithm used by key manager factory for SSL connections. Default value is the key manager factory algorithm configured for the Java Virtual Machine.stringSunX509 low
ssl.secure.random.implementationThe SecureRandom PRNG implementation to use for SSL cryptography operations.stringnull low
ssl.trustmanager.algorithmThe algorithm used by trust manager factory for SSL connections. Default value is the trust manager factory algorithm configured for the Java Virtual Machine.stringPKIX low

3.4.2 Old Consumer Configs

The essential old consumer configurations are the following:

  • group.id
  • zookeeper.connect
PROPERTYDEFAULTDESCRIPTION
group.id A string that uniquely identifies the group of consumer processes to which this consumer belongs. By setting the same group id multiple processes indicate that they are all part of the same consumer group.
zookeeper.connect Specifies the ZooKeeper connection string in the form hostname:port where host and port are the host and port of a ZooKeeper server. To allow connecting through other ZooKeeper nodes when that ZooKeeper machine is down you can also specify multiple hosts in the form hostname1:port1,hostname2:port2,hostname3:port3.

The server may also have a ZooKeeper chroot path as part of its ZooKeeper connection string which puts its data under some path in the global ZooKeeper namespace. If so the consumer should use the same chroot path in its connection string. For example to give a chroot path of /chroot/path you would give the connection string as hostname1:port1,hostname2:port2,hostname3:port3/chroot/path.

consumer.idnull

Generated automatically if not set.

socket.timeout.ms30 * 1000The socket timeout for network requests. The actual timeout set will be max.fetch.wait + socket.timeout.ms.
socket.receive.buffer.bytes64 * 1024The socket receive buffer for network requests
fetch.message.max.bytes1024 * 1024The number of bytes of messages to attempt to fetch for each topic-partition in each fetch request. These bytes will be read into memory for each partition, so this helps control the memory used by the consumer. The fetch request size must be at least as large as the maximum message size the server allows or else it is possible for the producer to send messages larger than the consumer can fetch.
num.consumer.fetchers1The number fetcher threads used to fetch data.
auto.commit.enabletrueIf true, periodically commit to ZooKeeper the offset of messages already fetched by the consumer. This committed offset will be used when the process fails as the position from which the new consumer will begin.
auto.commit.interval.ms60 * 1000The frequency in ms that the consumer offsets are committed to zookeeper.
queued.max.message.chunks2Max number of message chunks buffered for consumption. Each chunk can be up to fetch.message.max.bytes.
rebalance.max.retries4When a new consumer joins a consumer group the set of consumers attempt to "rebalance" the load to assign partitions to each consumer. If the set of consumers changes while this assignment is taking place the rebalance will fail and retry. This setting controls the maximum number of attempts before giving up.
fetch.min.bytes1The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request.
fetch.wait.max.ms100The maximum amount of time the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy fetch.min.bytes
rebalance.backoff.ms2000Backoff time between retries during rebalance. If not set explicitly, the value in zookeeper.sync.time.ms is used.
refresh.leader.backoff.ms200Backoff time to wait before trying to determine the leader of a partition that has just lost its leader.
auto.offset.resetlargest

What to do when there is no initial offset in ZooKeeper or if an offset is out of range:
* smallest : automatically reset the offset to the smallest offset
* largest : automatically reset the offset to the largest offset
* anything else: throw exception to the consumer

consumer.timeout.ms-1Throw a timeout exception to the consumer if no message is available for consumption after the specified interval
exclude.internal.topicstrueWhether messages from internal topics (such as offsets) should be exposed to the consumer.
client.idgroup id valueThe client id is a user-specified string sent in each request to help trace calls. It should logically identify the application making the request.
zookeeper.session.timeout.ms 6000ZooKeeper session timeout. If the consumer fails to heartbeat to ZooKeeper for this period of time it is considered dead and a rebalance will occur.
zookeeper.connection.timeout.ms6000The max time that the client waits while establishing a connection to zookeeper.
zookeeper.sync.time.ms 2000How far a ZK follower can be behind a ZK leader
offsets.storagezookeeperSelect where offsets should be stored (zookeeper or kafka).
offsets.channel.backoff.ms1000The backoff period when reconnecting the offsets channel or retrying failed offset fetch/commit requests.
offsets.channel.socket.timeout.ms10000Socket timeout when reading responses for offset fetch/commit requests. This timeout is also used for ConsumerMetadata requests that are used to query for the offset manager.
offsets.commit.max.retries5Retry the offset commit up to this many times on failure. This retry count only applies to offset commits during shut-down. It does not apply to commits originating from the auto-commit thread. It also does not apply to attempts to query for the offset coordinator before committing offsets. i.e., if a consumer metadata request fails for any reason, it will be retried and that retry does not count toward this limit.
dual.commit.enabledtrueIf you are using "kafka" as offsets.storage, you can dual commit offsets to ZooKeeper (in addition to Kafka). This is required during migration from zookeeper-based offset storage to kafka-based offset storage. With respect to any given consumer group, it is safe to turn this off after all instances within that group have been migrated to the new version that commits offsets to the broker (instead of directly to ZooKeeper).
partition.assignment.strategyrange

Select between the "range" or "roundrobin" strategy for assigning partitions to consumer streams.

The round-robin partition assignor lays out all the available partitions and all the available consumer threads. It then proceeds to do a round-robin assignment from partition to consumer thread. If the subscriptions of all consumer instances are identical, then the partitions will be uniformly distributed. (i.e., the partition ownership counts will be within a delta of exactly one across all consumer threads.) Round-robin assignment is permitted only if: (a) Every topic has the same number of streams within a consumer instance (b) The set of subscribed topics is identical for every consumer instance within the group.

Range partitioning works on a per-topic basis. For each topic, we lay out the available partitions in numeric order and the consumer threads in lexicographic order. We then divide the number of partitions by the total number of consumer streams (threads) to determine the number of partitions to assign to each consumer. If it does not evenly divide, then the first few consumers will have one extra partition.

More details about consumer configuration can be found in the scala class kafka.consumer.ConsumerConfig.

3.5 Kafka Connect Configs

Below is the configuration of the Kafka Connect framework.

NAMEDESCRIPTIONTYPEDEFAULTVALID VALUESIMPORTANCE
config.storage.topicThe name of the Kafka topic where connector configurations are storedstring  high
group.idA unique string that identifies the Connect cluster group this worker belongs to.string  high
key.converterConverter class used to convert between Kafka Connect format and the serialized form that is written to Kafka. This controls the format of the keys in messages written to or read from Kafka, and since this is independent of connectors it allows any connector to work with any serialization format. Examples of common formats include JSON and Avro.class  high
offset.storage.topicThe name of the Kafka topic where connector offsets are storedstring  high
status.storage.topicThe name of the Kafka topic where connector and task status are storedstring  high
value.converterConverter class used to convert between Kafka Connect format and the serialized form that is written to Kafka. This controls the format of the values in messages written to or read from Kafka, and since this is independent of connectors it allows any connector to work with any serialization format. Examples of common formats include JSON and Avro.class  high
internal.key.converterConverter class used to convert between Kafka Connect format and the serialized form that is written to Kafka. This controls the format of the keys in messages written to or read from Kafka, and since this is independent of connectors it allows any connector to work with any serialization format. Examples of common formats include JSON and Avro. This setting controls the format used for internal bookkeeping data used by the framework, such as configs and offsets, so users can typically use any functioning Converter implementation.class  low
internal.value.converterConverter class used to convert between Kafka Connect format and the serialized form that is written to Kafka. This controls the format of the values in messages written to or read from Kafka, and since this is independent of connectors it allows any connector to work with any serialization format. Examples of common formats include JSON and Avro. This setting controls the format used for internal bookkeeping data used by the framework, such as configs and offsets, so users can typically use any functioning Converter implementation.class  low
bootstrap.serversA list of host/port pairs to use for establishing the initial connection to the Kafka cluster. The client will make use of all servers irrespective of which servers are specified here for bootstrapping—this list only impacts the initial hosts used to discover the full set of servers. This list should be in the form host1:port1,host2:port2,.... Since these servers are just used for the initial connection to discover the full cluster membership (which may change dynamically), this list need not contain the full set of servers (you may want more than one, though, in case a server is down).listlocalhost:9092 high
heartbeat.interval.msThe expected time between heartbeats to the group coordinator when using Kafka's group management facilities. Heartbeats are used to ensure that the worker's session stays active and to facilitate rebalancing when new members join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances.int3000 high
rebalance.timeout.msThe maximum allowed time for each worker to join the group once a rebalance has begun. This is basically a limit on the amount of time needed for all tasks to flush any pending data and commit offsets. If the timeout is exceeded, then the worker will be removed from the group, which will cause offset commit failures.int60000 high
session.timeout.msThe timeout used to detect worker failures. The worker sends periodic heartbeats to indicate its liveness to the broker. If no heartbeats are received by the broker before the expiration of this session timeout, then the broker will remove the worker from the group and initiate a rebalance. Note that the value must be in the allowable range as configured in the broker configuration by group.min.session.timeout.ms and group.max.session.timeout.ms.int10000 high
ssl.key.passwordThe password of the private key in the key store file. This is optional for client.passwordnull high
ssl.keystore.locationThe location of the key store file. This is optional for client and can be used for two-way authentication for client.stringnull high
ssl.keystore.passwordThe store password for the key store file. This is optional for client and only needed if ssl.keystore.location is configured.passwordnull high
ssl.truststore.locationThe location of the trust store file.stringnull high
ssl.truststore.passwordThe password for the trust store file. If a password is not set access to the truststore is still available, but integrity checking is disabled.passwordnull high
connections.max.idle.msClose idle connections after the number of milliseconds specified by this config.long540000 medium
receive.buffer.bytesThe size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default will be used.int32768[0,...]medium
request.timeout.msThe configuration controls the maximum amount of time the client will wait for the response of a request. If the response is not received before the timeout elapses the client will resend the request if necessary or fail the request if retries are exhausted.int40000[0,...]medium
sasl.jaas.configJAAS login context parameters for SASL connections in the format used by JAAS configuration files. JAAS configuration file format is described here. The format for the value is: ' (=)*;'passwordnull medium
sasl.kerberos.service.nameThe Kerberos principal name that Kafka runs as. This can be defined either in Kafka's JAAS config or in Kafka's config.stringnull medium
sasl.mechanismSASL mechanism used for client connections. This may be any mechanism for which a security provider is available. GSSAPI is the default mechanism.stringGSSAPI medium
security.protocolProtocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.stringPLAINTEXT medium
send.buffer.bytesThe size of the TCP send buffer (SO_SNDBUF) to use when sending data. If the value is -1, the OS default will be used.int131072[0,...]medium
ssl.enabled.protocolsThe list of protocols enabled for SSL connections.listTLSv1.2,TLSv1.1,TLSv1 medium
ssl.keystore.typeThe file format of the key store file. This is optional for client.stringJKS medium
ssl.protocolThe SSL protocol used to generate the SSLContext. Default setting is TLS, which is fine for most cases. Allowed values in recent JVMs are TLS, TLSv1.1 and TLSv1.2. SSL, SSLv2 and SSLv3 may be supported in older JVMs, but their usage is discouraged due to known security vulnerabilities.stringTLS medium
ssl.providerThe name of the security provider used for SSL connections. Default value is the default security provider of the JVM.stringnull medium
ssl.truststore.typeThe file format of the trust store file.stringJKS medium
worker.sync.timeout.msWhen the worker is out of sync with other workers and needs to resynchronize configurations, wait up to this amount of time before giving up, leaving the group, and waiting a backoff period before rejoining.int3000 medium
worker.unsync.backoff.msWhen the worker is out of sync with other workers and fails to catch up within worker.sync.timeout.ms, leave the Connect cluster for this long before rejoining.int300000 medium
access.control.allow.methodsSets the methods supported for cross origin requests by setting the Access-Control-Allow-Methods header. The default value of the Access-Control-Allow-Methods header allows cross origin requests for GET, POST and HEAD.string"" low
access.control.allow.originValue to set the Access-Control-Allow-Origin header to for REST API requests.To enable cross origin access, set this to the domain of the application that should be permitted to access the API, or '*' to allow access from any domain. The default value only allows access from the domain of the REST API.string"" low
client.idAn id string to pass to the server when making requests. The purpose of this is to be able to track the source of requests beyond just ip/port by allowing a logical application name to be included in server-side request logging.string"" low
config.storage.replication.factorReplication factor used when creating the configuration storage topicshort3[1,...]low
metadata.max.age.msThe period of time in milliseconds after which we force a refresh of metadata even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions.long300000[0,...]low
metric.reportersA list of classes to use as metrics reporters. Implementing the org.apache.kafka.common.metrics.MetricsReporter interface allows plugging in classes that will be notified of new metric creation. The JmxReporter is always included to register JMX statistics.list"" low
metrics.num.samplesThe number of samples maintained to compute metrics.int2[1,...]low
metrics.recording.levelThe highest recording level for metrics.stringINFO[INFO, DEBUG]low
metrics.sample.window.msThe window of time a metrics sample is computed over.long30000[0,...]low
offset.flush.interval.msInterval at which to try committing offsets for tasks.long60000 low
offset.flush.timeout.msMaximum number of milliseconds to wait for records to flush and partition offset data to be committed to offset storage before cancelling the process and restoring the offset data to be committed in a future attempt.long5000 low
offset.storage.partitionsThe number of partitions used when creating the offset storage topicint25[1,...]low
offset.storage.replication.factorReplication factor used when creating the offset storage topicshort3[1,...]low
plugin.pathList of paths separated by commas (,) that contain plugins (connectors, converters, transformations). The list should consist of top level directories that include any combination of: a) directories immediately containing jars with plugins and their dependencies b) uber-jars with plugins and their dependencies c) directories immediately containing the package directory structure of classes of plugins and their dependencies Note: symlinks will be followed to discover dependencies or plugins. Examples: plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectorslistnull low
reconnect.backoff.max.msThe maximum amount of time in milliseconds to wait when reconnecting to a broker that has repeatedly failed to connect. If provided, the backoff per host will increase exponentially for each consecutive connection failure, up to this maximum. After calculating the backoff increase, 20% random jitter is added to avoid connection storms.long1000[0,...]low
reconnect.backoff.msThe base amount of time to wait before attempting to reconnect to a given host. This avoids repeatedly connecting to a host in a tight loop. This backoff applies to all connection attempts by the client to a broker.long50[0,...]low
rest.advertised.host.nameIf this is set, this is the hostname that will be given out to other workers to connect to.stringnull low
rest.advertised.portIf this is set, this is the port that will be given out to other workers to connect to.intnull low
rest.host.nameHostname for the REST API. If this is set, it will only bind to this interface.stringnull low
rest.portPort for the REST API to listen on.int8083 low
retry.backoff.msThe amount of time to wait before attempting to retry a failed request to a given topic partition. This avoids repeatedly sending requests in a tight loop under some failure scenarios.long100[0,...]low
sasl.kerberos.kinit.cmdKerberos kinit command path.string/usr/bin/kinit low
sasl.kerberos.min.time.before.reloginLogin thread sleep time between refresh attempts.long60000 low
sasl.kerberos.ticket.renew.jitterPercentage of random jitter added to the renewal time.double0.05 low
sasl.kerberos.ticket.renew.window.factorLogin thread will sleep until the specified window factor of time from last refresh to ticket's expiry has been reached, at which time it will try to renew the ticket.double0.8 low
ssl.cipher.suitesA list of cipher suites. This is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol. By default all the available cipher suites are supported.listnull low
ssl.endpoint.identification.algorithmThe endpoint identification algorithm to validate server hostname using server certificate.stringnull low
ssl.keymanager.algorithmThe algorithm used by key manager factory for SSL connections. Default value is the key manager factory algorithm configured for the Java Virtual Machine.stringSunX509 low
ssl.secure.random.implementationThe SecureRandom PRNG implementation to use for SSL cryptography operations.stringnull low
ssl.trustmanager.algorithmThe algorithm used by trust manager factory for SSL connections. Default value is the trust manager factory algorithm configured for the Java Virtual Machine.stringPKIX low
status.storage.partitionsThe number of partitions used when creating the status storage topicint5[1,...]low
status.storage.replication.factorReplication factor used when creating the status storage topicshort3[1,...]low
task.shutdown.graceful.timeout.msAmount of time to wait for tasks to shutdown gracefully. This is the total amount of time, not per task. All task have shutdown triggered, then they are waited on sequentially.long5000 low

3.6 Kafka Streams Configs

Below is the configuration of the Kafka Streams client library.

3.7 AdminClient Configs

Below is the configuration of the Kafka Admin client library.

NAMEDESCRIPTIONTYPEDEFAULTVALID VALUESIMPORTANCE
bootstrap.serversA list of host/port pairs to use for establishing the initial connection to the Kafka cluster. The client will make use of all servers irrespective of which servers are specified here for bootstrapping—this list only impacts the initial hosts used to discover the full set of servers. This list should be in the form host1:port1,host2:port2,.... Since these servers are just used for the initial connection to discover the full cluster membership (which may change dynamically), this list need not contain the full set of servers (you may want more than one, though, in case a server is down).list  high
ssl.key.passwordThe password of the private key in the key store file. This is optional for client.passwordnull high
ssl.keystore.locationThe location of the key store file. This is optional for client and can be used for two-way authentication for client.stringnull high
ssl.keystore.passwordThe store password for the key store file. This is optional for client and only needed if ssl.keystore.location is configured.passwordnull high
ssl.truststore.locationThe location of the trust store file.stringnull high
ssl.truststore.passwordThe password for the trust store file. If a password is not set access to the truststore is still available, but integrity checking is disabled.passwordnull high
client.idAn id string to pass to the server when making requests. The purpose of this is to be able to track the source of requests beyond just ip/port by allowing a logical application name to be included in server-side request logging.string"" medium
connections.max.idle.msClose idle connections after the number of milliseconds specified by this config.long300000 medium
receive.buffer.bytesThe size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default will be used.int65536[-1,...]medium
request.timeout.msThe configuration controls the maximum amount of time the client will wait for the response of a request. If the response is not received before the timeout elapses the client will resend the request if necessary or fail the request if retries are exhausted.int120000[0,...]medium
sasl.jaas.configJAAS login context parameters for SASL connections in the format used by JAAS configuration files. JAAS configuration file format is described here. The format for the value is: ' (=)*;'passwordnull medium
sasl.kerberos.service.nameThe Kerberos principal name that Kafka runs as. This can be defined either in Kafka's JAAS config or in Kafka's config.stringnull medium
sasl.mechanismSASL mechanism used for client connections. This may be any mechanism for which a security provider is available. GSSAPI is the default mechanism.stringGSSAPI medium
security.protocolProtocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.stringPLAINTEXT medium
send.buffer.bytesThe size of the TCP send buffer (SO_SNDBUF) to use when sending data. If the value is -1, the OS default will be used.int131072[-1,...]medium
ssl.enabled.protocolsThe list of protocols enabled for SSL connections.listTLSv1.2,TLSv1.1,TLSv1 medium
ssl.keystore.typeThe file format of the key store file. This is optional for client.stringJKS medium
ssl.protocolThe SSL protocol used to generate the SSLContext. Default setting is TLS, which is fine for most cases. Allowed values in recent JVMs are TLS, TLSv1.1 and TLSv1.2. SSL, SSLv2 and SSLv3 may be supported in older JVMs, but their usage is discouraged due to known security vulnerabilities.stringTLS medium
ssl.providerThe name of the security provider used for SSL connections. Default value is the default security provider of the JVM.stringnull medium
ssl.truststore.typeThe file format of the trust store file.stringJKS medium
metadata.max.age.msThe period of time in milliseconds after which we force a refresh of metadata even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions.long300000[0,...]low
metric.reportersA list of classes to use as metrics reporters. Implementing the org.apache.kafka.common.metrics.MetricsReporter interface allows plugging in classes that will be notified of new metric creation. The JmxReporter is always included to register JMX statistics.list"" low
metrics.num.samplesThe number of samples maintained to compute metrics.int2[1,...]low
metrics.recording.levelThe highest recording level for metrics.stringINFO[INFO, DEBUG]low
metrics.sample.window.msThe window of time a metrics sample is computed over.long30000[0,...]low
reconnect.backoff.max.msThe maximum amount of time in milliseconds to wait when reconnecting to a broker that has repeatedly failed to connect. If provided, the backoff per host will increase exponentially for each consecutive connection failure, up to this maximum. After calculating the backoff increase, 20% random jitter is added to avoid connection storms.long1000[0,...]low
reconnect.backoff.msThe base amount of time to wait before attempting to reconnect to a given host. This avoids repeatedly connecting to a host in a tight loop. This backoff applies to all connection attempts by the client to a broker.long50[0,...]low
retriesThe maximum number of times to retry a call before failing it.int5[0,...]low
retry.backoff.msThe amount of time to wait before attempting to retry a failed request. This avoids repeatedly sending requests in a tight loop under some failure scenarios.long100[0,...]low
sasl.kerberos.kinit.cmdKerberos kinit command path.string/usr/bin/kinit low
sasl.kerberos.min.time.before.reloginLogin thread sleep time between refresh attempts.long60000 low
sasl.kerberos.ticket.renew.jitterPercentage of random jitter added to the renewal time.double0.05 low
sasl.kerberos.ticket.renew.window.factorLogin thread will sleep until the specified window factor of time from last refresh to ticket's expiry has been reached, at which time it will try to renew the ticket.double0.8 low
ssl.cipher.suitesA list of cipher suites. This is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol. By default all the available cipher suites are supported.listnull low
ssl.endpoint.identification.algorithmThe endpoint identification algorithm to validate server hostname using server certificate.stringnull low
ssl.keymanager.algorithmThe algorithm used by key manager factory for SSL connections. Default value is the key manager factory algorithm configured for the Java Virtual Machine.stringSunX509 low
ssl.secure.random.implementationThe SecureRandom PRNG implementation to use for SSL cryptography operations.stringnull low
ssl.trustmanager.algorithmThe algorithm used by trust manager factory for SSL connections. Default value is the trust manager factory algorithm configured for the Java Virtual Machine.stringPKIX low

 

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/煮酒与君饮/article/detail/813197
推荐阅读
相关标签
  

闽ICP备14008679号