Kafka 配置属性
使用配置属性配置 Kafka 组件
摘要
使开源包含更多
红帽致力于替换我们的代码、文档和 Web 属性中存在问题的语言。我们从这四个术语开始:master、slave、黑名单和白名单。由于此项工作十分艰巨,这些更改将在即将推出的几个发行版本中逐步实施。有关更多详情,请参阅我们的首席技术官 Chris Wright 提供的消息。
第 1 章 代理配置属性
advertised.listeners
type: string
Default: null
Importance: high
Dynamic update: per-broker
如果与监听程序配置属性不同,则
监听程序
供客户端发布至 ZooKeeper。在 IaaS 环境中,这可能需要与代理绑定的接口不同。如果没有设置,则将使用监听程序
的值。与监听器
不同,它不可用于公告 0.0.0.0 meta-address。与监听器
不同,此属性中可能存在重复的端口,以便可以配置一个监听程序来公告另一个监听程序的地址。这在使用外部负载均衡器时很有用。auto.create.topics.enable
type: boolean
Default: true
Importance: high
Dynamic update: read-only
在服务器上启用自动创建主题。
auto.leader.rebalance.enable
type: boolean
Default: true
Importance: high
Dynamic update: read-only
启用自动领导平衡。后台线程会定期检查分区领导程序的分布,具体由
leader.imbalance.check.interval.seconds
进行配置。如果领导机平衡超过leader.imbalance.per.broker.percentage
,则触发对首选分区的领导重新平衡。background.threads
type: int
Default: 10
Valid Values: [1,…]
Importance: high
Dynamic update: cluster-wide
用于各种后台处理任务的线程数量。
broker.id
type: int
Default: -1
Importance: high
Dynamic update: read-only
此服务器的代理 ID。如果未设置,则会生成唯一的代理 id。为了避免 zookeeper 生成代理 ID 和用户配置的代理 ID 之间的冲突,则生成的代理 ID 从 reserved.broker.max.id + 1.
compression.type
type: string
Default: producer
Importance: high
Dynamic update: cluster-wide
指定给定主题的最后一个压缩类型。此配置接受标准的压缩码c('gzip', 'snappy', 'lz4', 'zstd')。它还接受"未压缩的",它等同于没有压缩 ; 和 'producer' 表示由制作者保留原始压缩代码。
control.plane.listener.name
type: string
Default: null
Importance: high
Dynamic update: read-only
用于控制器和代理间通信的监听程序名称。代理将使用 control.plane.listener.name 在 listeners 列表中查找端点,以侦听来自控制器的连接。例如,如果代理的配置为 : INTERNAL://192.1.1.8:9092, EXTERNAL://10.1.1.5:9093, CONTROLLER://192.1.1.8:9094 listener.security.protocol.map = INTERNAL:PLAINTEXT, EXTERNAL:SSL, CONTROLLER:SSL control.plane.listener.name = CONTROLLER://192.name = INTERNAL:PLAINTEXT = INTERNAL:SSL.909.在控制器一侧,当通过 zookeeper 发现代理发布端点时,它将使用 control.plane.listener.name 来查找端点,它将用来建立与代理的连接。例如,如果代理在 zookeeper 上公布的端点为 : "endpoints" : ["INTERNAL://broker1.example.com:9092","EXTERNAL://broker1.example.com:9093","CONTROLLER://broker1.example.com:9094"],控制器的配置为 : listener.security.protocol.map = : INTTERNAL://broker1.example.com:9093","CONTROLLER://broker1.example. EXTERNAL:SSL, CONTROLLER:SSL control.plane.listener.name = CONTROLLER,然后控制器将使用 "broker1.example.com:9094" with security protocol "SSL" 连接到代理。如果没有明确配置,则默认值为 null,并且没有用于控制器连接的专用端点。
controller.listener.names
type: string
Default: null
Importance: high
Dynamic update: read-only
控制器使用的监听程序名称的逗号分隔列表。这在 KRaft 模式下运行时需要。与控制器仲裁通信时,代理总是使用该列表中的第一个监听程序。注: 基于 ZK 的控制器不应该设置此配置。
controller.quorum.election.backoff.max.ms
type: int
Default: 1000(1第二个)
Importance: high
Dynamic update: read-only
在开始新选举前,以毫秒为单位的最长时间。这在帮助防止网格锁定选举的二进制指数级 backoff 机制中使用。
controller.quorum.election.timeout.ms
type: int
Default: 1000(1第二个)
Importance: high
Dynamic update: read-only
在触发新选举机制前,需要等待的最大时间(以毫秒为单位),而无需从领导机中获取。
controller.quorum.fetch.timeout.ms
type: int
Default: 2000(2 秒)
Importance: high
Dynamic update: read-only
从当前领导人获取成功时没有从当前领导人获取的最大时间,然后成为投票者的选举机制;最大时间从大多数仲裁获得,然后询问您是否有新的领导时期。
controller.quorum.voters
type: list
Default: ""
Valid Values: non-empty list
Importance: high
Dynamic update: read-only
在以逗号分开的 {id
}@{host}:{port}:{port}:{port} 列表中显示 voters 集合的 id
/endpoint 信息映射。例如:1@localhost:9092,2@localhost:9093,3@localhost:9094
.delete.topic.enable
type: boolean
Default: true
Importance: high
Dynamic update: read-only
启用删除主题。如果关闭了此配置,通过管理工具删除主题将无效。
leader.imbalance.check.interval.seconds
type: long
Default: 300
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
控制器触发分区重新平衡检查的频率。
leader.imbalance.per.broker.percentage
type: int
Default: 10
Importance: high
Dynamic update: read-only
每个代理允许的 leader imbalance 比率。如果每个代理超过这个值,控制器将触发领导平衡。该值以百分比的形式指定。
监听程序
Type: string
Default: PLAINTEXT://:9092
Importance: high
Dynamic update: per-broker
侦听器列表 - 我们将侦听的 URI 和侦听器名称分隔的 URI 列表。如果监听器名称不是安全协议,则必须设置
listener.security.protocol.map
。侦听器的名称和端口号必须是唯一的。将 hostname 指定为 0.0.0.0 以绑定到所有接口。将 hostname 留空,以绑定到默认接口。法律监听器列表示例:PLAINTEXT://myhost:9092,SSL://:9091 CLIENT://0.0.0.0:9092,REPLICATION://localhost:9093.log.dir
type: string
Default: /tmp/kafka-logs
Importance: high
Dynamic update: read-only
保存日志数据的目录(log.dirs 属性的upplemental)。
log.dirs
type: string
Default: null
Importance: high
Dynamic update: read-only
保存日志数据的目录。如果没有设置,则使用 log.dir 的值。
log.flush.interval.messages
type: long
Default: 9223372036854775807
Valid Values: [1,…]
Importance: high
Dynamic update: cluster-wide
在消息冲刷到磁盘前,日志分区中累计的消息数量。
log.flush.interval.ms
type: long
Default: null
Importance: high
Dynamic update: cluster-wide
在刷新到磁盘前,任何主题中的消息保存在内存中的最长时间(ms)。如果没有设置,则使用 log.flush.scheduler.interval.ms 值。
log.flush.offset.checkpoint.interval.ms
type: int
Default: 60000(1 minute)
Valid Values: [0,…]
Importance: high
Dynamic update: read-only
更新最后一次冲刷的持久记录的频率,该记录充当日志恢复点。
log.flush.scheduler.interval.ms
type: long
Default: 9223372036854775807
Importance: high
Dynamic update: read-only
日志冲刷器检查是否有日志需要冲刷到磁盘的频率。
log.flush.start.offset.checkpoint.interval.ms
type: int
Default: 60000(1 minute)
Valid Values: [0,…]
Importance: high
Dynamic update: read-only
我们更新日志启动偏移记录的频率。
log.retention.bytes
type: long
Default: -1
Importance: high
Dynamic update: cluster-wide
在删除日志前的最大大小。
log.retention.hours
type: int
Default: 168
Importance: high
Dynamic update: read-only
在删除日志文件前(以小时为单位)保留日志文件的小时数,用于log.retention.ms 属性。
log.retention.minutes
type: int
Default: null
Importance: high
Dynamic update: read-only
在删除日志文件前保留日志文件的数量(以分钟为单位),使用 secondary(在分钟内)记录日志.retention.ms 属性。如果没有设置,则使用 log.retention.hours 的值。
log.retention.ms
type: long
Default: null
Importance: high
Dynamic update: cluster-wide
在删除日志文件前保留日志文件的数量(以毫秒为单位),则使用 log.retention.minutes 的值。如果设置为 -1,则不应用时间限制。
log.roll.hours
type: int
Default: 168
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
推出新日志段前(以小时为单位),log.roll.ms 属性前的最长时间。
log.roll.jitter.hours
type: int
Default: 0
Valid Values: [0,…]
Importance: high
Dynamic update: read-only
从 logRollTimeMillis(小时)中减去最大 jitter (以小时为单位),从 log.roll.jitter.ms 属性中减去。
log.roll.jitter.ms
type: long
Default: null
Importance: high
Dynamic update: cluster-wide
从 logRollTimeMillis(毫秒)中减去的最大值。如果没有设置,则使用 log.roll.jitter.hours 中的值。
log.roll.ms
type: long
Default: null
Importance: high
Dynamic update: cluster-wide
推出新日志片段前的最长时间(以毫秒为单位)。如果没有设置,则使用 log.roll.hours 中的值。
log.segment.bytes
type: int
Default: 1073741824(1 gibibyte)
Valid Values: [14,…]
Importance: high
Dynamic update: cluster-wide
单个日志文件的最大大小。
log.segment.delete.delay.ms
type: long
Default: 60000(1 minute)
Valid Values: [0,…]
Importance: high
Dynamic update: cluster-wide
从文件系统中删除文件前等待的时间。
message.max.bytes
type: int
Default: 1048588
Valid Values: [0,…]
Importance: high
Dynamic update: cluster-wide
Kafka 允许的最大记录批处理大小(如果启用了压缩,则经过压缩)。如果这个增大并且存在大于 0.10.2 的消费者,则必须增加消费者的获取大小,以便他们能够获取此大型的记录批处理。在最新的消息格式版本中,记录始终分组到批处理中以提高效率。在以前的消息格式版本中,未将未压缩记录分组为批处理,此限制仅适用于该案例中的单个记录。这可以针对主题级别
max.message.bytes
配置来设置。metadata.log.dir
type: string
Default: null
Importance: high
Dynamic update: read-only
此配置决定了我们将集群的元数据日志放在 KRaft 模式中。如果没有设置,则元数据日志将放置到 log.dirs 的第一个日志目录中。
metadata.log.max.record.bytes.between.snapshots
type: long
Default: 20971520
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
这是在生成新快照前,日志之间的最大字节数和高水位线。
metadata.log.segment.bytes
type: int
Default: 1073741824(1 gibibyte)
Valid Values: [12,…]
Importance: high
Dynamic update: read-only
单个元数据日志文件的最大大小。
metadata.log.segment.ms
type: long
Default: 604800000(7 天)
Importance: high
Dynamic update: read-only
推出新元数据日志文件前的最长时间(以毫秒为单位)。
metadata.max.retention.bytes
type: long
Default: -1
Importance: high
Dynamic update: read-only
在删除旧快照和日志文件前,元数据日志和快照的最大组合大小。由于在删除任何日志前必须至少存在一个快照,所以这是软限制。
metadata.max.retention.ms
type: long
Default: 604800000(7 天)
Importance: high
Dynamic update: read-only
在删除元数据日志文件或快照前保留元数据日志文件或快照的毫秒数。由于在删除任何日志前必须至少存在一个快照,所以这是软限制。
min.insync.replicas
type: int
Default: 1
Valid Values: [1,…]
Importance: high
Dynamic update: cluster-wide
当制作者将 acks 设为 "all"(或 "-1"),min.insync.replicas 指定必须确认写入成功的最小副本数。如果达到此最小值,则生产者将引发异常(NotEnoughReplicas 或 NotEnoughReplicasAppend)。当同时使用 min.insync.replicas 和 acks 时,您可以强制实现更大的持久保障。典型的场景是创建包含复制原因 3 的主题,将 min.insync.replicas 设置为 2,并使用"all"进行生成。这将确保,在大多数副本没有收到写入时,生产者会引发异常。
node.id
type: int
Default: -1
Importance: high
Dynamic update: read-only
与这个进程的角色关联的节点 ID 在
process.roles
为非空时 play。这在 KRaft 模式中运行时需要进行配置。num.io.threads
type: int
Default: 8
Valid Values: [1,…]
Importance: high
Dynamic update: cluster-wide
服务器用于处理请求的线程数量,其中可能包括磁盘 I/O。
num.network.threads
type: int
Default: 3
Valid Values: [1,…]
Importance: high
Dynamic update: cluster-wide
服务器用于从网络接收请求并向网络发送响应的线程数量。
num.recovery.threads.per.data.dir
type: int
Default: 1
Valid Values: [1,…]
Importance: high
Dynamic update: cluster-wide
每个数据目录的线程数量,用于在启动时进行日志恢复,并在关闭时清除。
num.replica.alter.log.dirs.threads
type: int
Default: null
Importance: high
Dynamic update: read-only
可以在日志目录之间移动副本的线程数量,其中可能包括磁盘 I/O。
num.replica.fetchers
type: int
Default: 1
Importance: high
Dynamic update: cluster-wide
用于从源代理复制消息的提取器线程数。增加这个值可提高后续代理中的 I/O 并行程度。
offset.metadata.max.bytes
type: int
Default: 4096(4 kibibytes)
Importance: high
Dynamic update: read-only
与偏移提交关联的元数据条目的最大大小。
offsets.commit.required.acks
type: short
Default: -1
Importance: high
Dynamic update: read-only
接受提交前所需的 acks。通常,不应覆盖默认(-1)。
offsets.commit.timeout.ms
type: int
Default: 5000(5 秒)
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
偏移提交将延迟到偏移主题的所有副本都接收提交或达到此超时。这与制作者请求超时类似。
offsets.load.buffer.size
type: int
Default: 5242880
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
当将偏移偏移加载到缓存中时,从偏移片段读取的批处理大小(如果记录太大,则被覆盖)。
offsets.retention.check.interval.ms
type: long
Default: 600000 (10 分钟)
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
检查过时的偏移的频率。
offsets.retention.minutes
type: int
Default: 10080
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
消费者组丢弃了其所有消费者(即为空),在放弃之前,其偏移将保留在此保留周期内。对于单机消费者(使用手动分配),在最后一次提交后加上这个保留周期后,偏移将过期。
offsets.topic.compression.codec
type: int
Default: 0
Importance: high
Dynamic update: read-only
用于偏移主题的压缩代码 - 可以利用压缩来实现"atomic"提交。
offsets.topic.num.partitions
type: int
Default: 50
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
偏移提交主题的分区数量(部署后不会更改)。
offsets.topic.replication.factor
type: short
Default: 3
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
偏移主题的复制因素(设置更高的功能以确保可用性)。在集群大小满足此复制因要求前,内部主题创建将失败。
offsets.topic.segment.bytes
type: int
Default: 104857600(100 mebibytes)
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
偏移主题片段字节数应保持相对小,以便加快日志压缩和缓存负载。
process.roles
type: list
Default: ""
Valid Values: [broker, controller]
Importance: high
Dynamic update: read-only
此过程 play 的角色:'broker'、'controller' 或 'broker,controller'(如果两者都是)。此配置仅适用于 KRaft(Kafka Raft)模式(而不是 ZooKeeper)中的集群。对于 Zookeeper 集群保留此配置未定义或为空。
queued.max.requests
type: int
Default: 500
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
在阻塞网络线程前,data-plane 允许排队的请求数量。
replica.fetch.min.bytes
type: int
Default: 1
Importance: high
Dynamic update: read-only
每个获取响应需要的最低字节。如果没有足够字节,请等待最多
replica.fetch.wait.max.ms
(broker config)。replica.fetch.wait.max.ms
type: int
Default: 500
Importance: high
Dynamic update: read-only
遵循方副本发布的每个获取者请求的最大等待时间。这个值应该始终小于 replica.lag.time.max.ms,以防止频繁缩小 ISR 用于低吞吐量主题。
replica.high.watermark.checkpoint.interval.ms
type: long
Default: 5000(5 秒)
Importance: high
Dynamic update: read-only
高水水位线的频率被保存到磁盘。
replica.lag.time.max.ms
type: long
Default: 30000 (30 秒)
Importance: high
Dynamic update: read-only
如果后续程序未发送任何抓取请求,或者尚未消耗到领导日志偏移,则领导机将从 isr 中删除后续程序。
replica.socket.receive.buffer.bytes
type: int
Default: 65536(64 kibibytes)
Importance: high
Dynamic update: read-only
该套接字接收网络请求的缓冲。
replica.socket.timeout.ms
type: int
Default: 30000 (30 秒)
Importance: high
Dynamic update: read-only
网络请求的套接字超时。其值应至少为 replica.fetch.wait.max.ms。
request.timeout.ms
type: int
Default: 30000 (30 秒)
Importance: high
Dynamic update: read-only
配置控制客户端等待请求响应的最大时间。如果在超时前没有收到响应,客户端会在需要时重新发送请求,或者在重试耗尽时失败请求。
sasl.mechanism.controller.protocol
type: string
Default: GSSAPI
Importance: high
Dynamic update: read-only
用于与控制器通信的 SASL 机制。默认为 GSSAPI。
socket.receive.buffer.bytes
type: int
Default: 102400(100 kibibytes)
Importance: high
Dynamic update: read-only
套接字的 SO_RCVBUF 缓冲。如果值为 -1,则使用 OS 默认。
socket.request.max.bytes
type: int
Default: 104857600(100 mebibytes)
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
套接字请求中的最大字节数。
socket.send.buffer.bytes
type: int
Default: 102400(100 kibibytes)
Importance: high
Dynamic update: read-only
套接字的 SO_SNDBUF 缓冲。如果值为 -1,则使用 OS 默认。
transaction.max.timeout.ms
type: int
Default: 900000 (15 分钟)
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
事务允许的最大超时时间。如果客户端请求的事务时间超过这个值,则代理会在 InitProducerIdRequest 返回一个错误。这可防止客户端太大的超时,这可能会阻止用户读取事务中包含的主题。
transaction.state.log.load.buffer.size
type: int
Default: 5242880
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
当将制作者 ID 和事务加载到缓存中时,从事务日志片段读取的批量大小(如果记录太大,则为soft-limit 覆盖)。
transaction.state.log.min.isr
type: int
Default: 2
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
覆盖 min.insync.replicas 配置,用于事务主题。
transaction.state.log.num.partitions
type: int
Default: 50
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
事务主题的分区数量(部署后不会更改)。
transaction.state.log.replication.factor
type: short
Default: 3
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
事务主题的复制因素(设置更高的功能以确保可用性)。在集群大小满足此复制因要求前,内部主题创建将失败。
transaction.state.log.segment.bytes
type: int
Default: 104857600(100 mebibytes)
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
事务主题网段字节数相对应保持小,以便加快日志压缩和缓存负载。
transactional.id.expiration.ms
type: int
Default: 604800000(7 days)
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
攻击协调器将在不收到当前事务状态更新前等待其事务状态更新的时间(以防御其事务 ID)。此设置还影响制作者 ID 到期 - 发生者 ID 后在上次写入者 ID 后过期。请注意,如果因为主题的保留设置而删除制作者 ID 的最后一个写入,则制作者 ID 可能会更早过期。
unclean.leader.election.enable
type: boolean
Default: false
Importance: high
Dynamic update: cluster-wide
指明是否将 ISR 集中的副本作为最后的手段被选为领导人,即使这样做可能会导致数据丢失。
zookeeper.connect
type: string
Default: null
Importance: high
Dynamic update: read-only
指定 ZooKeeper 连接字符串,格式为
hostname:port
,其中 host 和 port 是 ZooKeeper 服务器的主机和端口。当 ZooKeeper 机器停机时,要允许通过其他 ZooKeeper 节点进行连接,也可以使用主机名1:port1,hostname2:port2,hostname3:port3
来指定多个主机。服务器也可以具有 ZooKeeper chroot 路径作为其 ZooKeeper 连接字符串的一部分,将其数据置于全局 ZooKeeper 命名空间中的一些路径下。例如:为/chroot/path
提供 chroot 路径,您可以为连接字符串指定hostname1:port1,hostname2:port2,hostname3:port3/chroot/path
。zookeeper.connection.timeout.ms
type: int
Default: null
Importance: high
Dynamic update: read-only
客户端等待建立与 zookeeper 的连接的最大时间。如果没有设置,则使用 zookeeper.session.timeout.ms 值。
zookeeper.max.in.flight.requests
type: int
Default: 10
Valid Values: [1,…]
Importance: high
Dynamic update: read-only
在阻塞前,客户端将发送到 Zookeeper 的最大未确认请求数量。
zookeeper.session.timeout.ms
type: int
Default: 18000(18 秒)
Importance: high
Dynamic update: read-only
zookeeper 会话超时。
zookeeper.set.acl
type: boolean
Default: false
Importance: high
Dynamic update: read-only
将客户端设置为使用安全 ACL。
broker.heartbeat.interval.ms
type: int
Default: 2000(2 秒)
Importance: medium
Dynamic update: read-only
代理心跳之间毫秒的时间长度。在 KRaft 模式中运行时使用。
broker.id.generation.enable
type: boolean
Default: true
Importance: medium
Dynamic update: read-only
在服务器上启用自动代理 ID 生成。启用为 reserved.broker.max.id 配置的值时,应检查。
broker.rack
type: string
Default: null
Importance: medium
Dynamic update: read-only
代理的机架。这将用于 rack 感知复制,以进行容错。示例:
RACK1、
us-east-1d
.broker.session.timeout.ms
type: int
Default: 9000(9 秒)
Importance: medium
Dynamic update: read-only
在没有心跳时,代理租期最后一次的时间长度(毫秒)。在 KRaft 模式中运行时使用。
connections.max.idle.ms
type: long
Default: 600000 (10 分钟)
Importance: medium
Dynamic update: read-only
闲置连接超时:服务器套接字处理器线程关闭闲置超过此连接的连接。
connections.max.reauth.ms
type: long
Default: 0
Importance: medium
Dynamic update: read-only
当明确设置为正数(默认为 0,不是正数)时,没有超过配置的值的会话生命周期将在验证时与 v2.2.0 或更高版本的客户端通信。代理将断开在会话生命周期内没有重新验证的此类连接,然后用于除 re-authentication 以外的任何目的。配置名称可以选择在小写中带有监听程序前缀和 SASL 机制名称前缀。例如,listener.name.sasl_ssl.oauthbearer.connections.max.reauth.ms=3600000。
controlled.shutdown.enable
type: boolean
Default: true
Importance: medium
Dynamic update: read-only
启用服务器的受控关闭。
controlled.shutdown.max.retries
type: int
Default: 3
Importance: medium
Dynamic update: read-only
由于多个原因,控制关闭可能会失败。这决定了在发生此类失败时的重试次数。
controlled.shutdown.retry.backoff.ms
type: long
Default: 5000(5 秒)
Importance: medium
Dynamic update: read-only
在进行每次重试前,系统需要从导致之前故障的状态中恢复系统(Controller 故障切换、副本延迟等)。此配置决定了重试前要等待的时间。
controller.quorum.append.linger.ms
type: int
Default: 25
Importance: medium
Dynamic update: read-only
领导时间(以毫秒为单位)等待写入累计时间,然后再将它们刷新到磁盘。
controller.quorum.request.timeout.ms
type: int
Default: 2000(2 秒)
Importance: medium
Dynamic update: read-only
配置控制客户端等待请求响应的最大时间。如果在超时前没有收到响应,客户端会在需要时重新发送请求,或者在重试耗尽时失败请求。
controller.socket.timeout.ms
type: int
Default: 30000 (30 秒)
Importance: medium
Dynamic update: read-only
controller-to-broker 频道的套接字超时。
default.replication.factor
type: int
Default: 1
Importance: medium
Dynamic update: read-only
自动创建主题的默认复制因素。
delegation.token.expiry.time.ms
type: long
Default: 86400000 (1 天)
Valid Values: [1,…]
Importance: medium
Dynamic update: read-only
在令牌需要续订前,以毫秒为单位的令牌有效时间。默认值为 1 天。
delegation.token.master.key
type: password
Default: null
Importance: medium
Dynamic update: read-only
DEPRECATED:用于 delegation.token.secret.key 的别名,应改为使用它而不是此配置。
delegation.token.max.lifetime.ms
type: long
Default: 604800000(7 days)
Valid Values: [1,…]
Importance: medium
Dynamic update: read-only
令牌最长生命周期超过其无法续订的生命期。默认值为 7 天。
delegation.token.secret.key
type: password
Default: null
Importance: medium
Dynamic update: read-only
生成和验证委托令牌的机密密钥。必须在所有代理中配置相同的密钥。如果密钥未设置或设为空字符串,代理将禁用委派令牌支持。
delete.records.purgatory.purge.interval.requests
type: int
Default: 1
Importance: medium
Dynamic update: read-only
删除记录请求的清除间隔(以请求数为单位)。
fetch.max.bytes
type: int
Default: 57671680(55 mebibytes)
Valid Values: [1024,…]
Importance: medium
Dynamic update: read-only
我们返回获取请求的字节数。必须至少为 1024。
fetch.purgatory.purge.interval.requests
type: int
Default: 1000
Importance: medium
Dynamic update: read-only
获取请求清除的时间间隔(以请求数为单位)。
group.initial.rebalance.delay.ms
type: int
Default: 3000(3 秒)
Importance: medium
Dynamic update: read-only
组协调器将在执行第一次重新平衡前等待更多消费者加入新组的时间。较长的延迟意味着潜在的重新平衡会较小,但会增加时间,直到处理开始为止。
group.max.session.timeout.ms
type: int
Default: 1800000 (30 分钟)
Importance: medium
Dynamic update: read-only
注册消费者允许的最大会话超时。超时时间较长,使消费者有更多时间在心跳间处理信息,以检测失败。
group.max.size
type: int
Default: 2147483647
Valid Values: [1,…]
Importance: medium
Dynamic update: read-only
单个消费者组可容纳的最大消费者数量。
group.min.session.timeout.ms
type: int
Default: 6000(6 秒)
Importance: medium
Dynamic update: read-only
注册消费者的最小允许会话超时。较短的超时时间会导致以更频繁的消费者心态来加快故障检测,这可能会给代理资源造成负担。
initial.broker.registration.timeout.ms
type: int
Default: 60000 (1 分钟)
Importance: medium
Dynamic update: read-only
最初注册控制器仲裁时,在声明失败并退出代理进程前要等待的毫秒数。
inter.broker.listener.name
type: string
Default: null
Importance: medium
Dynamic update: read-only
用于代理间通信的监听程序名称。如果未设置,则侦听器的名称由 security.inter.broker.protocol 定义。它是设置此和 security.inter.broker.protocol 属性的错误。
inter.broker.protocol.version
type: string
Default: 3.2-IV0
Valid Values: [0.8.0, 0.8.1, 0.8.2, 0.9.0.0 0.10.0-IV0, 0.10.0-IV1, 0.10.1-IV1, 0.10.1-IV1, 0.10.1-IV2, 0.10.2-IV0, 0.10.0-IV0 0.11.0-iv0, 0.11.0-iv1, 0.11.0-iv2, 1.0-iv0, 1.1-iv0, 2.0-iv0, 2.1-iv1, 2.1-iv1, 2.1-iv2, 2.2-iv0, 2.2-iv1, 2.3-iv0, 2.3-iv0, 2.3-iv1, 2.3-iv1, 2.4-IV0, 2.4-IV1, 2.5-IV0, 2.6-IV0, 2.7-IV00, 2.7-IV1, 2.7-IV2, 2.8-IV0, 2.8-IV1, 3.0-IV1, 3.0-IV1, 3.1-IV0, 3.2-IV00, 2.7-IV1, 2.7-IV2, 2.8-IV0, 2.8-IV1
指定将使用的 inter-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 Check ApiVersion for the full list。
log.cleaner.backoff.ms
type: long
Default: 15000(15 秒)
Valid Values: [0,…]
Importance: medium
Dynamic update: cluster-wide
没有日志清理时睡眠的时间量。
log.cleaner.dedupe.buffer.size
type: long
Default: 134217728
Importance: medium
Dynamic update: cluster-wide
用于所有清理线程的日志 deduplication 的总内存。
log.cleaner.delete.retention.ms
type: long
Default: 86400000(1 天)
Importance: medium
Dynamic update: cluster-wide
为日志压缩主题保留 delete tombstone 标记的时间长度。此设置还可在消费者开始从偏移 0 开始读取的时间绑定,以确保它们获得最终阶段的有效快照(在完成扫描之前,可以收集到其他的删除操作)。
log.cleaner.enable
type: boolean
Default: true
Importance: medium
Dynamic update: read-only
启用日志清理流程,以在服务器上运行。如果使用任何包含 cleanup.policy=compact 的主题,则应启用,包括内部偏移主题。如果禁用了这些主题,则这些主题将不会被紧凑,并持续增加大小。
log.cleaner.io.buffer.load.factor
type: double
Default: 0.9
Importance: medium
Dynamic update: cluster-wide
log cleaner dedupe 缓冲加载因素。重复数据删除缓冲区可以成为其百分比。更高的值将允许一次清理更多日志,但会导致更多散列冲突。
log.cleaner.io.buffer.size
type: int
Default: 524288
Valid Values: [0,…]
Importance: medium
Dynamic update: cluster-wide
用于在所有清理线程间日志清理 I/O 缓冲区的总内存。
log.cleaner.io.max.bytes.per.second
Type: double
Default: 1.7976931348623157E308
Importance: medium
Dynamic update: cluster-wide
日志清理将会被节流,以便其读写 i/o 的读取和写入 i/o 的总和平均小于这个值。
log.cleaner.max.compaction.lag.ms
type: long
Default: 9223372036854775807
Importance: medium
Dynamic update: cluster-wide
在日志中,消息会保持在日志中的压缩状态的最长时间。只适用于正在压缩的日志。
log.cleaner.min.cleanable.ratio
type: double
Default: 0.5
Importance: medium
Dynamic update: cluster-wide
脏日志的最小比率是日志的总日志,以符合清理条件。如果还指定了 log.cleaner.max.compaction.lag.ms 或 log.cleaner.min.compaction.lag.ms 配置,则日志压缩器会认为日志符合紧凑值,即已达到脏的比率阈值,日志至少具有脏(不acted)记录。 或者(ii)如果日志在 log.cleaner.max.compaction.lag.ms 周期内有脏的脏(不紧凑)记录。
log.cleaner.min.compaction.lag.ms
type: long
Default: 0
Importance: medium
Dynamic update: cluster-wide
日志消息在日志中保持不限定的最小时间。只适用于正在压缩的日志。
log.cleaner.threads
type: int
Default: 1
Valid Values: [0,…]
Importance: medium
Dynamic update: cluster-wide
用于日志清理的后台线程数量。
log.cleanup.policy
type: list
Default: delete
Valid Values: [compact, delete]
Importance: medium
Dynamic update: cluster-wide
除保留窗口外的片段的默认清理策略。以逗号分隔的有效策略列表。有效策略为:"删除"和"compact"。
log.index.interval.bytes
type: int
Default: 4096(4 kibibytes)
Valid Values: [0,…]
Importance: medium
Dynamic update: cluster-wide
我们向偏移索引中添加条目的时间间隔。
log.index.size.max.bytes
type: int
Default: 10485760(10 mebibytes)
Valid Values: [4,…]
Importance: medium
Dynamic update: cluster-wide
偏移索引的最大大小,以字节为单位。
log.message.format.version
type: string
Default: 3.0-IV1 Valid Values: [0.8.0, 0.8.1, 0.8.2, 0.9.0, 0.10.0-IV0, 0.10.0-IV1, 0.10.1-IV1, 0.10.1-IV1, 0.10.1-IV2, 0.10.2-IV0, 0.10.0-IV0
0.11.0-iv0, 0.11.0-iv1, 0.11.0-iv2, 1.0-iv0, 1.1-iv0, 2.0-iv0, 2.1-iv1, 2.1-iv1, 2.1-iv2, 2.2-iv0, 2.2-iv1, 2.3-iv0, 2.3-iv0, 2.3-iv1, 2.3-iv1, 2.4-IV0, 2.4-IV1, 2.5-IV0, 2.6-IV0, 2.7-IV00, 2.7-IV1, 2.7-IV2, 2.8-IV0, 2.8-IV1, 3.0-IV1, 3.0-IV1, 3.1-IV0, 3.2-IV00, 2.7-IV1, 2.7-IV2, 2.8-IV0, 2.8-IV1
指定代理将用于将信息附加到日志的消息格式版本。该值应该是有效的 ApiVersion。一些示例为: 0.8.2、0.9.0.0、0.0.0,检查 ApiVersion 获取更多详情。通过设置特定的消息格式版本,用户需要验证磁盘上的所有现有信息都小于或等于指定版本。设定这个值会错误地导致具有旧版本的用户出现问题,因为它们会收到没有理解的格式的消息。
log.message.timestamp.difference.max.ms
type: long
Default: 9223372036854775807
Importance: medium
Dynamic update: cluster-wide
代理收到消息和消息中指定的时间戳之间允许的最大差别。如果 log.message.timestamp.type=CreateTime,如果时间戳超过这个阈值,则会拒绝一条消息。如果 log.message.timestamp.type=LogAppendTime,则这个配置会被忽略。如果 log.retention.ms 应该不大于 log.retention.ms,以避免不必要的日志滚动。
log.message.timestamp.type
type: string
Default: CreateTime
Valid Values: [CreateTime, LogAppendTime]
Importance: medium
Dynamic update: cluster-wide
定义消息中的时间戳是消息创建时间或日志附加时间。该值应该是
CreateTime
或LogAppendTime
。log.preallocate
type: boolean
Default: false
Importance: medium
Dynamic update: cluster-wide
创建新片段时是否应预先分配文件?如果您在 Windows 上使用 Kafka,您可能需要将其设置为 true。
log.retention.check.interval.ms
type: long
Default: 300000 (5 分钟)
Valid Values: [1,…]
Importance: medium
Dynamic update: read-only
日志清理器检查日志是否有资格删除的频率,以毫秒为单位。
max.connection.creation.rate
type: int
Default: 2147483647
Valid Values: [0,…]
Importance: medium
Dynamic update: cluster-wide
我们在任何时间允许的最大连接创建率。也可以通过为配置名称添加前缀为
listener.name.internal.max.connection.creation.rate
.Broker 范围连接速率限制来配置监听程序级别的限制,而应根据应用程序要求配置监听程序限制。达到监听器或代理限制时,新的连接将会被节流,但有 inter-broker 侦听器。只有在达到监听器级别速率限制时,才会对 inter-broker 侦听程序上的连接进行节流。max.connections
type: int
Default: 2147483647
Valid Values: [0,…]
Importance: medium
Dynamic update: cluster-wide
我们在任何时间点上允许的最大连接数。除了使用 max.connections.per.ip 配置的任何 per-ip 限制外,还会应用这个限制。也可以通过为配置名称添加前缀为
listener.name.internal.max.connections
来配置监听程序级别的限制。代理范围内的限制应该根据代理容量配置,而监听器限制应根据应用程序要求进行配置。如果启用了监听器或代理限制,则阻止新连接。即使达到代理范围限制,也会允许 inter-broker 侦听程序的连接。在这种情况下,最近在另一个监听器上使用的连接将关闭。max.connections.per.ip
type: int
Default: 2147483647
Valid Values: [0,…]
Importance: medium
Dynamic update: cluster-wide
来自每个 ip 地址允许的最大连接数。如果使用 max.connections.per.ip.overrides 属性配置覆盖,则可以将其设置为 0。如果达到限制,则会丢弃来自 ip 地址的新连接。
max.connections.per.ip.overrides
type: string
Default: ""
Importance: medium
Dynamic update: cluster-wide
Per-ip 或 hostname 的逗号分隔列表会覆盖到默认最大连接数。例如,值为 "hostName:100,127.0.0.1:200"。
max.incremental.fetch.session.cache.slots
type: int
Default: 1000
Valid Values: [0,…]
Importance: medium
Dynamic update: read-only
我们维护的增量获取会话的最大数量。
num.partitions
type: int
Default: 1
Valid Values: [1,…]
Importance: medium
Dynamic update: read-only
每个主题的默认日志分区数量。
password.encoder.old.secret
type: password
Default: null
Importance: medium
Dynamic update: read-only
用于动态配置的编码密码的旧 secret。只有在更新 secret 时才需要。如果指定,所有动态编码的密码都会使用这个旧的 secret 进行解码,并在代理启动时使用 password.encoder.secret 重新编码。
password.encoder.secret
type: password
Default: null
Importance: medium
Dynamic update: read-only
用于为这个代理动态配置的编码密码的 secret。
principal.builder.class
type: class
Default: org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder
Importance: medium
Dynamic update: per-broker
实施 KafkaPrincipalBuilder 接口的类完全限定名称,用于构建在授权期间使用的 KafkaPrincipal 对象。如果没有定义主体构建程序,则默认行为取决于使用中的安全协议。对于 SSL 身份验证,该主体将使用
ssl.principal.mapping.rules
在可分辨名称中(如果不提供客户端验证)中的规则进行派生;否则,如果不需要客户端身份验证,则主体名称为 ANONYMOUS。对于 SASL 身份验证,如果 GSSAPI 正在使用sasl.kerberos.principal.local.rules
定义的规则,以及 SASL 验证 ID 用于其他机制的 SASL 验证 ID。对于 PLAINTEXT,主体将是 ANONYMOUS。producer.purgatory.purge.interval.requests
type: int
Default: 1000
Importance: medium
Dynamic update: read-only
制作者请求的清除间隔(以请求数量表示)。
queued.max.request.bytes
type: long
Default: -1
Importance: medium
Dynamic update: read-only
在读取更多请求前允许的已排队字节数。
replica.fetch.backoff.ms
type: int
Default: 1000(1秒)
Valid Values: [0,…]
Importance: medium
Dynamic update: read-only
发生获取分区错误时休眠的时间。
replica.fetch.max.bytes
type: int
Default: 1048576(1 mebibyte)
Valid Values: [0,…]
Importance: medium
Dynamic update: read-only
尝试获取每个分区的消息的字节数。如果获取的第一个非空分区中第一个非空分区中的第一个记录大于这个值,则记录批处理将不是绝对最大值,则仍会返回记录批处理,以确保可以进行进度。代理接受的最大记录批处理大小通过
message.max.bytes
(broker config)或max.message.bytes
(topic 配置)定义。replica.fetch.response.max.bytes
type: int
Default: 10485760(10 mebibytes)
Valid Values: [0,…]
Importance: medium
Dynamic update: read-only
整个获取响应预期的最大字节数。记录的批处理将计入批处理中,如果提取的第一个非空白分区中的第一个记录大于这个值,则记录批处理仍会返回,以确保进行进度。因此,这不是绝对最大值。代理接受的最大记录批处理大小通过
message.max.bytes
(broker config)或max.message.bytes
(topic 配置)定义。replica.selector.class
type: string
Default: null
Importance: medium
Dynamic update: read-only
实现 ReplicaSelector 的完全限定类名称。这供代理用来查找首选的读取副本。默认情况下,我们使用返回领导的实现。
reserved.broker.max.id
type: int
Default: 1000
Valid Values: [0,…]
Importance: medium
Dynamic update: read-only
可用于 broker.id 的最大数量。
sasl.client.callback.handler.class
type: class
Default: null
Importance: medium
Dynamic update: read-only
实现了 AuthenticateCallbackHandler 接口的 SASL 客户端回调处理器类的完全限定名称。
sasl.enabled.mechanisms
type: list
Default: GSSAPI
Importance: medium
Dynamic update: per-broker
在 Kafka 服务器中启用的 SASL 机制列表。该列表可以包含任何可用的安全提供程序的机制。默认启用 GSSAPI。
sasl.jaas.config
type: password
Default: null
Importance: medium
Dynamic update: per-broker
JAAS 登录上下文参数,用于 SASL 连接,其格式由 JAAS 配置文件使用。此处 介绍了 JAAS 配置文件格式。该值格式为:
loginModuleClass controlFlag(optionName=optionValue)*;
.对于代理,配置必须在小写中带有监听前缀和 SASL 机制名称前缀。例如: listener.name.sasl_ssl.scram-sha-256.sasl.jaas.config=com.example.ScramLoginModule;.sasl.kerberos.kinit.cmd
type: string
Default: /usr/bin/kinit
Importance: medium
Dynamic update: per-broker
Kerberos kinit 命令路径.
sasl.kerberos.min.time.before.relogin
type: long
Default: 60000
Importance: medium
Dynamic update: per-broker
刷新尝试之间的登录线程睡眠时间。
sasl.kerberos.principal.to.local.rules
type: list
Default: DEFAULT
Importance: medium
Dynamic update: per-broker
用于从主体名称映射到短名称(通常是操作系统用户名)的规则列表。规则按照顺序评估,第一个与主体名称匹配的规则被用来将其映射到一个短名称。之后的所有规则将被忽略。默认情况下,格式为 {username}/{hostname}@{REALM} 的主体名称被映射到 {username}。有关格式的更多详情,请参阅安全 授权和 acls。请注意,如果
principal.builder.class
配置提供了 KafkaPrincipalBuilder 的扩展,则此配置会被忽略。sasl.kerberos.service.name
type: string
Default: null
Importance: medium
Dynamic update: per-broker
Kafka 运行的 Kerberos 主体名称。这可以在 Kafka 的 JAAS 配置或 Kafka 配置中定义。
sasl.kerberos.ticket.renew.jitter
type: double
Default: 0.05
Importance: medium
Dynamic update: per-broker
添加至续订时间的随机jitter 的百分比。
sasl.kerberos.ticket.renew.window.factor
type: double
Default: 0.8
Importance: medium
Dynamic update: per-broker
登录线程将休眠,直到达到从最后刷新到 ticket 的过期时间,此时它将尝试续订票据。
sasl.login.callback.handler.class
type: class
Default: null
Importance: medium
Dynamic update: read-only
实现了 AuthenticateCallbackHandler 接口的 SASL 登录回调处理器类的完全限定名称。对于代理,登录回调处理器配置必须以监听器前缀和 SASL 机制名称作为前缀作为前缀。例如: listener.name.sasl_ssl.scram-sha-256.sasl.login.callback.handler.class=com.example.CustomScramLoginCallbackHandler。
sasl.login.class
type: class
Default: null
Importance: medium
Dynamic update: read-only
实施登录接口的类的完全限定名称。对于代理,登录配置必须在小写中带有监听前缀和 SASL 机制名称前缀。For example, listener.name.sasl_ssl.scram-sha-256.sasl.login.class=com.example.CustomScramLogin.
sasl.login.refresh.buffer.seconds
type: short
Default: 300
Importance: medium
Dynamic update: per-broker
在刷新凭证时要维护的凭证前的缓冲时间(以秒为单位)。如果刷新将超过缓冲区秒数的过期时间,则刷新将移动至可维护尽可能多的缓冲区时间。如果未指定值,则使用法定值在 0 到 3600(1 小时)之间;如果没有指定值,则使用默认值 300(5 分钟)。如果值和 sasl.login.refresh.min.period.seconds,如果它们的总和超过凭证的剩余生命周期,则都会被忽略。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.min.period.seconds
type: short
Default: 60
Importance: medium
Dynamic update: per-broker
在刷新凭证前等待登录刷新线程所需的最小时间(以秒为单位)。合法值介于 0 到 900(15 分钟)之间;如果没有指定值,则使用默认值 60(1 分钟)。如果值和 sasl.login.refresh.buffer.seconds 都会被忽略,如果它们的总和超过凭证的剩余生命周期。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.window.factor
type: double
Default: 0.8
Importance: medium
Dynamic update: per-broker
登录刷新线程将休眠,直到达到指定的窗口因素为止(在达到凭证时),它将尝试刷新凭证。法律值介于 0.5(50%)到 1.0(100%)之间,如果未指定值,则使用默认值 0.8(80%)。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.window.jitter
type: double
Default: 0.05
Importance: medium
Dynamic update: per-broker
相对于凭证的生命周期的最大随机 jitter 量,它们被添加到登录刷新线程睡眠时间。法律值介于 0 到 0.25(25%)之间(25%),如果没有指定值,则使用默认值 0.05(5%)。目前仅适用于 OAUTHBEARER。
sasl.mechanism.inter.broker.protocol
type: string
Default: GSSAPI
Importance: medium
Dynamic update: per-broker
用于内部代理通信的 SASL 机制。默认为 GSSAPI。
sasl.oauthbearer.jwks.endpoint.url
type: string
Default: null
Importance: medium
Dynamic update: read-only
可以检索来自提供程序的 JWKS(JSON Web Key Set) 的 OAuth/OIDC 供应商 URL。URL 可以是基于 HTTP(S)或基于文件的。如果 URL 基于 HTTP(S),则 JWKS 数据将通过代理启动时配置的 URL 从 OAuth/OIDC 供应商中检索。所有 then-current 密钥将缓存在代理中,用于传入请求。如果为 JWT 收到身份验证请求,如果包含尚未在缓存中但未包括在缓存中的"kid"标头声明值,则将根据需要再次查询 JWKS 端点。但是,代理会在收到任何 JWT 请求前轮询每个 sasl.oauthbearer.jwks.endpoint.refresh.ms 毫秒的 URL 来刷新缓存。如果 URL 是基于文件,代理将从启动时配置的位置载入 JWKS 文件。如果 JWT 包含不在 JWKS 文件中的"kid"标头值,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.token.endpoint.url
type: string
Default: null
Importance: medium
Dynamic update: read-only
OAuth/OIDC 身份提供程序的 URL。如果 URL 基于 HTTP(S),则代表签发者的令牌端点 URL,请求将基于 sasl.jaas.config 中的配置进行登录。如果 URL 基于文件,它将指定包含由 OAuth/OIDC 身份提供程序发布的访问令牌(在 JWT 序列化表单中)的文件用于授权。
sasl.server.callback.handler.class
type: class
Default: null
Importance: medium
Dynamic update: read-only
实现了 AuthenticateCallbackHandler 接口的 SASL 服务器回调处理器类的完全限定名称。服务器回调处理程序必须以监听器前缀和 SASL 机制名称作为前缀(小写)。例如: listener.name.sasl_ssl.plain.sasl.server.callback.handler.class=com.example.CustomPlainCallbackHandler。
security.inter.broker.protocol
type: string
Default: PLAINTEXT
Importance: medium
Dynamic update: read-only
用于在代理之间通信的安全协议。有效值为:PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL.这是一个同时设置此和 inter.broker.listener.name 属性的错误。
socket.connection.setup.timeout.max.ms
type: long
Default: 30000 (30 秒)
Importance: medium
Dynamic update: read-only
客户端等待套接字连接建立的最大时间。对于每个连续的连接失败,连接设置超时将按这个最大值增加。为避免连接 storms,则 0.2 的随机化工厂将应用到超时,从而导致计算值上面的 20% 到 20% 的随机范围。
socket.connection.setup.timeout.ms
type: long
Default: 10000 (10 秒)
Importance: medium
Dynamic update: read-only
客户端等待套接字连接建立的时间。如果在超时前没有构建连接,客户端将关闭套接字频道。
socket.listen.backlog.size
type: int
Default: 50
Valid Values: [1,…]
Importance: medium
Dynamic update: read-only
套接字中待处理连接的最大数量。在 Linux 中,您可能还需要配置
somaxconn
和tcp_max_syn_backlog
内核参数来使配置生效。ssl.cipher.suites
type: list
Default: ""
Importance: medium
Dynamic update: per-broker
密码套件列表。这是验证、加密、MAC 和密钥交换算法的命名组合,用于通过 TLS 或 SSL 网络协议协商网络连接的安全设置。默认情况下,支持所有可用的密码套件。
ssl.client.auth
type: string
Default: none
Valid Values: [required, requested, none]
Importance: medium
Dynamic update: per-broker
配置 kafka 代理以请求客户端身份验证。以下设置是通用的:
-
需要
SSL.client.auth=required
If 设为所需的客户端身份验证。 -
SSL.client.auth=requested
这意味着客户端身份验证是可选的。如果设置了这个选项,则客户端可以选择不提供对其验证信息的验证信息 -
SSL.client.auth=none
意味着不需要客户端身份验证。
-
需要
ssl.enabled.protocols
type: list
Default: TLSv1.2,TLSv1.3
Importance: medium
Dynamic update: per-broker
为 SSL 连接启用的协议列表。使用 Java 11 或更新版本运行时,默认值为 'TLSv1.2,TLSv1.3',否则为 'TLSv1.2'。使用 Java 11 的默认值时,如果客户端和服务器同时支持 TLSv1.3,则客户端和服务器将首选 TLSv1.3,否则将回退到 TLSv1.2(假设这两个支持至少支持 TLSv1.2)。在大多数情况下,这个默认设置应该是正常的。另请参阅
ssl.protocol
的配置文档。ssl.key.password
type: password
Default: null
Importance: medium
Dynamic update: per-broker
密钥存储文件或在"ssl.keystore.key"中指定的 PEM 密钥的密码。只有在配置了双向验证时,客户端才需要这样做。
ssl.keymanager.algorithm
type: string
Default: SunX509
Importance: medium
Dynamic update: per-broker
密钥管理器工厂用于 SSL 连接的算法。默认值为 Java 虚拟机配置的主要管理器工厂算法。
ssl.keystore.certificate.chain
type: password
Default: null
Importance: medium
Dynamic update: per-broker
使用由 'ssl.keystore.type" 指定的格式的证书链。默认 SSL 引擎工厂仅支持使用 X.509 证书列表的 PEM 格式。
ssl.keystore.key
type: password
Default: null
Importance: medium
Dynamic update: per-broker
使用 'ssl.keystore.type" 指定的格式的私钥。默认 SSL 引擎工厂只支持使用 PKCS#8 密钥的 PEM 格式。如果密钥已加密,则必须使用 'ssl.key.password' 指定密钥密码。
ssl.keystore.location
type: string
Default: null
Importance: medium
Dynamic update: per-broker
密钥存储文件的位置。对于客户端,这是可选的,可用于客户端双向身份验证。
ssl.keystore.password
type: password
Default: null
Importance: medium
Dynamic update: per-broker
密钥存储文件的存储密码。对于客户端,这是可选的,只有在配置了 'ssl.keystore.location' 时才需要。PEM 格式不支持密钥存储密码。
ssl.keystore.type
type: string
Default: JKS
Importance: medium
Dynamic update: per-broker
密钥存储文件的文件格式。对于客户端,这是可选的。
ssl.protocol
type: string
Default: TLSv1.3
Importance: medium
Dynamic update: per-broker
用于生成 SSLContext 的 SSL 协议。使用 Java 11 或更新版本运行时,默认值为 'TLSv1.3',否则为 'TLSv1.2'。对于大多数用例来说,这个值应该非常正常。最近 JVM 中允许的值是 'TLSv1.2' 和 'TLSv1.3'。'TLS', 'TLSv1.1', 'SSL', 'SSLv2' 和 'SSLv3' 在以前的 JVM 中可能被支持,但由于已知的安全漏洞,不建议使用它们。使用这个配置和 'ssl.enabled.protocols' 的默认值,如果服务器不支持 'TLSv1.3',客户端将降级为 'TLSv1.2'。如果此配置被设置为 'TLSv1.2',则客户端不会使用 'TLSv1.3',即使它是 ssl.enabled.protocols 中的值之一,服务器仅支持 'TLSv1.3'。
ssl.provider
type: string
Default: null
Importance: medium
Dynamic update: per-broker
用于 SSL 连接的安全供应商名称。默认值是 JVM 的默认安全供应商。
ssl.trustmanager.algorithm
type: string
Default: PKIX
Importance: medium
Dynamic update: per-broker
信任管理器工厂用于 SSL 连接的算法。默认值为 Java 虚拟机配置的信任管理器工厂算法。
ssl.truststore.certificates
type: password
Default: null
Importance: medium
Dynamic update: per-broker
可信证书,格式为 'ssl.truststore.type'。默认 SSL 引擎工厂仅支持使用 X.509 证书使用 PEM 格式。
ssl.truststore.location
type: string
Default: null
Importance: medium
Dynamic update: per-broker
信任存储文件的位置。
ssl.truststore.password
type: password
Default: null
Importance: medium
Dynamic update: per-broker
信任存储文件的密码。如果没有设置密码,则仍然将使用配置的信任存储文件,但会禁用完整性检查。信任存储密码不支持 PEM 格式。
ssl.truststore.type
type: string
Default: JKS
Importance: medium
Dynamic update: per-broker
信任存储文件的文件格式。
zookeeper.clientCnxnSocket
type: string
Default: null
Importance: medium
Dynamic update: read-only
在使用 ZooKeeper 的 TLS 连接时,通常设置为
org.apache.zookeeper.ClientCnxnSocketNetty
。覆盖通过同名zookeeper.clientCnxnSocket
系统属性设置的任何显式值。zookeeper.ssl.client.enable
type: boolean
Default: false
Importance: medium
Dynamic update: read-only
将客户端设置为在连接到 ZooKeeper 时使用 TLS。一个显式值会覆盖通过
zookeeper.client.secure
系统属性设置的任何值(注意不同的名称)。如果未设置任何设置,则默认为 false;如果为 true,zookeeper.clientCnxnSocket
必须设置为org.apache.zookeeper.ClientCnxnSocketNetty
);其他要设置的值可能包括zookeeper.ssl.cipher.suites
、zookeeper.ssl.crl.enable、zookeeper.ssl.crl.enable
。zookeeper.ssl.enabled.protocol.protocols
,zookeeper.ssl.endpoint.identification.algorithm
,zookeeper.ssl.keystore.location
,zookeeper.ssl.keystore.password
,zookeeper.ssl.keystore.type
,zookeeper.ssl.ocsp.enable
, , zookeeper.ssl.keystore.zookeeper.ssl.protocol
,zookeeper.ssl.truststore.location
,zookeeper.ssl.truststore.password
,zookeeper.ssl.truststore.type
.zookeeper.ssl.keystore.location
type: string
Default: null
Importance: medium
Dynamic update: read-only
将客户端证书与 ZooKeeper 的 TLS 连接一起使用时的密钥存储位置。覆盖通过
zookeeper.ssl.keyStore.location
系统属性设置的任何显式值(请注意 camelCase)。zookeeper.ssl.keystore.password
type: password
Default: null
Importance: medium
Dynamic update: read-only
将客户端证书与 ZooKeeper 的 TLS 连接一起使用时,密钥存储密码。覆盖通过
zookeeper.ssl.keyStore.password
系统属性设置的任何显式值(请注意 camelCase)。请注意,ZooKeeper 不支持与密钥存储密码不同的密钥密码,因此请确保将密钥存储中的密钥密码设置为与密钥存储密码相同;否则,尝试 Zookeeper 的连接将失败。zookeeper.ssl.keystore.type
type: string
Default: null
Importance: medium
Dynamic update: read-only
将客户端证书与 ZooKeeper 的 TLS 连接一起使用时,密钥存储类型。覆盖通过
zookeeper.ssl.keyStore.type
系统属性设置的任何显式值(请注意 camelCase)。null
默认值表示将根据密钥存储的文件名扩展名来自动检测类型。zookeeper.ssl.truststore.location
type: string
Default: null
Importance: medium
Dynamic update: read-only
使用 TLS 连接 ZooKeeper 时的信任存储位置。覆盖通过
zookeeper.ssl.trustStore.location
系统属性设置的任何显式值(请注意 camelCase)。zookeeper.ssl.truststore.password
type: password
Default: null
Importance: medium
Dynamic update: read-only
使用 TLS 连接到 ZooKeeper 时信任存储密码。覆盖通过
zookeeper.ssl.trustStore.password
系统属性设置的任何显式值(请注意 camelCase)。zookeeper.ssl.truststore.type
type: string
Default: null
Importance: medium
Dynamic update: read-only
使用 TLS 连接 ZooKeeper 时的 truststore 类型。覆盖通过
zookeeper.ssl.trustStore.type
系统属性设置的任何显式值(请注意 camelCase)。null
默认值表示根据信任存储的文件名扩展自动探测 类型。alter.config.policy.class.name
type: class
Default: null
Importance: low
Dynamic update: read-only
更改配置策略类,用于验证。类应该实现
org.apache.kafka.server.policy.AlterConfigPolicy
接口。alter.log.dirs.replication.quota.window.num
type: int
Default: 11
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
要在内存中保留的示例数量,用于更改日志缓存复制配额。
alter.log.dirs.replication.quota.window.size.seconds
type: int
Default: 1
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
用于更改日志缓存复制配额的每个示例的时间范围。
authorizer.class.name
type: string
Default: ""
Importance: low
Dynamic update: read-only
实现
org.apache.kafka.server.authorizer.Authorizer
接口的类的完全限定名称,代理将用于授权。client.quota.callback.class
type: class
Default: null
Importance: low
Dynamic update: read-only
实施 ClientQuotaCallback 接口的类的完全限定名称,用于决定应用到客户端请求的配额限制。默认情况下,使用 <user>、<client-id>、<user> 或 <client-id> 配额保存在 ZooKeeper 中。对于任何给定请求,与会话的用户主体匹配的最具体配额,并应用请求的 client-id。
connection.failed.authentication.delay.ms
type: int
Default: 100
Valid Values: [0,…]
Importance: low
Dynamic update: read-only
在身份验证失败时关闭连接延迟:这是连接关闭的时间(以毫秒为单位),在身份验证失败时会延迟连接。这必须配置为小于 connection.max.idle.ms,以防止连接超时。
controller.quorum.retry.backoff.ms
type: int
Default: 20
Importance: low
Dynamic update: read-only
尝试重试失败的请求到给定主题分区前等待的时间。这可避免在某些故障情况下在紧密循环中重复发送请求。
controller.quota.window.num
type: int
Default: 11
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
为控制器修改配额保留的内存中的示例数量。
controller.quota.window.size.seconds
type: int
Default: 1
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
控制器变异配额的每个示例的时间范围。
create.topic.policy.class.name
type: class
Default: null
Importance: low
Dynamic update: read-only
创建用于验证的主题策略类。该类应当实施
org.apache.kafka.server.policy.CreateTopicPolicy
接口。delegation.token.expiry.check.interval.ms
type: long
Default: 3600000(1 小时)
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
扫描间隔,以删除已过期的委托令牌。
kafka.metrics.polling.interval.secs
type: int
Default: 10
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
指标轮询间隔(以秒为单位)可用于 kafka.metrics.reporters 实现。
kafka.metrics.reporters
type: list
Default: ""
Importance: low
Dynamic update: read-only
用作 Yammer 指标定制报告器的类列表。报告器应该实现
kafka.metrics.KafkaMetricsReporter
特征。如果客户端想要在自定义报告器上公开 JMX 操作,则自定义报告器还需要额外实现扩展kafka.metrics.KafkaMetricsReporterMBean
特征的 MBean 特征,以便注册的 MBean 约定符合标准 MBean 规范。listener.security.protocol.map
Description: string
Default: PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
Importance: low
Dynamic update: per-broker
在侦听器名称和安全协议之间映射。这必须为相同的安全协议定义,才能在多个端口或 IP 中使用。例如,即使同时需要 SSL,也可以分隔内部和外部流量。以后,用户可以用名称 INTERNAL 和 EXTERNAL 定义监听器:INTERNAL
:SSL,EXTERNAL:SSL
。如上所示,键和值用冒号隔开,映射条目用逗号分开。每个监听器名称应只在映射中出现一次。可以通过将规范化前缀(小写)名添加到配置名称,为每个监听程序配置不同的安全(SSL 和 SASL)设置。例如,若要为 INTERNAL 侦听器设置不同的密钥存储,将设置具有名称listener.name.internal.ssl.keystore.location
的配置。如果没有设置监听程序名称的配置,则配置将回退到通用配置(如ssl.keystore.location
)。请注意,如果不提供显式映射并且不使用其他安全协议,则假定在 KRaft 是从controller.listener.names
到 PLAINTEXT 定义的监听程序名称的默认映射。log.message.downconversion.enable
type: boolean
Default: true
Importance: low
Dynamic update: cluster-wide
此配置控制是否启用了 down-conversion 的消息格式来满足请求量。当设置为
false
时,代理将不会为希望旧消息格式的用户执行 down-conversion。代理以UNSUPPORTED_VERSION
错误响应,以消耗来自此类老客户端的请求。此配置不适用于复制器可能需要的任何消息格式转换。metric.reporters
type: list
Default: ""
Importance: low
Dynamic update: cluster-wide
用作指标报告器的类列表。实施
org.apache.kafka.common.metrics.MetricsReporter
接口允许在新指标创建通知的类中插入。JmxReporter 始终包含在内以注册 JMX 统计。metrics.num.samples
type: int
Default: 2
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
为计算指标维护的示例数量。
metrics.recording.level
type: string
Default: INFO
Importance: low
Dynamic update: read-only
指标的最高记录级别。
metrics.sample.window.ms
type: long
Default: 30000 (30 秒)
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
计算指标示例的时间窗口。
password.encoder.cipher.algorithm
type: string
Default: AES/CBC/PKCS5Padding
Importance: low
Dynamic update: read-only
用于动态配置的密码的 Cipher 算法。
password.encoder.iterations
type: int
Default: 4096
Valid Values: [1024,…]
Importance: low
Dynamic update: read-only
用于动态配置的编码密码的迭代计数。
password.encoder.key.length
type: int
Default: 128
Valid Values: [8,…]
Importance: low
Dynamic update: read-only
用于动态配置的密码的编码密钥长度。
password.encoder.keyfactory.algorithm
type: string
Default: null
Importance: low
Dynamic update: read-only
用于动态配置的密码的 SecretKeyonnectionFactoryy 算法。默认为 PBKDF2WithHmacSHA512(如果可用),否则为 PBKDF2WithHmacSHA1。
quota.window.num
type: int
Default: 11
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
在内存中为客户端配额保留的示例数量。
quota.window.size.seconds
type: int
Default: 1
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
客户端配额的每个样本的时间范围。
replication.quota.window.num
type: int
Default: 11
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
在内存中保留复制配额的示例数量。
replication.quota.window.size.seconds
type: int
Default: 1
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
复制配额的每个示例的时间范围。
sasl.login.connect.timeout.ms
type: int
Default: null
Importance: low
Dynamic update: read-only
(可选)外部身份验证供应商连接超时的值(以毫秒为单位)。目前仅适用于 OAUTHBEARER。
sasl.login.read.timeout.ms
type: int
Default: null
Importance: low
Dynamic update: read-only
(可选)外部身份验证提供程序读取超时的值(以毫秒为单位)。目前仅适用于 OAUTHBEARER。
sasl.login.retry.backoff.max.ms
type: long
Default: 10000 (10 秒)
Importance: low
Dynamic update: read-only
(可选)登录尝试外部身份验证提供程序之间等待的最长这个值(可选)值(以毫秒为单位)。登录使用 sasl.login.retry.backoff.ms 设置的初始等待算法,并在尝试为 sasl.login.retry.backoff.max.ms 设置指定的最大等待长度之间加倍长度。目前仅适用于 OAUTHBEARER。
sasl.login.retry.backoff.ms
type: long
Default: 100
Importance: low
Dynamic update: read-only
(可选)在登录尝试外部身份验证提供程序之间初始等待的值(可选)值(以毫秒为单位)。登录使用 sasl.login.retry.backoff.ms 设置的初始等待算法,并在尝试为 sasl.login.retry.backoff.max.ms 设置指定的最大等待长度之间加倍长度。目前仅适用于 OAUTHBEARER。
sasl.oauthbearer.clock.skew.seconds
type: int
Default: 30
Importance: low
Dynamic update: read-only
(可选)值,允许 OAuth/OIDC 身份提供程序和代理的时间不同。
sasl.oauthbearer.expected.audience
type: list
Default: null
Importance: low
Dynamic update: read-only
(可选)代理的逗号分隔设置,用于验证已经为其中一个预期对象发布 JWT。为标准的 OAuth "aud" 声明将检查 JWT,如果设置了这个值,代理将匹配 JWT 的"aud"声明的值,以查看是否存在准确的匹配项。如果没有匹配项,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.expected.issuer
type: string
Default: null
Importance: low
Dynamic update: read-only
(可选)用于验证 JWT 是由预期签发者创建的代理的设置。将检查标准的 OAuth "iss" 声明,如果设置了此值,代理就与 JWT 的"iss"声明中的内容完全匹配。如果没有匹配项,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.jwks.endpoint.refresh.ms
type: long
Default: 3600000(1 小时)
Importance: low
Dynamic update: read-only
(可选)代理在刷新其 JWKS(JSON Web Key Set)缓存时要等待的值,其中包含用于验证 JWT 签名的密钥。
sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms
type: long
Default: 10000 (10 秒)
Importance: low
Dynamic update: read-only
(可选)尝试从外部身份验证提供程序检索 JWKS(JSON Web Key Set)之间的最大等待的值(以毫秒为单位)。JWKS 检索使用基于 sasl.oauthbearer.jwks.endpoint.retry.max.ms 设置的初始等待长度的指数级 backoff 算法,并在 sasl.oauthbearer.jwks.max.ms 设置指定的最大等待时间。
sasl.oauthbearer.jwks.endpoint.retry.backoff.ms
type: long
Default: 100
Importance: low
Dynamic update: read-only
(可选)初始等待的值(JSON Web Key Set)从外部身份验证提供程序检索尝试。JWKS 检索使用基于 sasl.oauthbearer.jwks.endpoint.retry.max.ms 设置的初始等待长度的指数级 backoff 算法,并在 sasl.oauthbearer.jwks.max.ms 设置指定的最大等待时间。
sasl.oauthbearer.scope.claim.name
type: string
Default: scope
Importance: low
Dynamic update: read-only
范围的 OAuth 声明通常命名为"scope",但此(可选)设置可以提供用于 JWT 有效负载声明中包含的范围的不同名称,如果 OAuth/OIDC 供应商将该声明使用不同的名称。
sasl.oauthbearer.sub.claim.name
type: string
Default: sub
Importance: low
Dynamic update: read-only
主体的 OAuth 声明通常命名为"sub",但此(可选)设置可以为 JWT 有效负载声明中包含的主题提供不同的名称,如果 OAuth/OIDC 供应商将该声明使用不同的名称。
security.providers
type: string
Default: null
Importance: low
Dynamic update: read-only
每次返回供应商实施安全算法的可配置创建者类列表。这些类应当实施
org.apache.kafka.common.security.auth.SecurityProviderCreator
接口。ssl.endpoint.identification.algorithm
type: string
Default: https
Importance: low
Dynamic update: per-broker
使用服务器证书验证服务器主机名的端点标识算法。
ssl.engine.factory.class
type: class
Default: null
Importance: low
Dynamic update: per-broker
类型为 org.apache.kafka.common.security.auth.SslEngineFactory 的类来提供 SSLEngine 对象。默认值为 org.apache.kafka.common.ssl.DefaultSslEngineFactory。
ssl.principal.mapping.rules
type: string
Default: DEFAULT
Importance: low
Dynamic update: read-only
用于映射的规则列表与客户端证书区分名称到短名称。规则按照顺序评估,第一个与主体名称匹配的规则被用来将其映射到一个短名称。之后的所有规则将被忽略。默认情况下,区分 X.500 证书的名称将是主体。有关格式的更多详情,请参阅安全 授权和 acls。请注意,如果
principal.builder.class
配置提供了 KafkaPrincipalBuilder 的扩展,则此配置会被忽略。ssl.secure.random.implementation
type: string
Default: null
Importance: low
Dynamic update: per-broker
用于 SSL 加密操作的 SecureRandom PRNG 实施。
transaction.abort.timed.out.transaction.cleanup.interval.ms
type: int
Default: 10000 (10 秒)
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
恢复已超时的事务的时间间隔。
transaction.remove.expired.transaction.cleanup.interval.ms
type: int
Default: 3600000(1 小时)
Valid Values: [1,…]
Importance: low
Dynamic update: read-only
移除由于 transaction.
id.expiration.ms
所经过的事务所过期的时间间隔。zookeeper.ssl.cipher.suites
type: list
Default: null
Importance: low
Dynamic update: read-only
指定在 ZooKeeper TLS 协商(csv)中使用的启用的密码套件。覆盖通过
zookeeper.ssl.ciphersuites
系统属性设置的任何显式值(注意单个单词"ciphersuites")。null
默认值表示启用的密码套件列表由所使用的 Java 运行时决定。zookeeper.ssl.crl.enable
type: boolean
Default: false
Importance: low
Dynamic update: read-only
指定是否在 ZooKeeper TLS 协议中启用证书撤销列表。覆盖通过
zookeeper.ssl.crl
系统属性设置的任何显式值(请注意较短的名称)。zookeeper.ssl.enabled.protocols
type: list
Default: null
Importance: low
Dynamic update: read-only
指定 ZooKeeper TLS 协商(csv)中的已启用的协议。覆盖通过
zookeeper.ssl.enabledProtocols
系统属性设置的任何显式值(请注意 camelCase)。默认值null
表示启用的协议将是zookeeper.ssl.protocol
配置属性值。zookeeper.ssl.endpoint.identification.algorithm
type: string
Default: HTTPS
Importance: low
Dynamic update: read-only
指定是否在 ZooKeeper TLS 协商过程中启用主机名验证,同时(不区分大小写)"https"表示启用了 ZooKeeper 主机名验证,并明确的空白值表示禁用(禁止测试目的)。一个显式值会覆盖通过
zookeeper.ssl.hostnameVerification
系统属性设置的任何 "true" 或 "false" 值(注意不同的名称和值;true 表示 https 和 false 表示空白)。zookeeper.ssl.ocsp.enable
type: boolean
Default: false
Importance: low
Dynamic update: read-only
指定是否在 ZooKeeper TLS 协议中启用在线证书状态协议。覆盖通过
zookeeper.ssl.ocsp
系统属性设置的任何显式值(请注意较短的名称)。zookeeper.ssl.protocol
type: string
Default: TLSv1.2
Importance: low
Dynamic update: read-only
指定 ZooKeeper TLS 协商中使用的协议。一个显式值会覆盖通过相同名为
zookeeper.ssl.protocol
系统属性设置的任何值。
第 2 章 主题配置属性
cleanup.policy
type: list
Default: delete
Valid Values: [compact, delete]
Server Default Property: log.cleanup.policy
Importance: medium
是"删除"或"compact"或两者的字符串。这个字符串指定了在旧日志片段中使用的保留策略。当达到其保留时间或大小限制时,默认策略("删除")将丢弃旧的片段。"compact"设置将在主题中启用 日志压缩。
compression.type
type : string
Default: producer
Valid Values: [uncompressed, zstd, lz4, snappy, gzip, producer]
Server Default Property: compression.type
Importance: medium
指定给定主题的最后一个压缩类型。此配置接受标准的压缩码c('gzip', 'snappy', 'lz4', 'zstd')。它还接受"未压缩的",它等同于没有压缩 ; 和 'producer' 表示由制作者保留原始压缩代码。
delete.retention.ms
type: long
Default: 86400000(1 天)
Valid Values: [0,…]
Server Default Property: log.cleaner.delete.retention.ms
Importance: medium
为 日志压缩主题保留 delete tombstone 标记的时间长度。此设置还可在消费者开始从偏移 0 开始读取的时间绑定,以确保它们获得最终阶段的有效快照(在完成扫描之前,可以收集到其他的删除操作)。
file.delete.delay.ms
type: long
Default: 60000(1 minute)
Valid Values: [0,…]
Server Default Property: log.segment.delete.delay.ms
Importance: medium
从文件系统中删除文件前等待的时间。
flush.messages
type: long
Default: 9223372036854775807
Valid Values: [0,…]
Server Default Property: log.flush.interval.messages
Importance: medium
此设置允许指定一个间隔,我们将强制 fsync 写入日志中的数据。例如,如果将其设置为 1,在每次消息后都会进行 fsync;如果消息是 5,则每个五条消息后都会进行 fsync。通常,我们建议您不要为持久性设置并使用复制功能,并允许操作系统后台清除功能,因为它效率更高。此设置可以逐个主题覆盖(请参阅 各主题配置部分)。
flush.ms
type: long
Default: 9223372036854775807
Valid Values: [0,…]
Server Default Property: log.flush.interval.ms
Importance: medium
此设置允许指定时间间隔,在其中会强制重新同步写入日志的数据。例如,如果此值设为 1000,则在通过 1000 ms 后,我们将 fsync 过。通常,我们建议您不要为持久性设置并使用复制功能,并允许操作系统后台清除功能,因为它效率更高。
follower.replication.throttled.replicas
Description: list
Default: ""
Valid Values: [partitionId],[partitionId]:[brokerId]:[brokerId],…
Server Default Property: follower.replication.throttled.replicas
Importance: medium
在后面的一个副本列表中,日志复制应该节流。该列表应当描述一组副本,格式为 [PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:… or alternatively the wildcard '*' 可用于调整此主题的所有副本。
index.interval.bytes
type: int
Default: 4096(4 kibibytes)
Valid Values: [0,…]
Server Default Property: log.index.interval.bytes
Importance: medium
此设置控制 Kafka 在其偏移索引中添加索引条目的频率。默认设置可确保每个 4096 字节都只索引一个消息。通过索引,读取可以更接近日志里的确切位置,但索引越大。您可能不需要更改。
leader.replication.throttled.replicas
Description: list
Default: ""
Valid Values: [partitionId],[partitionId]:[brokerId]:[brokerId],…
Server Default Property: leader.replication.throttled.replicas
Importance: medium
在领导端应节流日志复制的副本列表。该列表应当描述一组副本,格式为 [PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:… or alternatively the wildcard '*' 可用于调整此主题的所有副本。
max.compaction.lag.ms
Description: long
Default: 9223372036854775807
Valid Values: [1,…]
Server Default Property: log.cleaner.max.compaction.max.compaction.lag.ms
Importance: medium
在日志中,消息会保持在日志中的压缩状态的最长时间。只适用于正在压缩的日志。
max.message.bytes
type: int
Default: 1048588
Valid Values: [0,…]
Server Default Property: message.max.bytes
Importance: medium
Kafka 允许的最大记录批处理大小(如果启用了压缩,则经过压缩)。如果这个增大并且存在大于 0.10.2 的消费者,则必须增加消费者的获取大小,以便他们能够获取此大型的记录批处理。在最新的消息格式版本中,记录始终分组到批处理中以提高效率。在以前的消息格式版本中,未将未压缩记录分组为批处理,这种限制仅适用于该示例中的单个记录。
message.format.version
type: string
Default: 3.0-IV1 Valid Values: [0.8.0, 0.8.1, 0.8.2, 0.9.0, 0.10.0-IV0, 0.10.0-IV1, 0.10.1-IV1, 0.10.1-IV1, 0.10.1-IV2, 0.10.2-IV0, 0.10.0-IV0
0.11.0-iv0, 0.11.0-iv1, 0.11.0-iv2, 1.0-iv0, 1.1-iv0, 2.0-iv0, 2.1-iv1, 2.1-iv1, 2.1-iv2, 2.2-iv0, 2.2-iv1, 2.3-iv0, 2.3-iv0, 2.3-iv1, 2.3-iv1, 2.4-IV0, 2.4-IV1, 2.5-IV0, 2.6-IV0, 2.7-IV0, 2.7-IV1, 2.7-IV2, 2.8-IV0, 2.8-IV1, 3.0-IV0, 3.0-IV1, 3.1-IV0, 3.2-IV0] Server Default Property: log.message.format.version
[DEPRECATED] 指定代理将使用的消息格式版本,将消息附加到日志中。如果
inter.broker.protocol.version
是3.0
或更高版本(忽略实际配置值),则此配置的值始终假定为 3.0。否则,该值应该是有效的 ApiVersion。一些示例为: 0.10.0、1.1、2.8、3.0。通过设置特定的消息格式版本,用户需要验证磁盘上的所有现有信息都小于或等于指定版本。设定这个值会错误地导致具有旧版本的用户出现问题,因为它们会收到没有理解的格式的消息。message.timestamp.difference.max.ms
Description: long
Default: 9223372036854775807
Valid Values: [0,…]
Server Default Property: log.message.timestamp.difference.max.ms
Importance: medium
代理收到消息和消息中指定的时间戳之间允许的最大差别。如果 message.timestamp.type=CreateTime,如果时间戳超过这个阈值,则会拒绝一条消息。如果 message.timestamp.type=LogAppendTime,则此配置会被忽略。
message.timestamp.type
type : string
Default: CreateTime
Valid Values: [CreateTime, LogAppendTime]
Server Default Property: log.message.timestamp.type
Importance: medium
定义消息中的时间戳是消息创建时间或日志附加时间。该值应该是
CreateTime
或LogAppendTime
。min.cleanable.dirty.ratio
type: double
Default: 0.5
Valid Values: [0,…,1]
Server Default Property: log.cleaner.min.clean.ratio
Importance: medium
此配置控制日志紧凑器尝试清理日志的频率(假设 日志压缩 被启用)。默认情况下,我们将避免清理超过 50% 的日志。这种比例使日志空间达到最大的空间(最多 50% 的日志重复)。较高的比率意味着较少的、效率更高的清理,但在日志中意味着更多空间。如果还指定了 max.compaction.lag.ms 或 min.compaction.lag.ms 配置,则日志压缩器会在日志被认为在达到(i)达到脏比率阈值时,日志至少具有 min.compaction.lag.lag.lag.ms 持续时间(未匹配)的记录。 或者(ii)如果日志在 max.compaction.lag.lag.ms 周期内有脏(不相关)记录。
min.compaction.lag.ms
Description: long
Default: 0
Valid Values: [0,…]
Server Default Property: log.cleaner.min.compaction.lag.ms
Importance: medium
日志消息在日志中保持不限定的最小时间。只适用于正在压缩的日志。
min.insync.replicas
type: int
Default: 1
Valid Values: [1,…]
Server Default Property: min.insync.replicas
Importance: medium
当制作者将 acks 设为 "all"(或 "-1"),此配置指定了必须确认写入操作成功的最小副本数。如果达到此最小值,则生产者将引发异常(NotEnoughReplicas 或 NotEnoughReplicasAppend)。当同时使用
min.insync.replicas
和acks
时,您可以强制实现更大的持久保障。典型的场景是创建包含复制原因 3 的主题,将min.insync.replicas
设置为 2,并使用"all"进行生成。这将确保,在大多数副本没有收到写入时,生产者会引发异常。
预分配
type: boolean
Default: false
Server Default Property: log.preallocate
Importance: medium
满足在创建新日志段时,应在磁盘上预分配该文件。
retention.bytes
Type: long
Default: -1
Server Default Property: log.retention.bytes
Importance: medium
此配置控制分区(由日志片段)在我们使用"删除"保留策略时丢弃旧日志片段,以释放空间的最大大小。默认情况下,大小限制仅有一个时间限制。由于这个限制在分区级别强制实施,因此根据分区的数量来计算主题保留(以字节为单位)。
retention.ms
type: long
Default: 604800000(7 days)
Valid Values: [-1,…]
Server Default Property: log.retention.ms
Importance: medium
此配置会控制在我们使用"删除"保留策略时丢弃旧日志片段,以释放空间前的最长时间。这代表一个 SLA,了解消费者必须如何读取其数据。如果设置为 -1,则不应用时间限制。
segment.bytes
type: int
Default: 1073741824(1 gibibyte)
Valid Values: [14,…]
Server Default Property: log.segment.bytes
Importance: medium
此配置控制日志的片段文件大小。保留和清理总是一次完成一个文件,因此较大的片段大小意味着较少的文件,但对保留的更精细的控制。
segment.index.bytes
type: int
Default: 10485760(10 mebibytes)
Valid Values: [0,…]
Server Default Property: log.index.size.max.bytes
Importance: medium
此配置控制将偏移映射到文件位置的索引大小。我们预先分配这个索引文件,只有在日志推出后将其缩小。您通常不需要更改此设置。
segment.jitter.ms
type: long
Default: 0
Valid Values: [0,…]
Server Default Property: log.roll.jitter.ms
Importance: medium
从调度的网段滚动时间中减去最大随机危机,以避免在网段滚动中产生任何影响。
segment.ms
type: long
Default: 604800000(7 days)
Valid Values: [1,…]
Server Default Property: log.roll.ms
Importance: medium
此配置控制在这段文件未满的情况下,Kafka 将强制日志推出的时间,以确保保留可以删除或紧凑旧数据。
unclean.leader.election.enable
type: boolean
Default: false
Server Default Property: unclean.leader.election.enable
Importance: medium
指明是否将 ISR 集中的副本作为最后的手段被选为领导人,即使这样做可能会导致数据丢失。
message.downconversion.enable
type: boolean
Default: true
Server Default Property: log.message.downconversion.enable
Importance: low
此配置控制是否启用了 down-conversion 的消息格式来满足请求量。当设置为
false
时,代理将不会为希望旧消息格式的用户执行 down-conversion。代理以UNSUPPORTED_VERSION
错误响应,以消耗来自此类老客户端的请求。此配置不适用于复制器可能需要的任何消息格式转换。
第 3 章 消费者配置属性
key.deserializer
type: class
Importance: high
Deserializer 类用于实现
org.apache.kafka.common.serialization.Deserializer
接口的关键。value.deserializer
type: class
Importance: high
Deserializer 类用于实现
org.apache.kafka.common.serialization.Deserializer
接口的值。bootstrap.servers
type: list
Default: ""
Valid Values: non-null string
Importance: high
用于建立到 Kafka 集群的初始连接的主机/端口对列表。客户端将使用此处为 bootstrap 指定了哪些服务器的所有服务器 - 此列表仅影响用于发现全部服务器集的初始主机。此列表应当采用
host1:port1,host2:port2,…
的形式。由于这些服务器仅用于初始连接,以发现完整的群集成员身份(可能会动态更改),因此该列表不包含完整的服务器集合(尽管服务器停机时,您可能需要多个服务器)。fetch.min.bytes
type: int
Default: 1
Valid Values: [0,…]
Importance: high
服务器应返回获取请求的最小数据量。如果数据不足,请求会等待大量数据在回答请求前等待累积数据。默认设置为 1 字节意味着,获取请求一旦有单一字节数据可用,或获取请求超时等待数据到达时立即回答。将其设置为大于 1 的对象将会导致服务器等待大量数据累积,从而以某些额外延迟的成本提高服务器吞吐量。
group.id
type: string
Default: null
Importance: high
此使用者所属的标识消费者组的唯一字符串。如果使用者使用 subscription
(topic)
或基于 Kafka 的偏移管理策略,则需要此属性。heartbeat.interval.ms
type: int
Default: 3000 (3 秒)
Importance: high
使用 Kafka 的组群管理工具时,心跳到消费者协调器之间的预期时间。心跳用于确保消费者的会话保持活跃状态,并在新消费者加入或离开该组时促进重新平衡。该值必须小于
session.timeout.ms
,但通常设置的值通常不应超过 1/3。它可以被调整,以便控制正常重新平衡的预期时间。max.partition.fetch.bytes
type: int
Default: 1048576(1 mebibyte)
Valid Values: [0,…]
Importance: high
服务器的每个分区中要返回的最大数据量。记录在批处理中由消费者获取。如果获取的第一个非空分区中的第一个记录批处理大于这个限制,则批处理仍将返回,以确保消费者能够进行进度。代理接受的最大记录批处理大小通过
message.max.bytes
(broker config)或max.message.bytes
(topic 配置)定义。有关限制消费者请求大小,请参阅 fetch.max.bytes。session.timeout.ms
type: int
Default: 45000(45 秒)
Importance: high
在使用 Kafka 的组管理工具时,用于检测客户端失败的超时时间。客户端发送定期心跳,以指示其对代理的存活度。如果在这个会话超时前代理没有收到 heartbeat,则代理会从组中删除此客户端并启动重新平衡。请注意,该值必须在代理配置中由
group.min.session.timeout.ms
和group.max.session.timeout.ms
配置中的允许范围内配置。ssl.key.password
type: password
Default: null
Importance: high
密钥存储文件或在"ssl.keystore.key"中指定的 PEM 密钥的密码。只有在配置了双向验证时,客户端才需要这样做。
ssl.keystore.certificate.chain
type: password
Default: null
Importance: high
使用由 'ssl.keystore.type" 指定的格式的证书链。默认 SSL 引擎工厂仅支持使用 X.509 证书列表的 PEM 格式。
ssl.keystore.key
type: password
Default: null
Importance: high
使用 'ssl.keystore.type" 指定的格式的私钥。默认 SSL 引擎工厂只支持使用 PKCS#8 密钥的 PEM 格式。如果密钥已加密,则必须使用 'ssl.key.password' 指定密钥密码。
ssl.keystore.location
type: string
Default: null
Importance: high
密钥存储文件的位置。对于客户端,这是可选的,可用于客户端双向身份验证。
ssl.keystore.password
type: password
Default: null
Importance: high
密钥存储文件的存储密码。对于客户端,这是可选的,只有在配置了 'ssl.keystore.location' 时才需要。PEM 格式不支持密钥存储密码。
ssl.truststore.certificates
type: password
Default: null
Importance: high
可信证书,格式为 'ssl.truststore.type'。默认 SSL 引擎工厂仅支持使用 X.509 证书使用 PEM 格式。
ssl.truststore.location
type: string
Default: null
Importance: high
信任存储文件的位置。
ssl.truststore.password
type: password
Default: null
Importance: high
信任存储文件的密码。如果没有设置密码,则仍然将使用配置的信任存储文件,但会禁用完整性检查。信任存储密码不支持 PEM 格式。
allow.auto.create.topics
type: boolean
Default: true
Importance: medium
在订阅某个主题或分配主题时,允许在代理上创建自动主题。只有在代理允许使用
auto.create.topics.enable
broker 配置时,才会自动创建要订阅的主题。当使用早于 0.11.0 的代理时,此配置必须设置为false
。auto.offset.reset
type: string
Default: latest
Valid Values: [latest, earliest, none]
Importance: medium
当 Kafka 中没有初始偏移,或者服务器中不存在当前的偏移量时(例如,因为删除了这些数据,则应该怎么办):
- earliest:自动将偏移重置为最早的偏移
- latest:自动将偏移重置为最新偏移
- none:为使用者组发现之前误差,则抛出异常
- 任何其他对象:给消费者引发异常。
client.dns.lookup
type: string
Default: use_all_dns_ips
Valid Values: [use_all_dns_ips, resolve_canonical_bootstrap_servers_only]
Importance: medium
控制客户端如何使用 DNS 查找。如果设置为
use_all_dns_ips
,则按顺序连接到每个返回的 IP 地址,直到建立成功连接为止。断开连接后,将使用下一个 IP。使用完所有 IP 后,客户端会从主机名再次解析 IP(JVM 和 OS 缓存 DNS 名称查找)。如果设置为resolve_canonical_bootstrap_servers_only
,将每个 bootstrap 地址解析为规范名称列表。在 bootstrap 阶段后,这个行为与use_all_dns_ips
相同。connections.max.idle.ms
type: long
Default: 540000 (9 分钟)
Importance: medium
在此配置指定的毫秒数后关闭闲置连接。
default.api.timeout.ms
type: int
Default: 60000(1 minute)
Valid Values: [0,…]
Importance: medium
指定客户端 API 的超时(以毫秒为单位)。此配置用作所有没有指定
超时
参数的客户端操作的默认超时时间。enable.auto.commit
type: boolean
Default: true
Importance: medium
如果为 true,则使用者的偏移将定期在后台提交。
exclude.internal.topics
type: boolean
Default: true
Importance: medium
与订阅模式匹配的内部主题是否应该不包括在订阅中。始终可以明确订阅内部主题。
fetch.max.bytes
type: int
Default: 52428800(50 mebibytes)
Valid Values: [0,…]
Importance: medium
服务器应返回获取请求的最大数据量。记录在消费者的批处理中,如果获取的第一个非空分区中超过这个值,则记录批处理仍将返回,以确保使用者能够进行进展。因此,这不是绝对最大值。代理接受的最大记录批处理大小通过
message.max.bytes
(broker config)或max.message.bytes
(topic 配置)定义。请注意,使用者并行执行多个获取。group.instance.id
type: string
Default: null
Importance: medium
由最终用户提供的消费者实例的唯一标识符。只允许非空字符串。如果设置,使用者被视为静态成员,这意味着任何时候都只允许具有此 ID 的实例。这可与较大的会话超时结合使用,以避免出现临时不可用(如进程重启)导致的重新平衡问题。如果没有设置,消费者会将该组加入为动态成员,这是传统行为。
isolation.level
type: string
Default: read_uncommitted
Valid Values: [read_committed, read_uncommitted]
Importance: medium
控制如何读取事务写入的消息。如果设置为
read_committed
,则 consumer.poll()将只返回已提交的事务信息。如果设置为read_uncommited
(默认值),则 consumer.poll()将返回所有消息,甚至已中止的事务消息。在任一模式下,非事务性消息都会无条件地返回。消息始终以偏移顺序返回。因此,在
read_committed
模式下,consumer.poll()仅会将消息返回到最后稳定偏移(LSO),这比第一个开放事务的偏移小。在相关事务完成后,属于持续事务的消息后会出现任何信息。因此,当发生机交易时,read_committed
的客户将无法读取到高水位线。Further, when in `read_committed` the seekToEnd method will return the LSO .
max.poll.interval.ms
type: int
Default: 300000 (5 分钟)
Valid Values: [1,…]
Importance: medium
使用消费者组管理时调用 poll()之间的最大延迟。这样一来,在获取更多记录前,可以闲置这个时间。如果在此超时的过期前没有调用 poll(),则消费者被视为失败,组将重新平衡,以将分区重新分配给另一个成员。对于使用非null
group.instance.id
(达到此超时)的用户,分区不会被立即重新分配。相反,消费者将停止发送 heartbeat,分区将在session.timeout.ms
过期后重新分配。这会镜像已关闭的静态消费者的行为。max.poll.records
type: int
Default: 500
Valid Values: [1,…]
Importance: medium
单个调用中返回到 poll()中的最大记录数。请注意,
max.poll.records
不会影响底层获取行为。使用者将缓存来自每个获取请求的记录,并从每个轮询逐步返回记录。partition.assignment.strategy
type: list
Default: class org.apache.kafka.clients.consumer.RangeAssignor,class org.apache.kafka.clients.consumer.C teamingStickyAssignor
Valid Values: non-null string
Importance: medium
类名称或类类型的列表,按首选项排序,客户端将使用这些分区分配策略在消费者实例之间分发在使用组管理时在消费者实例之间分发分区所有权。可用的选项有:
-
org.apache.kafka.clients.consumer.RangeAssignor
: Assigns 分区基于各个主题。 -
org.apache.kafka.clients.consumer.Round SD
inAssignor : 以轮循方式将分区分配给消费者。 -
org.apache.kafka.clients.consumer.StickyAssignor
: Guarantee 分配一个最大平衡的分配,同时尽可能保留很多现有分区分配。 org.apache.kafka.clients.consumer.C teamingStickyAssignor
: Follows same StickyAssignor 逻辑,但允许合作重新平衡。默认分配器为 [RangeAssignor, C teamingStickyAssignor],它默认使用 RangeAssignor,但允许升级到 C collaborativeStickyAssignor,它只是一个滚动退回,从列表中删除 RangeAssignor。
通过实施
org.apache.kafka.clients.consumer.ConsumerPartitionAssignor
接口,您可以插入自定义分配策略。
-
receive.buffer.bytes
type: int
Default: 65536(64 kibibytes)
Valid Values: [-1,…]
Importance: medium
在读取数据时要使用的 TCP 接收缓冲区(SO_RCVBUF)的大小。如果值为 -1,则使用 OS 默认。
request.timeout.ms
type: int
Default: 30000 (30 秒)
Valid Values: [0,…]
Importance: medium
配置控制客户端等待请求响应的最大时间。如果在超时前没有收到响应,客户端会在需要时重新发送请求,或者在重试耗尽时失败请求。
sasl.client.callback.handler.class
type: class
Default: null
Importance: medium
实现了 AuthenticateCallbackHandler 接口的 SASL 客户端回调处理器类的完全限定名称。
sasl.jaas.config
type: password
Default: null
Importance: medium
JAAS 登录上下文参数,用于 SASL 连接,其格式由 JAAS 配置文件使用。此处 介绍了 JAAS 配置文件格式。该值格式为:
loginModuleClass controlFlag(optionName=optionValue)*;
.对于代理,配置必须在小写中带有监听前缀和 SASL 机制名称前缀。例如: listener.name.sasl_ssl.scram-sha-256.sasl.jaas.config=com.example.ScramLoginModule;.sasl.kerberos.service.name
type: string
Default: null
Importance: medium
Kafka 运行的 Kerberos 主体名称。这可以在 Kafka 的 JAAS 配置或 Kafka 配置中定义。
sasl.login.callback.handler.class
type: class
Default: null
Importance: medium
实现了 AuthenticateCallbackHandler 接口的 SASL 登录回调处理器类的完全限定名称。对于代理,登录回调处理器配置必须以监听器前缀和 SASL 机制名称作为前缀作为前缀。例如: listener.name.sasl_ssl.scram-sha-256.sasl.login.callback.handler.class=com.example.CustomScramLoginCallbackHandler。
sasl.login.class
type: class
Default: null
Importance: medium
实施登录接口的类的完全限定名称。对于代理,登录配置必须在小写中带有监听前缀和 SASL 机制名称前缀。For example, listener.name.sasl_ssl.scram-sha-256.sasl.login.class=com.example.CustomScramLogin.
sasl.mechanism
type: string
Default: GSSAPI
Importance: medium
用于客户端连接的 SASL 机制。这可以是提供安全提供程序的任何机制。GSSAPI 是默认机制。
sasl.oauthbearer.jwks.endpoint.url
type: string
Default: null
Importance: medium
可以检索来自提供程序的 JWKS(JSON Web Key Set) 的 OAuth/OIDC 供应商 URL。URL 可以是基于 HTTP(S)或基于文件的。如果 URL 基于 HTTP(S),则 JWKS 数据将通过代理启动时配置的 URL 从 OAuth/OIDC 供应商中检索。所有 then-current 密钥将缓存在代理中,用于传入请求。如果为 JWT 收到身份验证请求,如果包含尚未在缓存中但未包括在缓存中的"kid"标头声明值,则将根据需要再次查询 JWKS 端点。但是,代理会在收到任何 JWT 请求前轮询每个 sasl.oauthbearer.jwks.endpoint.refresh.ms 毫秒的 URL 来刷新缓存。如果 URL 是基于文件,代理将从启动时配置的位置载入 JWKS 文件。如果 JWT 包含不在 JWKS 文件中的"kid"标头值,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.token.endpoint.url
type: string
Default: null
Importance: medium
OAuth/OIDC 身份提供程序的 URL。如果 URL 基于 HTTP(S),则代表签发者的令牌端点 URL,请求将基于 sasl.jaas.config 中的配置进行登录。如果 URL 基于文件,它将指定包含由 OAuth/OIDC 身份提供程序发布的访问令牌(在 JWT 序列化表单中)的文件用于授权。
security.protocol
type: string
Default: PLAINTEXT
Importance: medium
用于与代理通信的协议。有效值为:PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL.
send.buffer.bytes
type: int
Default: 131072(128 kibibytes)
Valid Values: [-1,…]
Importance: medium
在发送数据时要使用的 TCP 发送缓冲区(SO_SNDBUF)的大小。如果值为 -1,则使用 OS 默认。
socket.connection.setup.timeout.max.ms
type: long
Default: 30000 (30 秒)
Importance: medium
客户端等待套接字连接建立的最大时间。对于每个连续的连接失败,连接设置超时将按这个最大值增加。为避免连接 storms,则 0.2 的随机化工厂将应用到超时,从而导致计算值上面的 20% 到 20% 的随机范围。
socket.connection.setup.timeout.ms
type: long
Default: 10000 (10 秒)
Importance: medium
客户端等待套接字连接建立的时间。如果在超时前没有构建连接,客户端将关闭套接字频道。
ssl.enabled.protocols
type: list
Default: TLSv1.2,TLSv1.3
Importance: medium
为 SSL 连接启用的协议列表。使用 Java 11 或更新版本运行时,默认值为 'TLSv1.2,TLSv1.3',否则为 'TLSv1.2'。使用 Java 11 的默认值时,如果客户端和服务器同时支持 TLSv1.3,则客户端和服务器将首选 TLSv1.3,否则将回退到 TLSv1.2(假设这两个支持至少支持 TLSv1.2)。在大多数情况下,这个默认设置应该是正常的。另请参阅
ssl.protocol
的配置文档。ssl.keystore.type
type: string
Default: JKS
Importance: medium
密钥存储文件的文件格式。对于客户端,这是可选的。
ssl.protocol
type: string
Default: TLSv1.3
Importance: medium
用于生成 SSLContext 的 SSL 协议。使用 Java 11 或更新版本运行时,默认值为 'TLSv1.3',否则为 'TLSv1.2'。对于大多数用例来说,这个值应该非常正常。最近 JVM 中允许的值是 'TLSv1.2' 和 'TLSv1.3'。'TLS', 'TLSv1.1', 'SSL', 'SSLv2' 和 'SSLv3' 在以前的 JVM 中可能被支持,但由于已知的安全漏洞,不建议使用它们。使用这个配置和 'ssl.enabled.protocols' 的默认值,如果服务器不支持 'TLSv1.3',客户端将降级为 'TLSv1.2'。如果此配置被设置为 'TLSv1.2',则客户端不会使用 'TLSv1.3',即使它是 ssl.enabled.protocols 中的值之一,服务器仅支持 'TLSv1.3'。
ssl.provider
type: string
Default: null
Importance: medium
用于 SSL 连接的安全供应商名称。默认值是 JVM 的默认安全供应商。
ssl.truststore.type
type: string
Default: JKS
Importance: medium
信任存储文件的文件格式。
auto.commit.interval.ms
type: int
Default: 5000(5 秒)
Valid Values: [0,…]
Importance: low
如果
enable.auto.commit
设为true
,则消费者偏移的频率自动给 Kafka。check.crcs
type: boolean
Default: true
Importance: low
自动检查所消耗的记录的 CRC32。这样可确保不会对消息发生任何有线或磁盘上的崩溃。这个检查增加了一些开销,因此在寻求性能时可能会禁用它。
client.id
type: string
Default: ""
Importance: low
在发出请求时要传递给服务器的 id 字符串。这样做的目的是通过允许逻辑应用程序名称包含在服务器端请求日志记录中,跟踪超过 ip/port 的请求源。
client.rack
type: string
Default: ""
Importance: low
此客户端的机架标识符。这可以是指示此客户端物理位置的任意字符串值。它与代理配置 'broker.rack' 对应。
fetch.max.wait.ms
type: int
Default: 500
Valid Values: [0,…]
Importance: low
如果没有足够的数据立即满足通过 fetch.min.bytes 指定的要求,则服务器在回答获取请求前将阻断的最大时间。
interceptor.classes
type: list
Default: ""
Valid Values: non-null string
Importance: low
用作拦截器的类列表。通过实施
org.apache.kafka.clients.consumer.ConsumerInterceptor
接口,可以截取消费者收到的记录(可能模拟)。默认情况下,没有拦截器。metadata.max.age.ms
type: long
Default: 300000 (5 分钟)
Valid Values: [0,…]
Importance: low
毫秒内的时间(以毫秒为单位),即使在我们未看到任何分区领导更改的情况下,也会强制更新元数据,以便主动发现任何新的代理或分区。
metric.reporters
type: list
Default: ""
Valid Values: non-null string
Importance: low
用作指标报告器的类列表。实施
org.apache.kafka.common.metrics.MetricsReporter
接口允许在新指标创建通知的类中插入。JmxReporter 始终包含在内以注册 JMX 统计。metrics.num.samples
type: int
Default: 2
Valid Values: [1,…]
Importance: low
为计算指标维护的示例数量。
metrics.recording.level
type: string
Default: INFO
Valid Values: [INFO, DEBUG, SSP]
Importance: low
指标的最高记录级别。
metrics.sample.window.ms
type: long
Default: 30000 (30 秒)
Valid Values: [0,…]
Importance: low
计算指标示例的时间窗口。
reconnect.backoff.max.ms
type: long
Default: 1000(1秒)
Valid Values: [0,…]
Importance: low
重新连接到重复连接失败的代理时要等待的最大时间(毫秒)。如果提供,每个主机的 backoff 将为每个连续的连接失败指数增加,最高最高。在计算了 backoff 增长后,会添加 20% 的随机 jitter 来避免连接停滞。
reconnect.backoff.ms
type: long
Default: 50
Valid Values: [0,…]
Importance: low
在尝试重新连接给定主机前要等待的基本时间。这可避免在紧张循环中重复连接到主机。这个 backoff 适用于客户端到代理的所有连接尝试。
retry.backoff.ms
type: long
Default: 100
Valid Values: [0,…]
Importance: low
尝试重试失败的请求到给定主题分区前等待的时间。这可避免在某些故障情况下在紧密循环中重复发送请求。
sasl.kerberos.kinit.cmd
type: string
Default: /usr/bin/kinit
Importance: low
Kerberos kinit 命令路径.
sasl.kerberos.min.time.before.relogin
type: long
Default: 60000
Importance: low
刷新尝试之间的登录线程睡眠时间。
sasl.kerberos.ticket.renew.jitter
type: double
Default: 0.05
Importance: low
添加至续订时间的随机jitter 的百分比。
sasl.kerberos.ticket.renew.window.factor
type: double
Default: 0.8
Importance: low
登录线程将休眠,直到达到从最后刷新到 ticket 的过期时间,此时它将尝试续订票据。
sasl.login.connect.timeout.ms
type: int
Default: null
Importance: low
(可选)外部身份验证供应商连接超时的值(以毫秒为单位)。目前仅适用于 OAUTHBEARER。
sasl.login.read.timeout.ms
type: int
Default: null
Importance: low
(可选)外部身份验证提供程序读取超时的值(以毫秒为单位)。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.buffer.seconds
type: short
Default: 300
Valid Values: [0,…,3600]
Importance: low
在刷新凭证时要维护的凭证前的缓冲时间(以秒为单位)。如果刷新将超过缓冲区秒数的过期时间,则刷新将移动至可维护尽可能多的缓冲区时间。如果未指定值,则使用法定值在 0 到 3600(1 小时)之间;如果没有指定值,则使用默认值 300(5 分钟)。如果值和 sasl.login.refresh.min.period.seconds,如果它们的总和超过凭证的剩余生命周期,则都会被忽略。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.min.period.seconds
type: short
Default: 60
Valid Values: [0,…,900]
Importance: low
在刷新凭证前等待登录刷新线程所需的最小时间(以秒为单位)。合法值介于 0 到 900(15 分钟)之间;如果没有指定值,则使用默认值 60(1 分钟)。如果值和 sasl.login.refresh.buffer.seconds 都会被忽略,如果它们的总和超过凭证的剩余生命周期。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.window.factor
type: double
Default: 0.8
Valid Values: [0.5,…,1.0]
Importance: low
登录刷新线程将休眠,直到达到指定的窗口因素为止(在达到凭证时),它将尝试刷新凭证。法律值介于 0.5(50%)到 1.0(100%)之间,如果未指定值,则使用默认值 0.8(80%)。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.window.jitter
type: double
Default: 0.05
Valid Values: [0.0,…,0.25]
Importance: low
相对于凭证的生命周期的最大随机 jitter 量,它们被添加到登录刷新线程睡眠时间。法律值介于 0 到 0.25(25%)之间(25%),如果没有指定值,则使用默认值 0.05(5%)。目前仅适用于 OAUTHBEARER。
sasl.login.retry.backoff.max.ms
type: long
Default: 10000 (10 秒)
Importance: low
(可选)登录尝试外部身份验证提供程序之间等待的最长这个值(可选)值(以毫秒为单位)。登录使用 sasl.login.retry.backoff.ms 设置的初始等待算法,并在尝试为 sasl.login.retry.backoff.max.ms 设置指定的最大等待长度之间加倍长度。目前仅适用于 OAUTHBEARER。
sasl.login.retry.backoff.ms
type: long
Default: 100
Importance: low
(可选)在登录尝试外部身份验证提供程序之间初始等待的值(可选)值(以毫秒为单位)。登录使用 sasl.login.retry.backoff.ms 设置的初始等待算法,并在尝试为 sasl.login.retry.backoff.max.ms 设置指定的最大等待长度之间加倍长度。目前仅适用于 OAUTHBEARER。
sasl.oauthbearer.clock.skew.seconds
type: int
Default: 30
Importance: low
(可选)值,允许 OAuth/OIDC 身份提供程序和代理的时间不同。
sasl.oauthbearer.expected.audience
type: list
Default: null
Importance: low
(可选)代理的逗号分隔设置,用于验证已经为其中一个预期对象发布 JWT。为标准的 OAuth "aud" 声明将检查 JWT,如果设置了这个值,代理将匹配 JWT 的"aud"声明的值,以查看是否存在准确的匹配项。如果没有匹配项,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.expected.issuer
type: string
Default: null
Importance: low
(可选)用于验证 JWT 是由预期签发者创建的代理的设置。将检查标准的 OAuth "iss" 声明,如果设置了此值,代理就与 JWT 的"iss"声明中的内容完全匹配。如果没有匹配项,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.jwks.endpoint.refresh.ms
type: long
Default: 3600000(1 小时)
Importance: low
(可选)代理在刷新其 JWKS(JSON Web Key Set)缓存时要等待的值,其中包含用于验证 JWT 签名的密钥。
sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms
type: long
Default: 10000 (10 秒)
Importance: low
(可选)尝试从外部身份验证提供程序检索 JWKS(JSON Web Key Set)之间的最大等待的值(以毫秒为单位)。JWKS 检索使用基于 sasl.oauthbearer.jwks.endpoint.retry.max.ms 设置的初始等待长度的指数级 backoff 算法,并在 sasl.oauthbearer.jwks.max.ms 设置指定的最大等待时间。
sasl.oauthbearer.jwks.endpoint.retry.backoff.ms
type: long
Default: 100
Importance: low
(可选)初始等待的值(JSON Web Key Set)从外部身份验证提供程序检索尝试。JWKS 检索使用基于 sasl.oauthbearer.jwks.endpoint.retry.max.ms 设置的初始等待长度的指数级 backoff 算法,并在 sasl.oauthbearer.jwks.max.ms 设置指定的最大等待时间。
sasl.oauthbearer.scope.claim.name
type: string
Default: scope
Importance: low
范围的 OAuth 声明通常命名为"scope",但此(可选)设置可以提供用于 JWT 有效负载声明中包含的范围的不同名称,如果 OAuth/OIDC 供应商将该声明使用不同的名称。
sasl.oauthbearer.sub.claim.name
type: string
Default: sub
Importance: low
主体的 OAuth 声明通常命名为"sub",但此(可选)设置可以为 JWT 有效负载声明中包含的主题提供不同的名称,如果 OAuth/OIDC 供应商将该声明使用不同的名称。
security.providers
type: string
Default: null
Importance: low
每次返回供应商实施安全算法的可配置创建者类列表。这些类应当实施
org.apache.kafka.common.security.auth.SecurityProviderCreator
接口。ssl.cipher.suites
type: list
Default: null
Importance: low
密码套件列表。这是验证、加密、MAC 和密钥交换算法的命名组合,用于通过 TLS 或 SSL 网络协议协商网络连接的安全设置。默认情况下,支持所有可用的密码套件。
ssl.endpoint.identification.algorithm
type: string
Default: https
Importance: low
使用服务器证书验证服务器主机名的端点标识算法。
ssl.engine.factory.class
type: class
Default: null
Importance: low
类型为 org.apache.kafka.common.security.auth.SslEngineFactory 的类来提供 SSLEngine 对象。默认值为 org.apache.kafka.common.ssl.DefaultSslEngineFactory。
ssl.keymanager.algorithm
type: string
Default: SunX509
Importance: low
密钥管理器工厂用于 SSL 连接的算法。默认值为 Java 虚拟机配置的主要管理器工厂算法。
ssl.secure.random.implementation
type: string
Default: null
Importance: low
用于 SSL 加密操作的 SecureRandom PRNG 实施。
ssl.trustmanager.algorithm
type: string
Default: PKIX
Importance: low
信任管理器工厂用于 SSL 连接的算法。默认值为 Java 虚拟机配置的信任管理器工厂算法。
第 4 章 制作者配置属性
key.serializer
type: class
Importance: high
序列化器类用于实现
org.apache.kafka.common.serialization.Serializer
接口的关键。value.serializer
type: class
Importance: high
序列化器类用于实现
org.apache.kafka.common.serialization.Serializer
接口的值。bootstrap.servers
type: list
Default: ""
Valid Values: non-null string
Importance: high
用于建立到 Kafka 集群的初始连接的主机/端口对列表。客户端将使用此处为 bootstrap 指定了哪些服务器的所有服务器 - 此列表仅影响用于发现全部服务器集的初始主机。此列表应当采用
host1:port1,host2:port2,…
的形式。由于这些服务器仅用于初始连接,以发现完整的群集成员身份(可能会动态更改),因此该列表不包含完整的服务器集合(尽管服务器停机时,您可能需要多个服务器)。buffer.memory
type: long
Default: 33554432
Valid Values: [0,…]
Importance: high
制作者可用于将等待发送到服务器的内存总量为:如果记录的发送速度快于可以发送到服务器,则制作者将阻断用于增加异常的
max.block.ms
。此设置应当与制作者将使用的总内存相对应,但不是硬绑定,因为生产者所使用的所有内存都用于缓冲区。有些额外的内存将用于压缩(如果启用压缩),以及维护动态请求。
compression.type
type: string
Default: none
Importance: high
由制作者生成的所有数据的压缩类型。默认为 none(即无压缩)。有效值为
无
、gzip
、snappy
、lz4
或zstd
。压缩是全批数据,因此批处理将影响压缩率(更批处理意味着更好的压缩方式)。retries
type: int
Default: 2147483647
Valid Values: [0,…,2147483647]
Importance: high
设置大于零的值将导致客户端重新发送发送失败的记录,并显示可能存在瞬时错误。请注意,如果客户端在收到错误时重新输入记录,这个重试不相同。如果在通过
delivery.timeout.ms
配置超时前,在重试次数之前,生成请求将失败,然后再成功确认前过期。用户通常应选择保留此配置未设置,而是使用delivery.timeout.ms
控制重试行为。启用幂等性需要此配置值大于 0。如果没有设置冲突的配置并且没有显式启用幂等性,则会禁用幂等性。
在将
enable.ideotence
设置为false
和max.in.flight.requests.per.connection
到 1 时允许重试,因为如果两个批处理发送到单个分区,如果两个批处理发送到单个分区,第一个失败并重试,第二个批处理中的记录可能会首先出现。ssl.key.password
type: password
Default: null
Importance: high
密钥存储文件或在"ssl.keystore.key"中指定的 PEM 密钥的密码。只有在配置了双向验证时,客户端才需要这样做。
ssl.keystore.certificate.chain
type: password
Default: null
Importance: high
使用由 'ssl.keystore.type" 指定的格式的证书链。默认 SSL 引擎工厂仅支持使用 X.509 证书列表的 PEM 格式。
ssl.keystore.key
type: password
Default: null
Importance: high
使用 'ssl.keystore.type" 指定的格式的私钥。默认 SSL 引擎工厂只支持使用 PKCS#8 密钥的 PEM 格式。如果密钥已加密,则必须使用 'ssl.key.password' 指定密钥密码。
ssl.keystore.location
type: string
Default: null
Importance: high
密钥存储文件的位置。对于客户端,这是可选的,可用于客户端双向身份验证。
ssl.keystore.password
type: password
Default: null
Importance: high
密钥存储文件的存储密码。对于客户端,这是可选的,只有在配置了 'ssl.keystore.location' 时才需要。PEM 格式不支持密钥存储密码。
ssl.truststore.certificates
type: password
Default: null
Importance: high
可信证书,格式为 'ssl.truststore.type'。默认 SSL 引擎工厂仅支持使用 X.509 证书使用 PEM 格式。
ssl.truststore.location
type: string
Default: null
Importance: high
信任存储文件的位置。
ssl.truststore.password
type: password
Default: null
Importance: high
信任存储文件的密码。如果没有设置密码,则仍然将使用配置的信任存储文件,但会禁用完整性检查。信任存储密码不支持 PEM 格式。
batch.size
type: int
Default: 16384
Valid Values: [0,…]
Importance: medium
当多个记录发送到同一分区时,生产者将尝试将记录组合成更少的请求。这有助于客户端和服务器上的性能。此配置以字节为单位控制默认的批处理大小。
不会尝试超过这个大小的批处理记录。
发送到代理的请求将包含多个批处理,每个分区对应一个可用数据。
较小的批处理大小会较小常用值,并且可能会降低吞吐量(一个批处理大小为零将完全禁用批处理)。非常大的批处理大小可能会更严重地使用内存,因为我们总是会在额外记录中为指定批处理大小分配一个缓冲区。
注:此设置提供要发送的批处理大小上限。如果我们的分区数量少于此字节,那么我们会"闲置者",等待更多记录显示更多记录。
此
linger.ms
设置默认为 0,这意味着我们会立即发送一个记录,即使是累积的批处理大小,该批处理大小位于这个batch.size
设置下。client.dns.lookup
type: string
Default: use_all_dns_ips
Valid Values: [use_all_dns_ips, resolve_canonical_bootstrap_servers_only]
Importance: medium
控制客户端如何使用 DNS 查找。如果设置为
use_all_dns_ips
,则按顺序连接到每个返回的 IP 地址,直到建立成功连接为止。断开连接后,将使用下一个 IP。使用完所有 IP 后,客户端会从主机名再次解析 IP(JVM 和 OS 缓存 DNS 名称查找)。如果设置为resolve_canonical_bootstrap_servers_only
,将每个 bootstrap 地址解析为规范名称列表。在 bootstrap 阶段后,这个行为与use_all_dns_ips
相同。client.id
type: string
Default: ""
Importance: medium
在发出请求时要传递给服务器的 id 字符串。这样做的目的是通过允许逻辑应用程序名称包含在服务器端请求日志记录中,跟踪超过 ip/port 的请求源。
connections.max.idle.ms
type: long
Default: 540000 (9 分钟)
Importance: medium
在此配置指定的毫秒数后关闭闲置连接。
delivery.timeout.ms
type: int
Default: 120000 (2 分钟)
Valid Values: [0,…]
Importance: medium
在调用
send()
返回后报告成功或失败的时间上有上限。这会限制在发送前会延迟记录的总时间、从代理等待确认的时间(如果预期),以及可重试发送失败的时间。如果遇到无法恢复的错误,则制作者可能会报告比这个配置更早的记录无法发送记录,或者将重试添加到已达到早期交付到期期限的批处理中。此配置的值应大于或等于request.timeout.ms
和linger.ms
的总和。linger.ms
type: long
Default: 0
Valid Values: [0,…]
Importance: medium
制作者将到达请求传输之间的所有记录分组到一个批处理请求中。通常,只有在记录到达的速度快于发送日志时,才会被加载。然而,在某些情况下,客户端可能希望降低请求数量,甚至处于中等负载。此设置通过添加少量人的延时(即不立即发送记录)来实现这个值,生产者将等待发送到给定延迟,以便发送其他记录以便可以组合使用。这可以被视为 TCP 中的 Nagle 的算法相似。此设置达到批处理延迟的上限:一旦我们获得
批处理.size
值得记录,无论无论此设置如何,任何设置都会立即发送。但是,如果我们小于这个分区,我们会为指定时间"闲置者"显示,等待更多记录显示。此设置默认为 0(即无延迟)。例如,设置linger.ms=5
将对发送的请求数量产生影响,但会将最多 5ms 延迟添加到没有负载时发送的记录。max.block.ms
type: long
Default: 60000(1 minute)
Valid Values: [0,…]
Importance: medium
该配置控制
KafkaProducer'send(),
,partitionsFor()
initTransactions()
,sendOffsetsToTransaction()
,commitTransaction()
和abortTransaction()
方法将阻断。对于send()
,此超时会绑定等待元数据获取和缓冲区分配的总时间(在用户提供的序列化器或分区器中不计于此超时)。对于partitionsFor()
,此超时会绑定等待元数据的时间(如果不可用)。事务相关的方法总是块,但当事务协调器无法发现或没有在超时时间中响应时可能会超时。max.request.size
type: int
Default: 1048576
Valid Values: [0,…]
Importance: medium
请求的最大值(以字节为单位)。此设置将限制制作者将在单个请求中发送的记录批处理数量,以避免发送大量请求。这也实际上是在最大不压缩记录批处理大小之上的上限。请注意,服务器在记录批处理大小(如果启用了压缩后)有自己的大写,这可能与此不同。
partitioner.class
type: class
Default: org.apache.kafka.clients.producer.internals.DefaultPartitioner
Importance: medium
用于决定在生成记录时要发送到的类。可用的选项有:
-
org.apache.kafka.clients.producer.internals.DefaultPartitioner
: default partitioner。此策略将尝试在批处理已满或linger.ms
之前放弃分区。它适用于策略: - 如果没有指定分区但存在密钥,则根据密钥的哈希选择一个分区
-
如果没有分区或密钥,请选择当批处理已满时更改的粘性分区,或者启用
linger.ms
。 -
org.apache.kafka.clients.producer.RoundRobinPartitioner
:这个分区策略是连续记录中的每个记录都会被发送到不同的分区(无论提供了 'key' 或 not ),直到我们耗尽分区并再次重新开始。注意:一个已知问题会在创建新批处理时产生不均匀的分发问题。请检查 KAFKA-9965 以了解更多详细信息。 org.apache.kafka.clients.producer.UniformStickyPartitioner
:这个分区策略将尝试放弃分区(无论是否提供了"key"或没有)直到批处理已满,或linger.ms
才会启动。通过实施
org.apache.kafka.clients.producer.Partitioner
接口,您可以插入自定义分区程序。
-
receive.buffer.bytes
type: int
Default: 32768(32 kibibytes)
Valid Values: [-1,…]
Importance: medium
在读取数据时要使用的 TCP 接收缓冲区(SO_RCVBUF)的大小。如果值为 -1,则使用 OS 默认。
request.timeout.ms
type: int
Default: 30000 (30 秒)
Valid Values: [0,…]
Importance: medium
配置控制客户端等待请求响应的最大时间。如果在超时前没有收到响应,客户端会在需要时重新发送请求,或者在重试耗尽时失败请求。这应该大于
replica.lag.max.ms
(代理配置),以减少因为不必要的制作者重试而出现消息重复的可能性。sasl.client.callback.handler.class
type: class
Default: null
Importance: medium
实现了 AuthenticateCallbackHandler 接口的 SASL 客户端回调处理器类的完全限定名称。
sasl.jaas.config
type: password
Default: null
Importance: medium
JAAS 登录上下文参数,用于 SASL 连接,其格式由 JAAS 配置文件使用。此处 介绍了 JAAS 配置文件格式。该值格式为:
loginModuleClass controlFlag(optionName=optionValue)*;
.对于代理,配置必须在小写中带有监听前缀和 SASL 机制名称前缀。例如: listener.name.sasl_ssl.scram-sha-256.sasl.jaas.config=com.example.ScramLoginModule;.sasl.kerberos.service.name
type: string
Default: null
Importance: medium
Kafka 运行的 Kerberos 主体名称。这可以在 Kafka 的 JAAS 配置或 Kafka 配置中定义。
sasl.login.callback.handler.class
type: class
Default: null
Importance: medium
实现了 AuthenticateCallbackHandler 接口的 SASL 登录回调处理器类的完全限定名称。对于代理,登录回调处理器配置必须以监听器前缀和 SASL 机制名称作为前缀作为前缀。例如: listener.name.sasl_ssl.scram-sha-256.sasl.login.callback.handler.class=com.example.CustomScramLoginCallbackHandler。
sasl.login.class
type: class
Default: null
Importance: medium
实施登录接口的类的完全限定名称。对于代理,登录配置必须在小写中带有监听前缀和 SASL 机制名称前缀。For example, listener.name.sasl_ssl.scram-sha-256.sasl.login.class=com.example.CustomScramLogin.
sasl.mechanism
type: string
Default: GSSAPI
Importance: medium
用于客户端连接的 SASL 机制。这可以是提供安全提供程序的任何机制。GSSAPI 是默认机制。
sasl.oauthbearer.jwks.endpoint.url
type: string
Default: null
Importance: medium
可以检索来自提供程序的 JWKS(JSON Web Key Set) 的 OAuth/OIDC 供应商 URL。URL 可以是基于 HTTP(S)或基于文件的。如果 URL 基于 HTTP(S),则 JWKS 数据将通过代理启动时配置的 URL 从 OAuth/OIDC 供应商中检索。所有 then-current 密钥将缓存在代理中,用于传入请求。如果为 JWT 收到身份验证请求,如果包含尚未在缓存中但未包括在缓存中的"kid"标头声明值,则将根据需要再次查询 JWKS 端点。但是,代理会在收到任何 JWT 请求前轮询每个 sasl.oauthbearer.jwks.endpoint.refresh.ms 毫秒的 URL 来刷新缓存。如果 URL 是基于文件,代理将从启动时配置的位置载入 JWKS 文件。如果 JWT 包含不在 JWKS 文件中的"kid"标头值,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.token.endpoint.url
type: string
Default: null
Importance: medium
OAuth/OIDC 身份提供程序的 URL。如果 URL 基于 HTTP(S),则代表签发者的令牌端点 URL,请求将基于 sasl.jaas.config 中的配置进行登录。如果 URL 基于文件,它将指定包含由 OAuth/OIDC 身份提供程序发布的访问令牌(在 JWT 序列化表单中)的文件用于授权。
security.protocol
type: string
Default: PLAINTEXT
Importance: medium
用于与代理通信的协议。有效值为:PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL.
send.buffer.bytes
type: int
Default: 131072(128 kibibytes)
Valid Values: [-1,…]
Importance: medium
在发送数据时要使用的 TCP 发送缓冲区(SO_SNDBUF)的大小。如果值为 -1,则使用 OS 默认。
socket.connection.setup.timeout.max.ms
type: long
Default: 30000 (30 秒)
Importance: medium
客户端等待套接字连接建立的最大时间。对于每个连续的连接失败,连接设置超时将按这个最大值增加。为避免连接 storms,则 0.2 的随机化工厂将应用到超时,从而导致计算值上面的 20% 到 20% 的随机范围。
socket.connection.setup.timeout.ms
type: long
Default: 10000 (10 秒)
Importance: medium
客户端等待套接字连接建立的时间。如果在超时前没有构建连接,客户端将关闭套接字频道。
ssl.enabled.protocols
type: list
Default: TLSv1.2,TLSv1.3
Importance: medium
为 SSL 连接启用的协议列表。使用 Java 11 或更新版本运行时,默认值为 'TLSv1.2,TLSv1.3',否则为 'TLSv1.2'。使用 Java 11 的默认值时,如果客户端和服务器同时支持 TLSv1.3,则客户端和服务器将首选 TLSv1.3,否则将回退到 TLSv1.2(假设这两个支持至少支持 TLSv1.2)。在大多数情况下,这个默认设置应该是正常的。另请参阅
ssl.protocol
的配置文档。ssl.keystore.type
type: string
Default: JKS
Importance: medium
密钥存储文件的文件格式。对于客户端,这是可选的。
ssl.protocol
type: string
Default: TLSv1.3
Importance: medium
用于生成 SSLContext 的 SSL 协议。使用 Java 11 或更新版本运行时,默认值为 'TLSv1.3',否则为 'TLSv1.2'。对于大多数用例来说,这个值应该非常正常。最近 JVM 中允许的值是 'TLSv1.2' 和 'TLSv1.3'。'TLS', 'TLSv1.1', 'SSL', 'SSLv2' 和 'SSLv3' 在以前的 JVM 中可能被支持,但由于已知的安全漏洞,不建议使用它们。使用这个配置和 'ssl.enabled.protocols' 的默认值,如果服务器不支持 'TLSv1.3',客户端将降级为 'TLSv1.2'。如果此配置被设置为 'TLSv1.2',则客户端不会使用 'TLSv1.3',即使它是 ssl.enabled.protocols 中的值之一,服务器仅支持 'TLSv1.3'。
ssl.provider
type: string
Default: null
Importance: medium
用于 SSL 连接的安全供应商名称。默认值是 JVM 的默认安全供应商。
ssl.truststore.type
type: string
Default: JKS
Importance: medium
信任存储文件的文件格式。
acks
type: string
Default: all
Valid Values: [all, -1, 0, 1]
Importance: low
致谢的数量要求在考虑请求完成前收到领导机。这控制发送的记录的持久性。允许以下设置:
-
acks=0
If 设为零,那么制作者将不等待服务器出现任何确认。记录将立即添加到套接字缓冲区并被视为已发送。无法保证服务器在这个示例中收到记录,并且重试
配置不会生效(因为客户端通常不知道任何失败)。每个记录返回的偏移将始终设置为-1
。 -
acks=1
意味着领导者会将记录写入其本地日志,但不会等待所有后续人员完全确认记录。在这种情况下,领导机在确认记录后立即失败,但随后再复制该记录将会丢失。 acks=all
意味着领导机将等待整套 in-sync 副本确认记录。这保证记录不会丢失,只要至少一个 in-sync 副本仍保持活动。这是最强大的可用保证。这等同于 acks=-1 设置。请注意,启用幂等性要求此配置值为"all"。如果没有设置冲突的配置并且没有显式启用幂等性,则会禁用幂等性。
-
enable.idempotence
type: boolean
Default: true
Importance: low
当设置为 'true' 时,制作者将确保在流中写入每个消息的一个副本。如果 'false',因为代理失败等,制作者会重试,可能会在流中写入重试的消息重复。请注意,启用 idempotence 需要
max.in.flight.requests.per.connection
的值小于或等于 5(可为任何允许值保留消息排序),重试
大于 0,acks
必须是 'all'。如果没有设置任何冲突配置,则默认启用 Idempence。如果没有设置冲突的配置并且没有显式启用幂等性,则会禁用幂等性。如果显式启用 idempence 并设置了冲突的配置,则会抛出
ConfigException
。interceptor.classes
type: list
Default: ""
Valid Values: non-null string
Importance: low
用作拦截器的类列表。通过实施
org.apache.kafka.clients.producer.ProducerInterceptor
接口,您可以在将制作者发布到 Kafka 集群前拦截(并可能损坏)制作者收到的记录。默认情况下,没有拦截器。max.in.flight.requests.per.connection
type: int
Default: 5
Valid Values: [1,…]
Importance: low
在阻塞之前,客户端将在单个连接上发送的最大未确认请求数。请注意,如果将此配置设定为大于 1
并启用.idempot
ence 被设置为 false,则因为重试(如果启用了重试,则会在失败后重新排序消息重新排序风险)。另外,启用幂等性要求此配置值小于或等于 5。如果没有设置冲突的配置并且没有显式启用幂等性,则会禁用幂等性。metadata.max.age.ms
type: long
Default: 300000 (5 分钟)
Valid Values: [0,…]
Importance: low
毫秒内的时间(以毫秒为单位),即使在我们未看到任何分区领导更改的情况下,也会强制更新元数据,以便主动发现任何新的代理或分区。
metadata.max.idle.ms
type: long
Default: 300000 (5 分钟)
Valid Values: [5000,…]
Importance: low
控制制作者将元数据缓存空闲的时长。如果最后生成了主题以超过元数据空闲持续时间所经过的时间,那么主题的元数据将被忘记,下一次访问将强制生成元数据获取请求。
metric.reporters
type: list
Default: ""
Valid Values: non-null string
Importance: low
用作指标报告器的类列表。实施
org.apache.kafka.common.metrics.MetricsReporter
接口允许在新指标创建通知的类中插入。JmxReporter 始终包含在内以注册 JMX 统计。metrics.num.samples
type: int
Default: 2
Valid Values: [1,…]
Importance: low
为计算指标维护的示例数量。
metrics.recording.level
type: string
Default: INFO
Valid Values: [INFO, DEBUG, SSP]
Importance: low
指标的最高记录级别。
metrics.sample.window.ms
type: long
Default: 30000 (30 秒)
Valid Values: [0,…]
Importance: low
计算指标示例的时间窗口。
reconnect.backoff.max.ms
type: long
Default: 1000(1秒)
Valid Values: [0,…]
Importance: low
重新连接到重复连接失败的代理时要等待的最大时间(毫秒)。如果提供,每个主机的 backoff 将为每个连续的连接失败指数增加,最高最高。在计算了 backoff 增长后,会添加 20% 的随机 jitter 来避免连接停滞。
reconnect.backoff.ms
type: long
Default: 50
Valid Values: [0,…]
Importance: low
在尝试重新连接给定主机前要等待的基本时间。这可避免在紧张循环中重复连接到主机。这个 backoff 适用于客户端到代理的所有连接尝试。
retry.backoff.ms
type: long
Default: 100
Valid Values: [0,…]
Importance: low
尝试重试失败的请求到给定主题分区前等待的时间。这可避免在某些故障情况下在紧密循环中重复发送请求。
sasl.kerberos.kinit.cmd
type: string
Default: /usr/bin/kinit
Importance: low
Kerberos kinit 命令路径.
sasl.kerberos.min.time.before.relogin
type: long
Default: 60000
Importance: low
刷新尝试之间的登录线程睡眠时间。
sasl.kerberos.ticket.renew.jitter
type: double
Default: 0.05
Importance: low
添加至续订时间的随机jitter 的百分比。
sasl.kerberos.ticket.renew.window.factor
type: double
Default: 0.8
Importance: low
登录线程将休眠,直到达到从最后刷新到 ticket 的过期时间,此时它将尝试续订票据。
sasl.login.connect.timeout.ms
type: int
Default: null
Importance: low
(可选)外部身份验证供应商连接超时的值(以毫秒为单位)。目前仅适用于 OAUTHBEARER。
sasl.login.read.timeout.ms
type: int
Default: null
Importance: low
(可选)外部身份验证提供程序读取超时的值(以毫秒为单位)。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.buffer.seconds
type: short
Default: 300
Valid Values: [0,…,3600]
Importance: low
在刷新凭证时要维护的凭证前的缓冲时间(以秒为单位)。如果刷新将超过缓冲区秒数的过期时间,则刷新将移动至可维护尽可能多的缓冲区时间。如果未指定值,则使用法定值在 0 到 3600(1 小时)之间;如果没有指定值,则使用默认值 300(5 分钟)。如果值和 sasl.login.refresh.min.period.seconds,如果它们的总和超过凭证的剩余生命周期,则都会被忽略。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.min.period.seconds
type: short
Default: 60
Valid Values: [0,…,900]
Importance: low
在刷新凭证前等待登录刷新线程所需的最小时间(以秒为单位)。合法值介于 0 到 900(15 分钟)之间;如果没有指定值,则使用默认值 60(1 分钟)。如果值和 sasl.login.refresh.buffer.seconds 都会被忽略,如果它们的总和超过凭证的剩余生命周期。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.window.factor
type: double
Default: 0.8
Valid Values: [0.5,…,1.0]
Importance: low
登录刷新线程将休眠,直到达到指定的窗口因素为止(在达到凭证时),它将尝试刷新凭证。法律值介于 0.5(50%)到 1.0(100%)之间,如果未指定值,则使用默认值 0.8(80%)。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.window.jitter
type: double
Default: 0.05
Valid Values: [0.0,…,0.25]
Importance: low
相对于凭证的生命周期的最大随机 jitter 量,它们被添加到登录刷新线程睡眠时间。法律值介于 0 到 0.25(25%)之间(25%),如果没有指定值,则使用默认值 0.05(5%)。目前仅适用于 OAUTHBEARER。
sasl.login.retry.backoff.max.ms
type: long
Default: 10000 (10 秒)
Importance: low
(可选)登录尝试外部身份验证提供程序之间等待的最长这个值(可选)值(以毫秒为单位)。登录使用 sasl.login.retry.backoff.ms 设置的初始等待算法,并在尝试为 sasl.login.retry.backoff.max.ms 设置指定的最大等待长度之间加倍长度。目前仅适用于 OAUTHBEARER。
sasl.login.retry.backoff.ms
type: long
Default: 100
Importance: low
(可选)在登录尝试外部身份验证提供程序之间初始等待的值(可选)值(以毫秒为单位)。登录使用 sasl.login.retry.backoff.ms 设置的初始等待算法,并在尝试为 sasl.login.retry.backoff.max.ms 设置指定的最大等待长度之间加倍长度。目前仅适用于 OAUTHBEARER。
sasl.oauthbearer.clock.skew.seconds
type: int
Default: 30
Importance: low
(可选)值,允许 OAuth/OIDC 身份提供程序和代理的时间不同。
sasl.oauthbearer.expected.audience
type: list
Default: null
Importance: low
(可选)代理的逗号分隔设置,用于验证已经为其中一个预期对象发布 JWT。为标准的 OAuth "aud" 声明将检查 JWT,如果设置了这个值,代理将匹配 JWT 的"aud"声明的值,以查看是否存在准确的匹配项。如果没有匹配项,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.expected.issuer
type: string
Default: null
Importance: low
(可选)用于验证 JWT 是由预期签发者创建的代理的设置。将检查标准的 OAuth "iss" 声明,如果设置了此值,代理就与 JWT 的"iss"声明中的内容完全匹配。如果没有匹配项,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.jwks.endpoint.refresh.ms
type: long
Default: 3600000(1 小时)
Importance: low
(可选)代理在刷新其 JWKS(JSON Web Key Set)缓存时要等待的值,其中包含用于验证 JWT 签名的密钥。
sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms
type: long
Default: 10000 (10 秒)
Importance: low
(可选)尝试从外部身份验证提供程序检索 JWKS(JSON Web Key Set)之间的最大等待的值(以毫秒为单位)。JWKS 检索使用基于 sasl.oauthbearer.jwks.endpoint.retry.max.ms 设置的初始等待长度的指数级 backoff 算法,并在 sasl.oauthbearer.jwks.max.ms 设置指定的最大等待时间。
sasl.oauthbearer.jwks.endpoint.retry.backoff.ms
type: long
Default: 100
Importance: low
(可选)初始等待的值(JSON Web Key Set)从外部身份验证提供程序检索尝试。JWKS 检索使用基于 sasl.oauthbearer.jwks.endpoint.retry.max.ms 设置的初始等待长度的指数级 backoff 算法,并在 sasl.oauthbearer.jwks.max.ms 设置指定的最大等待时间。
sasl.oauthbearer.scope.claim.name
type: string
Default: scope
Importance: low
范围的 OAuth 声明通常命名为"scope",但此(可选)设置可以提供用于 JWT 有效负载声明中包含的范围的不同名称,如果 OAuth/OIDC 供应商将该声明使用不同的名称。
sasl.oauthbearer.sub.claim.name
type: string
Default: sub
Importance: low
主体的 OAuth 声明通常命名为"sub",但此(可选)设置可以为 JWT 有效负载声明中包含的主题提供不同的名称,如果 OAuth/OIDC 供应商将该声明使用不同的名称。
security.providers
type: string
Default: null
Importance: low
每次返回供应商实施安全算法的可配置创建者类列表。这些类应当实施
org.apache.kafka.common.security.auth.SecurityProviderCreator
接口。ssl.cipher.suites
type: list
Default: null
Importance: low
密码套件列表。这是验证、加密、MAC 和密钥交换算法的命名组合,用于通过 TLS 或 SSL 网络协议协商网络连接的安全设置。默认情况下,支持所有可用的密码套件。
ssl.endpoint.identification.algorithm
type: string
Default: https
Importance: low
使用服务器证书验证服务器主机名的端点标识算法。
ssl.engine.factory.class
type: class
Default: null
Importance: low
类型为 org.apache.kafka.common.security.auth.SslEngineFactory 的类来提供 SSLEngine 对象。默认值为 org.apache.kafka.common.ssl.DefaultSslEngineFactory。
ssl.keymanager.algorithm
type: string
Default: SunX509
Importance: low
密钥管理器工厂用于 SSL 连接的算法。默认值为 Java 虚拟机配置的主要管理器工厂算法。
ssl.secure.random.implementation
type: string
Default: null
Importance: low
用于 SSL 加密操作的 SecureRandom PRNG 实施。
ssl.trustmanager.algorithm
type: string
Default: PKIX
Importance: low
信任管理器工厂用于 SSL 连接的算法。默认值为 Java 虚拟机配置的信任管理器工厂算法。
transaction.timeout.ms
type: int
Default: 60000 (1 分钟)
Importance: low
事务协调器在主动中止持续事务事务前,导致事务状态更新时间上限大于 broker 中的 transaction.max.timeout.ms 设置,请求将失败,并显示
InvalidTxnTimeoutException
错误。transactional.id
type: string
Default: null
Valid Values: non-empty string
Importance: low
用于事务发送的 TransactionalId。这启用了跨越多个制作者会话的可靠性语义,因为它允许客户端在启动任何新事务之前使用同一 TransactionalId 进行处理。如果未提供 TransactionalId,则生产者将限制为幂等发送。如果配置了 TransactionalId,则意味着
启用.ideot
ence。默认情况下不配置 TransactionId,这意味着无法使用事务。请注意,默认情况下,事务需要至少三个代理(生产的设置)集群;在开发时,您可以通过调整 broker 设置transaction.state.log.replication.factor
来更改此设置。
第 5 章 管理员客户端配置属性
bootstrap.servers
type: list
Importance: high
用于建立到 Kafka 集群的初始连接的主机/端口对列表。客户端将使用此处为 bootstrap 指定了哪些服务器的所有服务器 - 此列表仅影响用于发现全部服务器集的初始主机。此列表应当采用
host1:port1,host2:port2,…
的形式。由于这些服务器仅用于初始连接,以发现完整的群集成员身份(可能会动态更改),因此该列表不包含完整的服务器集合(尽管服务器停机时,您可能需要多个服务器)。ssl.key.password
type: password
Default: null
Importance: high
密钥存储文件或在"ssl.keystore.key"中指定的 PEM 密钥的密码。只有在配置了双向验证时,客户端才需要这样做。
ssl.keystore.certificate.chain
type: password
Default: null
Importance: high
使用由 'ssl.keystore.type" 指定的格式的证书链。默认 SSL 引擎工厂仅支持使用 X.509 证书列表的 PEM 格式。
ssl.keystore.key
type: password
Default: null
Importance: high
使用 'ssl.keystore.type" 指定的格式的私钥。默认 SSL 引擎工厂只支持使用 PKCS#8 密钥的 PEM 格式。如果密钥已加密,则必须使用 'ssl.key.password' 指定密钥密码。
ssl.keystore.location
type: string
Default: null
Importance: high
密钥存储文件的位置。对于客户端,这是可选的,可用于客户端双向身份验证。
ssl.keystore.password
type: password
Default: null
Importance: high
密钥存储文件的存储密码。对于客户端,这是可选的,只有在配置了 'ssl.keystore.location' 时才需要。PEM 格式不支持密钥存储密码。
ssl.truststore.certificates
type: password
Default: null
Importance: high
可信证书,格式为 'ssl.truststore.type'。默认 SSL 引擎工厂仅支持使用 X.509 证书使用 PEM 格式。
ssl.truststore.location
type: string
Default: null
Importance: high
信任存储文件的位置。
ssl.truststore.password
type: password
Default: null
Importance: high
信任存储文件的密码。如果没有设置密码,则仍然将使用配置的信任存储文件,但会禁用完整性检查。信任存储密码不支持 PEM 格式。
client.dns.lookup
type: string
Default: use_all_dns_ips
Valid Values: [use_all_dns_ips, resolve_canonical_bootstrap_servers_only]
Importance: medium
控制客户端如何使用 DNS 查找。如果设置为
use_all_dns_ips
,则按顺序连接到每个返回的 IP 地址,直到建立成功连接为止。断开连接后,将使用下一个 IP。使用完所有 IP 后,客户端会从主机名再次解析 IP(JVM 和 OS 缓存 DNS 名称查找)。如果设置为resolve_canonical_bootstrap_servers_only
,将每个 bootstrap 地址解析为规范名称列表。在 bootstrap 阶段后,这个行为与use_all_dns_ips
相同。client.id
type: string
Default: ""
Importance: medium
在发出请求时要传递给服务器的 id 字符串。这样做的目的是通过允许逻辑应用程序名称包含在服务器端请求日志记录中,跟踪超过 ip/port 的请求源。
connections.max.idle.ms
type: long
Default: 300000 (5 分钟)
Importance: medium
在此配置指定的毫秒数后关闭闲置连接。
default.api.timeout.ms
type: int
Default: 60000(1 minute)
Valid Values: [0,…]
Importance: medium
指定客户端 API 的超时(以毫秒为单位)。此配置用作所有没有指定
超时
参数的客户端操作的默认超时时间。receive.buffer.bytes
type: int
Default: 65536(64 kibibytes)
Valid Values: [-1,…]
Importance: medium
在读取数据时要使用的 TCP 接收缓冲区(SO_RCVBUF)的大小。如果值为 -1,则使用 OS 默认。
request.timeout.ms
type: int
Default: 30000 (30 秒)
Valid Values: [0,…]
Importance: medium
配置控制客户端等待请求响应的最大时间。如果在超时前没有收到响应,客户端会在需要时重新发送请求,或者在重试耗尽时失败请求。
sasl.client.callback.handler.class
type: class
Default: null
Importance: medium
实现了 AuthenticateCallbackHandler 接口的 SASL 客户端回调处理器类的完全限定名称。
sasl.jaas.config
type: password
Default: null
Importance: medium
JAAS 登录上下文参数,用于 SASL 连接,其格式由 JAAS 配置文件使用。此处 介绍了 JAAS 配置文件格式。该值格式为:
loginModuleClass controlFlag(optionName=optionValue)*;
.对于代理,配置必须在小写中带有监听前缀和 SASL 机制名称前缀。例如: listener.name.sasl_ssl.scram-sha-256.sasl.jaas.config=com.example.ScramLoginModule;.sasl.kerberos.service.name
type: string
Default: null
Importance: medium
Kafka 运行的 Kerberos 主体名称。这可以在 Kafka 的 JAAS 配置或 Kafka 配置中定义。
sasl.login.callback.handler.class
type: class
Default: null
Importance: medium
实现了 AuthenticateCallbackHandler 接口的 SASL 登录回调处理器类的完全限定名称。对于代理,登录回调处理器配置必须以监听器前缀和 SASL 机制名称作为前缀作为前缀。例如: listener.name.sasl_ssl.scram-sha-256.sasl.login.callback.handler.class=com.example.CustomScramLoginCallbackHandler。
sasl.login.class
type: class
Default: null
Importance: medium
实施登录接口的类的完全限定名称。对于代理,登录配置必须在小写中带有监听前缀和 SASL 机制名称前缀。For example, listener.name.sasl_ssl.scram-sha-256.sasl.login.class=com.example.CustomScramLogin.
sasl.mechanism
type: string
Default: GSSAPI
Importance: medium
用于客户端连接的 SASL 机制。这可以是提供安全提供程序的任何机制。GSSAPI 是默认机制。
sasl.oauthbearer.jwks.endpoint.url
type: string
Default: null
Importance: medium
可以检索来自提供程序的 JWKS(JSON Web Key Set) 的 OAuth/OIDC 供应商 URL。URL 可以是基于 HTTP(S)或基于文件的。如果 URL 基于 HTTP(S),则 JWKS 数据将通过代理启动时配置的 URL 从 OAuth/OIDC 供应商中检索。所有 then-current 密钥将缓存在代理中,用于传入请求。如果为 JWT 收到身份验证请求,如果包含尚未在缓存中但未包括在缓存中的"kid"标头声明值,则将根据需要再次查询 JWKS 端点。但是,代理会在收到任何 JWT 请求前轮询每个 sasl.oauthbearer.jwks.endpoint.refresh.ms 毫秒的 URL 来刷新缓存。如果 URL 是基于文件,代理将从启动时配置的位置载入 JWKS 文件。如果 JWT 包含不在 JWKS 文件中的"kid"标头值,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.token.endpoint.url
type: string
Default: null
Importance: medium
OAuth/OIDC 身份提供程序的 URL。如果 URL 基于 HTTP(S),则代表签发者的令牌端点 URL,请求将基于 sasl.jaas.config 中的配置进行登录。如果 URL 基于文件,它将指定包含由 OAuth/OIDC 身份提供程序发布的访问令牌(在 JWT 序列化表单中)的文件用于授权。
security.protocol
type: string
Default: PLAINTEXT
Importance: medium
用于与代理通信的协议。有效值为:PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL.
send.buffer.bytes
type: int
Default: 131072(128 kibibytes)
Valid Values: [-1,…]
Importance: medium
在发送数据时要使用的 TCP 发送缓冲区(SO_SNDBUF)的大小。如果值为 -1,则使用 OS 默认。
socket.connection.setup.timeout.max.ms
type: long
Default: 30000 (30 秒)
Importance: medium
客户端等待套接字连接建立的最大时间。对于每个连续的连接失败,连接设置超时将按这个最大值增加。为避免连接 storms,则 0.2 的随机化工厂将应用到超时,从而导致计算值上面的 20% 到 20% 的随机范围。
socket.connection.setup.timeout.ms
type: long
Default: 10000 (10 秒)
Importance: medium
客户端等待套接字连接建立的时间。如果在超时前没有构建连接,客户端将关闭套接字频道。
ssl.enabled.protocols
type: list
Default: TLSv1.2,TLSv1.3
Importance: medium
为 SSL 连接启用的协议列表。使用 Java 11 或更新版本运行时,默认值为 'TLSv1.2,TLSv1.3',否则为 'TLSv1.2'。使用 Java 11 的默认值时,如果客户端和服务器同时支持 TLSv1.3,则客户端和服务器将首选 TLSv1.3,否则将回退到 TLSv1.2(假设这两个支持至少支持 TLSv1.2)。在大多数情况下,这个默认设置应该是正常的。另请参阅
ssl.protocol
的配置文档。ssl.keystore.type
type: string
Default: JKS
Importance: medium
密钥存储文件的文件格式。对于客户端,这是可选的。
ssl.protocol
type: string
Default: TLSv1.3
Importance: medium
用于生成 SSLContext 的 SSL 协议。使用 Java 11 或更新版本运行时,默认值为 'TLSv1.3',否则为 'TLSv1.2'。对于大多数用例来说,这个值应该非常正常。最近 JVM 中允许的值是 'TLSv1.2' 和 'TLSv1.3'。'TLS', 'TLSv1.1', 'SSL', 'SSLv2' 和 'SSLv3' 在以前的 JVM 中可能被支持,但由于已知的安全漏洞,不建议使用它们。使用这个配置和 'ssl.enabled.protocols' 的默认值,如果服务器不支持 'TLSv1.3',客户端将降级为 'TLSv1.2'。如果此配置被设置为 'TLSv1.2',则客户端不会使用 'TLSv1.3',即使它是 ssl.enabled.protocols 中的值之一,服务器仅支持 'TLSv1.3'。
ssl.provider
type: string
Default: null
Importance: medium
用于 SSL 连接的安全供应商名称。默认值是 JVM 的默认安全供应商。
ssl.truststore.type
type: string
Default: JKS
Importance: medium
信任存储文件的文件格式。
metadata.max.age.ms
type: long
Default: 300000 (5 分钟)
Valid Values: [0,…]
Importance: low
毫秒内的时间(以毫秒为单位),即使在我们未看到任何分区领导更改的情况下,也会强制更新元数据,以便主动发现任何新的代理或分区。
metric.reporters
type: list
Default: ""
Importance: low
用作指标报告器的类列表。实施
org.apache.kafka.common.metrics.MetricsReporter
接口允许在新指标创建通知的类中插入。JmxReporter 始终包含在内以注册 JMX 统计。metrics.num.samples
type: int
Default: 2
Valid Values: [1,…]
Importance: low
为计算指标维护的示例数量。
metrics.recording.level
type: string
Default: INFO
Valid Values: [INFO, DEBUG, SSP]
Importance: low
指标的最高记录级别。
metrics.sample.window.ms
type: long
Default: 30000 (30 秒)
Valid Values: [0,…]
Importance: low
计算指标示例的时间窗口。
reconnect.backoff.max.ms
type: long
Default: 1000(1秒)
Valid Values: [0,…]
Importance: low
重新连接到重复连接失败的代理时要等待的最大时间(毫秒)。如果提供,每个主机的 backoff 将为每个连续的连接失败指数增加,最高最高。在计算了 backoff 增长后,会添加 20% 的随机 jitter 来避免连接停滞。
reconnect.backoff.ms
type: long
Default: 50
Valid Values: [0,…]
Importance: low
在尝试重新连接给定主机前要等待的基本时间。这可避免在紧张循环中重复连接到主机。这个 backoff 适用于客户端到代理的所有连接尝试。
retries
type: int
Default: 2147483647
Valid Values: [0,…,2147483647]
Importance: low
设置大于零的值将导致客户端重新发送任何失败的请求,并出现可能存在瞬时错误的请求。建议将值设置为零或
MAX_VALUE
,并使用对应的超时参数来控制客户端应重试请求的时长。retry.backoff.ms
type: long
Default: 100
Valid Values: [0,…]
Importance: low
尝试重试失败的请求前等待的时间。这可避免在某些故障情况下在紧密循环中重复发送请求。
sasl.kerberos.kinit.cmd
type: string
Default: /usr/bin/kinit
Importance: low
Kerberos kinit 命令路径.
sasl.kerberos.min.time.before.relogin
type: long
Default: 60000
Importance: low
刷新尝试之间的登录线程睡眠时间。
sasl.kerberos.ticket.renew.jitter
type: double
Default: 0.05
Importance: low
添加至续订时间的随机jitter 的百分比。
sasl.kerberos.ticket.renew.window.factor
type: double
Default: 0.8
Importance: low
登录线程将休眠,直到达到从最后刷新到 ticket 的过期时间,此时它将尝试续订票据。
sasl.login.connect.timeout.ms
type: int
Default: null
Importance: low
(可选)外部身份验证供应商连接超时的值(以毫秒为单位)。目前仅适用于 OAUTHBEARER。
sasl.login.read.timeout.ms
type: int
Default: null
Importance: low
(可选)外部身份验证提供程序读取超时的值(以毫秒为单位)。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.buffer.seconds
type: short
Default: 300
Valid Values: [0,…,3600]
Importance: low
在刷新凭证时要维护的凭证前的缓冲时间(以秒为单位)。如果刷新将超过缓冲区秒数的过期时间,则刷新将移动至可维护尽可能多的缓冲区时间。如果未指定值,则使用法定值在 0 到 3600(1 小时)之间;如果没有指定值,则使用默认值 300(5 分钟)。如果值和 sasl.login.refresh.min.period.seconds,如果它们的总和超过凭证的剩余生命周期,则都会被忽略。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.min.period.seconds
type: short
Default: 60
Valid Values: [0,…,900]
Importance: low
在刷新凭证前等待登录刷新线程所需的最小时间(以秒为单位)。合法值介于 0 到 900(15 分钟)之间;如果没有指定值,则使用默认值 60(1 分钟)。如果值和 sasl.login.refresh.buffer.seconds 都会被忽略,如果它们的总和超过凭证的剩余生命周期。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.window.factor
type: double
Default: 0.8
Valid Values: [0.5,…,1.0]
Importance: low
登录刷新线程将休眠,直到达到指定的窗口因素为止(在达到凭证时),它将尝试刷新凭证。法律值介于 0.5(50%)到 1.0(100%)之间,如果未指定值,则使用默认值 0.8(80%)。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.window.jitter
type: double
Default: 0.05
Valid Values: [0.0,…,0.25]
Importance: low
相对于凭证的生命周期的最大随机 jitter 量,它们被添加到登录刷新线程睡眠时间。法律值介于 0 到 0.25(25%)之间(25%),如果没有指定值,则使用默认值 0.05(5%)。目前仅适用于 OAUTHBEARER。
sasl.login.retry.backoff.max.ms
type: long
Default: 10000 (10 秒)
Importance: low
(可选)登录尝试外部身份验证提供程序之间等待的最长这个值(可选)值(以毫秒为单位)。登录使用 sasl.login.retry.backoff.ms 设置的初始等待算法,并在尝试为 sasl.login.retry.backoff.max.ms 设置指定的最大等待长度之间加倍长度。目前仅适用于 OAUTHBEARER。
sasl.login.retry.backoff.ms
type: long
Default: 100
Importance: low
(可选)在登录尝试外部身份验证提供程序之间初始等待的值(可选)值(以毫秒为单位)。登录使用 sasl.login.retry.backoff.ms 设置的初始等待算法,并在尝试为 sasl.login.retry.backoff.max.ms 设置指定的最大等待长度之间加倍长度。目前仅适用于 OAUTHBEARER。
sasl.oauthbearer.clock.skew.seconds
type: int
Default: 30
Importance: low
(可选)值,允许 OAuth/OIDC 身份提供程序和代理的时间不同。
sasl.oauthbearer.expected.audience
type: list
Default: null
Importance: low
(可选)代理的逗号分隔设置,用于验证已经为其中一个预期对象发布 JWT。为标准的 OAuth "aud" 声明将检查 JWT,如果设置了这个值,代理将匹配 JWT 的"aud"声明的值,以查看是否存在准确的匹配项。如果没有匹配项,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.expected.issuer
type: string
Default: null
Importance: low
(可选)用于验证 JWT 是由预期签发者创建的代理的设置。将检查标准的 OAuth "iss" 声明,如果设置了此值,代理就与 JWT 的"iss"声明中的内容完全匹配。如果没有匹配项,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.jwks.endpoint.refresh.ms
type: long
Default: 3600000(1 小时)
Importance: low
(可选)代理在刷新其 JWKS(JSON Web Key Set)缓存时要等待的值,其中包含用于验证 JWT 签名的密钥。
sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms
type: long
Default: 10000 (10 秒)
Importance: low
(可选)尝试从外部身份验证提供程序检索 JWKS(JSON Web Key Set)之间的最大等待的值(以毫秒为单位)。JWKS 检索使用基于 sasl.oauthbearer.jwks.endpoint.retry.max.ms 设置的初始等待长度的指数级 backoff 算法,并在 sasl.oauthbearer.jwks.max.ms 设置指定的最大等待时间。
sasl.oauthbearer.jwks.endpoint.retry.backoff.ms
type: long
Default: 100
Importance: low
(可选)初始等待的值(JSON Web Key Set)从外部身份验证提供程序检索尝试。JWKS 检索使用基于 sasl.oauthbearer.jwks.endpoint.retry.max.ms 设置的初始等待长度的指数级 backoff 算法,并在 sasl.oauthbearer.jwks.max.ms 设置指定的最大等待时间。
sasl.oauthbearer.scope.claim.name
type: string
Default: scope
Importance: low
范围的 OAuth 声明通常命名为"scope",但此(可选)设置可以提供用于 JWT 有效负载声明中包含的范围的不同名称,如果 OAuth/OIDC 供应商将该声明使用不同的名称。
sasl.oauthbearer.sub.claim.name
type: string
Default: sub
Importance: low
主体的 OAuth 声明通常命名为"sub",但此(可选)设置可以为 JWT 有效负载声明中包含的主题提供不同的名称,如果 OAuth/OIDC 供应商将该声明使用不同的名称。
security.providers
type: string
Default: null
Importance: low
每次返回供应商实施安全算法的可配置创建者类列表。这些类应当实施
org.apache.kafka.common.security.auth.SecurityProviderCreator
接口。ssl.cipher.suites
type: list
Default: null
Importance: low
密码套件列表。这是验证、加密、MAC 和密钥交换算法的命名组合,用于通过 TLS 或 SSL 网络协议协商网络连接的安全设置。默认情况下,支持所有可用的密码套件。
ssl.endpoint.identification.algorithm
type: string
Default: https
Importance: low
使用服务器证书验证服务器主机名的端点标识算法。
ssl.engine.factory.class
type: class
Default: null
Importance: low
类型为 org.apache.kafka.common.security.auth.SslEngineFactory 的类来提供 SSLEngine 对象。默认值为 org.apache.kafka.common.ssl.DefaultSslEngineFactory。
ssl.keymanager.algorithm
type: string
Default: SunX509
Importance: low
密钥管理器工厂用于 SSL 连接的算法。默认值为 Java 虚拟机配置的主要管理器工厂算法。
ssl.secure.random.implementation
type: string
Default: null
Importance: low
用于 SSL 加密操作的 SecureRandom PRNG 实施。
ssl.trustmanager.algorithm
type: string
Default: PKIX
Importance: low
信任管理器工厂用于 SSL 连接的算法。默认值为 Java 虚拟机配置的信任管理器工厂算法。
第 6 章 Kafka Connect 配置属性
config.storage.topic
type: string
Importance: high
存储连接器配置的 Kafka 主题的名称。
group.id
type: string
Importance: high
此 worker 所属的连接集群组的唯一字符串。
key.converter
type: class
Importance: high
用于在 Kafka Connect 格式和写入 Kafka 的序列化格式间转换器类。这控制写入 Kafka 或从 Kafka 读取的消息中的密钥格式,因为这独立于连接器,任何连接器可以使用任何序列化格式。常见格式示例包括 JSON 和 Avro。
offset.storage.topic
type: string
Importance: high
存储连接器偏移的 Kafka 主题的名称。
status.storage.topic
type: string
Importance: high
存储连接器和任务状态的 Kafka 主题的名称。
value.converter
type: class
Importance: high
用于在 Kafka Connect 格式和写入 Kafka 的序列化格式间转换器类。这控制写入或从 Kafka 读取的消息中的值格式,因为这独立于连接器,可以使用任何序列化格式。常见格式示例包括 JSON 和 Avro。
bootstrap.servers
type: list
Default: localhost:9092
Importance: high
用于建立到 Kafka 集群的初始连接的主机/端口对列表。客户端将使用此处为 bootstrap 指定了哪些服务器的所有服务器 - 此列表仅影响用于发现全部服务器集的初始主机。此列表应当采用
host1:port1,host2:port2,…
的形式。由于这些服务器仅用于初始连接,以发现完整的群集成员身份(可能会动态更改),因此该列表不包含完整的服务器集合(尽管服务器停机时,您可能需要多个服务器)。heartbeat.interval.ms
type: int
Default: 3000 (3 秒)
Importance: high
使用 Kafka 的组管理工具时,心跳到组协调器之间的预期时间。heartbeat 用于确保 worker 的会话保持活跃状态,并在新成员加入或离开组时进行重新平衡。该值必须小于
session.timeout.ms
,但通常设置的值通常不应超过 1/3。它可以被调整,以便控制正常重新平衡的预期时间。rebalance.timeout.ms
type: int
Default: 60000 (1 分钟)
Importance: high
在重新平衡开始后,每个 worker 能够加入该组的最大允许时间。这基本上是针对所有任务刷新任何待处理数据和提交偏移所需的时间。如果超过超时时间,则 worker 将从组中删除,这会导致偏移提交失败。
session.timeout.ms
type: int
Default: 10000 (10 秒)
Importance: high
用于检测 worker 故障的超时。worker 发送定期心跳,以指示其到代理的存活度。如果在这个会话超时前代理没有收到 heartbeat,则代理会从组中删除 worker 并启动重新平衡。请注意,该值必须在代理配置中由
group.min.session.timeout.ms
和group.max.session.timeout.ms
配置中的允许范围内配置。ssl.key.password
type: password
Default: null
Importance: high
密钥存储文件或在"ssl.keystore.key"中指定的 PEM 密钥的密码。只有在配置了双向验证时,客户端才需要这样做。
ssl.keystore.certificate.chain
type: password
Default: null
Importance: high
使用由 'ssl.keystore.type" 指定的格式的证书链。默认 SSL 引擎工厂仅支持使用 X.509 证书列表的 PEM 格式。
ssl.keystore.key
type: password
Default: null
Importance: high
使用 'ssl.keystore.type" 指定的格式的私钥。默认 SSL 引擎工厂只支持使用 PKCS#8 密钥的 PEM 格式。如果密钥已加密,则必须使用 'ssl.key.password' 指定密钥密码。
ssl.keystore.location
type: string
Default: null
Importance: high
密钥存储文件的位置。对于客户端,这是可选的,可用于客户端双向身份验证。
ssl.keystore.password
type: password
Default: null
Importance: high
密钥存储文件的存储密码。对于客户端,这是可选的,只有在配置了 'ssl.keystore.location' 时才需要。PEM 格式不支持密钥存储密码。
ssl.truststore.certificates
type: password
Default: null
Importance: high
可信证书,格式为 'ssl.truststore.type'。默认 SSL 引擎工厂仅支持使用 X.509 证书使用 PEM 格式。
ssl.truststore.location
type: string
Default: null
Importance: high
信任存储文件的位置。
ssl.truststore.password
type: password
Default: null
Importance: high
信任存储文件的密码。如果没有设置密码,则仍然将使用配置的信任存储文件,但会禁用完整性检查。信任存储密码不支持 PEM 格式。
client.dns.lookup
type: string
Default: use_all_dns_ips
Valid Values: [use_all_dns_ips, resolve_canonical_bootstrap_servers_only]
Importance: medium
控制客户端如何使用 DNS 查找。如果设置为
use_all_dns_ips
,则按顺序连接到每个返回的 IP 地址,直到建立成功连接为止。断开连接后,将使用下一个 IP。使用完所有 IP 后,客户端会从主机名再次解析 IP(JVM 和 OS 缓存 DNS 名称查找)。如果设置为resolve_canonical_bootstrap_servers_only
,将每个 bootstrap 地址解析为规范名称列表。在 bootstrap 阶段后,这个行为与use_all_dns_ips
相同。connections.max.idle.ms
type: long
Default: 540000 (9 分钟)
Importance: medium
在此配置指定的毫秒数后关闭闲置连接。
connector.client.config.override.policy
type: string
Default: All
Importance: medium
ConnectorClientConfigOverridePolicy
的类名称或别名。定义连接器可覆盖哪些客户端配置。默认实施是All
,这意味着连接器配置可覆盖所有客户端属性。框架中的其他可能策略包括None
(禁止从覆盖客户端属性)的连接器,以及允许连接器仅覆盖客户端主体。receive.buffer.bytes
type: int
Default: 32768(32 kibibytes)
Valid Values: [0,…]
Importance: medium
在读取数据时要使用的 TCP 接收缓冲区(SO_RCVBUF)的大小。如果值为 -1,则使用 OS 默认。
request.timeout.ms
type: int
Default: 40000 (40 秒)
Valid Values: [0,…]
Importance: medium
配置控制客户端等待请求响应的最大时间。如果在超时前没有收到响应,客户端会在需要时重新发送请求,或者在重试耗尽时失败请求。
sasl.client.callback.handler.class
type: class
Default: null
Importance: medium
实现了 AuthenticateCallbackHandler 接口的 SASL 客户端回调处理器类的完全限定名称。
sasl.jaas.config
type: password
Default: null
Importance: medium
JAAS 登录上下文参数,用于 SASL 连接,其格式由 JAAS 配置文件使用。此处 介绍了 JAAS 配置文件格式。该值格式为:
loginModuleClass controlFlag(optionName=optionValue)*;
.对于代理,配置必须在小写中带有监听前缀和 SASL 机制名称前缀。例如: listener.name.sasl_ssl.scram-sha-256.sasl.jaas.config=com.example.ScramLoginModule;.sasl.kerberos.service.name
type: string
Default: null
Importance: medium
Kafka 运行的 Kerberos 主体名称。这可以在 Kafka 的 JAAS 配置或 Kafka 配置中定义。
sasl.login.callback.handler.class
type: class
Default: null
Importance: medium
实现了 AuthenticateCallbackHandler 接口的 SASL 登录回调处理器类的完全限定名称。对于代理,登录回调处理器配置必须以监听器前缀和 SASL 机制名称作为前缀作为前缀。例如: listener.name.sasl_ssl.scram-sha-256.sasl.login.callback.handler.class=com.example.CustomScramLoginCallbackHandler。
sasl.login.class
type: class
Default: null
Importance: medium
实施登录接口的类的完全限定名称。对于代理,登录配置必须在小写中带有监听前缀和 SASL 机制名称前缀。For example, listener.name.sasl_ssl.scram-sha-256.sasl.login.class=com.example.CustomScramLogin.
sasl.mechanism
type: string
Default: GSSAPI
Importance: medium
用于客户端连接的 SASL 机制。这可以是提供安全提供程序的任何机制。GSSAPI 是默认机制。
sasl.oauthbearer.jwks.endpoint.url
type: string
Default: null
Importance: medium
可以检索来自提供程序的 JWKS(JSON Web Key Set) 的 OAuth/OIDC 供应商 URL。URL 可以是基于 HTTP(S)或基于文件的。如果 URL 基于 HTTP(S),则 JWKS 数据将通过代理启动时配置的 URL 从 OAuth/OIDC 供应商中检索。所有 then-current 密钥将缓存在代理中,用于传入请求。如果为 JWT 收到身份验证请求,如果包含尚未在缓存中但未包括在缓存中的"kid"标头声明值,则将根据需要再次查询 JWKS 端点。但是,代理会在收到任何 JWT 请求前轮询每个 sasl.oauthbearer.jwks.endpoint.refresh.ms 毫秒的 URL 来刷新缓存。如果 URL 是基于文件,代理将从启动时配置的位置载入 JWKS 文件。如果 JWT 包含不在 JWKS 文件中的"kid"标头值,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.token.endpoint.url
type: string
Default: null
Importance: medium
OAuth/OIDC 身份提供程序的 URL。如果 URL 基于 HTTP(S),则代表签发者的令牌端点 URL,请求将基于 sasl.jaas.config 中的配置进行登录。如果 URL 基于文件,它将指定包含由 OAuth/OIDC 身份提供程序发布的访问令牌(在 JWT 序列化表单中)的文件用于授权。
security.protocol
type: string
Default: PLAINTEXT
Importance: medium
用于与代理通信的协议。有效值为:PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL.
send.buffer.bytes
type: int
Default: 131072(128 kibibytes)
Valid Values: [0,…]
Importance: medium
在发送数据时要使用的 TCP 发送缓冲区(SO_SNDBUF)的大小。如果值为 -1,则使用 OS 默认。
ssl.enabled.protocols
type: list
Default: TLSv1.2,TLSv1.3
Importance: medium
为 SSL 连接启用的协议列表。使用 Java 11 或更新版本运行时,默认值为 'TLSv1.2,TLSv1.3',否则为 'TLSv1.2'。使用 Java 11 的默认值时,如果客户端和服务器同时支持 TLSv1.3,则客户端和服务器将首选 TLSv1.3,否则将回退到 TLSv1.2(假设这两个支持至少支持 TLSv1.2)。在大多数情况下,这个默认设置应该是正常的。另请参阅
ssl.protocol
的配置文档。ssl.keystore.type
type: string
Default: JKS
Importance: medium
密钥存储文件的文件格式。对于客户端,这是可选的。
ssl.protocol
type: string
Default: TLSv1.3
Importance: medium
用于生成 SSLContext 的 SSL 协议。使用 Java 11 或更新版本运行时,默认值为 'TLSv1.3',否则为 'TLSv1.2'。对于大多数用例来说,这个值应该非常正常。最近 JVM 中允许的值是 'TLSv1.2' 和 'TLSv1.3'。'TLS', 'TLSv1.1', 'SSL', 'SSLv2' 和 'SSLv3' 在以前的 JVM 中可能被支持,但由于已知的安全漏洞,不建议使用它们。使用这个配置和 'ssl.enabled.protocols' 的默认值,如果服务器不支持 'TLSv1.3',客户端将降级为 'TLSv1.2'。如果此配置被设置为 'TLSv1.2',则客户端不会使用 'TLSv1.3',即使它是 ssl.enabled.protocols 中的值之一,服务器仅支持 'TLSv1.3'。
ssl.provider
type: string
Default: null
Importance: medium
用于 SSL 连接的安全供应商名称。默认值是 JVM 的默认安全供应商。
ssl.truststore.type
type: string
Default: JKS
Importance: medium
信任存储文件的文件格式。
worker.sync.timeout.ms
type: int
Default: 3000 (3 秒)
Importance: medium
当 worker 与其他 worker 保持同步且需要重新同步配置时,请等待这段时间,让组离开组,并在重新加入前等待 backoff 周期。
worker.unsync.backoff.ms
type: int
Default: 300000 (5 分钟)
Importance: medium
当 worker 与其他 worker 保持同步且无法在 worker.sync.timeout.ms 中捕获,请在重新加入前让 Connect 集群保留这个时间。
access.control.allow.methods
type: string
Default: ""
Importance: low
通过设置 Access-Control-Allow-Methods 标头来设置支持跨源请求的方法。Access-Control-Allow-Methods 标头的默认值允许 GET、POST 和 HEAD 的跨源请求。
access.control.allow.origin
type: string
Default: ""
Importance: low
为 REST API 请求设置 Access-Control-Allow-Origin 标头的值。要启用跨原始访问,请将其设置为允许访问 API 的应用程序的域,或 '*' 允许从任何域访问。默认值仅允许从 REST API 域进行访问。
admin.listeners
type: list
Default: null
Valid Values: comma-separated URL 列表,例如: http://localhost:8080,https://localhost:8443。
重要性: low
以逗号分隔的 URI 列表,即 Admin REST API 将侦听。支持的协议有 HTTP 和 HTTPS。空白或空白字符串将禁用此功能。默认的行为是使用常规的监听程序(通过 'listeners' 属性指定)。
client.id
type: string
Default: ""
Importance: low
在发出请求时要传递给服务器的 id 字符串。这样做的目的是通过允许逻辑应用程序名称包含在服务器端请求日志记录中,跟踪超过 ip/port 的请求源。
config.providers
type: list
Default: ""
Importance: low
ConfigProvider
类的逗号分隔名称,以指定的顺序加载和使用。通过实施接口ConfigProvider
,您可以替换连接器配置中的变量引用,如用于外部化的 secret。config.storage.replication.factor
Type: short
Default: 3
Valid Values: Positive number 不大于 Kafka 集群中的代理数量,或 -1 使用代理的默认
导入级别 : 低
创建配置存储主题时使用的复制因素。
connect.protocol
type: string
Default: sessioned
Valid Values: [eager, compatible, sessioned]
Importance: low
Kafka 连接协议的兼容性模式。
header.converter
type: class
Default: org.apache.kafka.connect.storage.SimpleHeaderConverter
Importance: low
用于在 Kafka Connect 格式和写入 Kafka 的序列化格式间转换的HeaderConverter 类。这控制从 Kafka 写入或读取的消息中标头值的格式,因为这独立于连接器,任何连接器可以使用任何序列化格式。常见格式示例包括 JSON 和 Avro。默认情况下,SimpleHeaderConverter 用来序列化标头值到字符串,并通过推断 schema 来反序列化它们。
inter.worker.key.generation.algorithm
type: string
Default: HmacSHA256
Valid Values: worker JVM
Importance: low 支持的任何 KeyGenerator 算法
用于生成内部请求密钥的算法。
inter.worker.key.size
type: int
Default: null
Importance: low
用于签名内部请求的密钥大小(以位为单位)。如果 null,将使用密钥生成算法的默认密钥大小。
inter.worker.key.ttl.ms
type: int
Default: 3600000(1 hour)
Valid Values: [0,…,2147483647]
Importance: low
所生成的会话键的 TTL,用于内部请求验证(以毫秒为单位)。
inter.worker.signature.algorithm
type: string
Default: HmacSHA256
Valid Values: worker JVM
Importance: low 支持的所有 MAC 算法: low
用于签署内部请求的算法。
inter.worker.verification.algorithms
type: list
Default: HmacSHA256
Valid Values: 一个由 worker JVM
Importance 支持的一个或多个 MAC 算法的列表 : low
用于验证内部请求的允许算法列表。
监听程序
type: list
Default: http://:8083
Valid Values: comma-separated URL 列表,例如: http://localhost:8080,https://localhost:8443。
重要性: low
REST API 将侦听的以逗号分隔的 URI 列表。支持的协议有 HTTP 和 HTTPS。将 hostname 指定为 0.0.0.0 以绑定到所有接口。将 hostname 留空,以绑定到默认接口。法律监听器列表示例: HTTP://myhost:8083,HTTPS://myhost:8084。
metadata.max.age.ms
type: long
Default: 300000 (5 分钟)
Valid Values: [0,…]
Importance: low
毫秒内的时间(以毫秒为单位),即使在我们未看到任何分区领导更改的情况下,也会强制更新元数据,以便主动发现任何新的代理或分区。
metric.reporters
type: list
Default: ""
Importance: low
用作指标报告器的类列表。实施
org.apache.kafka.common.metrics.MetricsReporter
接口允许在新指标创建通知的类中插入。JmxReporter 始终包含在内以注册 JMX 统计。metrics.num.samples
type: int
Default: 2
Valid Values: [1,…]
Importance: low
为计算指标维护的示例数量。
metrics.recording.level
type: string
Default: INFO
Valid Values: [INFO, DEBUG]
Importance: low
指标的最高记录级别。
metrics.sample.window.ms
type: long
Default: 30000 (30 秒)
Valid Values: [0,…]
Importance: low
计算指标示例的时间窗口。
offset.flush.interval.ms
type: long
Default: 60000 (1 分钟)
Importance: low
尝试为任务提交偏移的时间间隔。
offset.flush.timeout.ms
type: long
Default: 5000(5 秒)
Importance: low
在取消进程前,需要等待记录刷新和分区偏移数据提交的最大毫秒数,并恢复偏移数据以在以后尝试提交。
offset.storage.partitions
type: int
Default: 25
Valid Values: Positive number 或 -1 使用代理的默认
导入: 低
创建偏移存储主题时使用的分区数量。
offset.storage.replication.factor
Type: short
Default: 3
Valid Values: Positive number 不大于 Kafka 集群中的代理数量,或 -1 使用代理的默认
导入级别 : 低
创建偏移存储主题时使用的复制因素。
plugin.path
type: list
Default: null
Importance: low
由逗号(,)分隔的路径列表,其中包括插件(连接器、转换器、转换)。该列表应当由包括任意组合的顶级目录组成:a)目录会立即包含插件及其依赖软件包的 jars 以及其依赖项 c)uber-jars,并立即包含各类插件及其依赖关系的软件包目录结构,注意:符号链接将附带发现依赖项或插件。示例: plugin.path=/usr/local/share/java、/usr/local/share/kafka/plugins、/opt/connectors Do 不使用此属性中的配置提供程序变量,因为 worker 的扫描程序使用原始路径,然后用于替换变量。
reconnect.backoff.max.ms
type: long
Default: 1000(1秒)
Valid Values: [0,…]
Importance: low
重新连接到重复连接失败的代理时要等待的最大时间(毫秒)。如果提供,每个主机的 backoff 将为每个连续的连接失败指数增加,最高最高。在计算了 backoff 增长后,会添加 20% 的随机 jitter 来避免连接停滞。
reconnect.backoff.ms
type: long
Default: 50
Valid Values: [0,…]
Importance: low
在尝试重新连接给定主机前要等待的基本时间。这可避免在紧张循环中重复连接到主机。这个 backoff 适用于客户端到代理的所有连接尝试。
response.http.headers.config
type: string
Default: ""
Valid Values: Comma-separated 标头规则,其中每个标头规则是 '[action] [header name] [header value]',并在标题规则中的任何部分包含逗号
Importance: low,并通过双引号括起来(可选)
REST API HTTP 响应标头规则。
rest.advertised.host.name
type: string
Default: null
Importance: low
如果设置此项,这是将提供给其他 worker 的主机名。
rest.advertised.listener
type: string
Default: null
Importance: low
设置公告的监听程序(HTTP 或 HTTPS),它将被提供给其他 worker。
rest.advertised.port
type: int
Default: null
Importance: low
如果设置此项,这是将提供给其他 worker 的端口。
rest.extension.classes
type: list
Default: ""
Importance: low
用逗号分开的
ConnectRestExtension
类的名称,以指定的顺序加载和调用。通过实施接口ConnectRestExtension
,您可以注入到 Connect 的 REST API 用户(如过滤器)中。通常用于添加自定义功能,如日志记录、安全性等。retry.backoff.ms
type: long
Default: 100
Valid Values: [0,…]
Importance: low
尝试重试失败的请求到给定主题分区前等待的时间。这可避免在某些故障情况下在紧密循环中重复发送请求。
sasl.kerberos.kinit.cmd
type: string
Default: /usr/bin/kinit
Importance: low
Kerberos kinit 命令路径.
sasl.kerberos.min.time.before.relogin
type: long
Default: 60000
Importance: low
刷新尝试之间的登录线程睡眠时间。
sasl.kerberos.ticket.renew.jitter
type: double
Default: 0.05
Importance: low
添加至续订时间的随机jitter 的百分比。
sasl.kerberos.ticket.renew.window.factor
type: double
Default: 0.8
Importance: low
登录线程将休眠,直到达到从最后刷新到 ticket 的过期时间,此时它将尝试续订票据。
sasl.login.connect.timeout.ms
type: int
Default: null
Importance: low
(可选)外部身份验证供应商连接超时的值(以毫秒为单位)。目前仅适用于 OAUTHBEARER。
sasl.login.read.timeout.ms
type: int
Default: null
Importance: low
(可选)外部身份验证提供程序读取超时的值(以毫秒为单位)。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.buffer.seconds
type: short
Default: 300
Valid Values: [0,…,3600]
Importance: low
在刷新凭证时要维护的凭证前的缓冲时间(以秒为单位)。如果刷新将超过缓冲区秒数的过期时间,则刷新将移动至可维护尽可能多的缓冲区时间。如果未指定值,则使用法定值在 0 到 3600(1 小时)之间;如果没有指定值,则使用默认值 300(5 分钟)。如果值和 sasl.login.refresh.min.period.seconds,如果它们的总和超过凭证的剩余生命周期,则都会被忽略。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.min.period.seconds
type: short
Default: 60
Valid Values: [0,…,900]
Importance: low
在刷新凭证前等待登录刷新线程所需的最小时间(以秒为单位)。合法值介于 0 到 900(15 分钟)之间;如果没有指定值,则使用默认值 60(1 分钟)。如果值和 sasl.login.refresh.buffer.seconds 都会被忽略,如果它们的总和超过凭证的剩余生命周期。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.window.factor
type: double
Default: 0.8
Valid Values: [0.5,…,1.0]
Importance: low
登录刷新线程将休眠,直到达到指定的窗口因素为止(在达到凭证时),它将尝试刷新凭证。法律值介于 0.5(50%)到 1.0(100%)之间,如果未指定值,则使用默认值 0.8(80%)。目前仅适用于 OAUTHBEARER。
sasl.login.refresh.window.jitter
type: double
Default: 0.05
Valid Values: [0.0,…,0.25]
Importance: low
相对于凭证的生命周期的最大随机 jitter 量,它们被添加到登录刷新线程睡眠时间。法律值介于 0 到 0.25(25%)之间(25%),如果没有指定值,则使用默认值 0.05(5%)。目前仅适用于 OAUTHBEARER。
sasl.login.retry.backoff.max.ms
type: long
Default: 10000 (10 秒)
Importance: low
(可选)登录尝试外部身份验证提供程序之间等待的最长这个值(可选)值(以毫秒为单位)。登录使用 sasl.login.retry.backoff.ms 设置的初始等待算法,并在尝试为 sasl.login.retry.backoff.max.ms 设置指定的最大等待长度之间加倍长度。目前仅适用于 OAUTHBEARER。
sasl.login.retry.backoff.ms
type: long
Default: 100
Importance: low
(可选)在登录尝试外部身份验证提供程序之间初始等待的值(可选)值(以毫秒为单位)。登录使用 sasl.login.retry.backoff.ms 设置的初始等待算法,并在尝试为 sasl.login.retry.backoff.max.ms 设置指定的最大等待长度之间加倍长度。目前仅适用于 OAUTHBEARER。
sasl.oauthbearer.clock.skew.seconds
type: int
Default: 30
Importance: low
(可选)值,允许 OAuth/OIDC 身份提供程序和代理的时间不同。
sasl.oauthbearer.expected.audience
type: list
Default: null
Importance: low
(可选)代理的逗号分隔设置,用于验证已经为其中一个预期对象发布 JWT。为标准的 OAuth "aud" 声明将检查 JWT,如果设置了这个值,代理将匹配 JWT 的"aud"声明的值,以查看是否存在准确的匹配项。如果没有匹配项,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.expected.issuer
type: string
Default: null
Importance: low
(可选)用于验证 JWT 是由预期签发者创建的代理的设置。将检查标准的 OAuth "iss" 声明,如果设置了此值,代理就与 JWT 的"iss"声明中的内容完全匹配。如果没有匹配项,代理将拒绝 JWT 和身份验证。
sasl.oauthbearer.jwks.endpoint.refresh.ms
type: long
Default: 3600000(1 小时)
Importance: low
(可选)代理在刷新其 JWKS(JSON Web Key Set)缓存时要等待的值,其中包含用于验证 JWT 签名的密钥。
sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms
type: long
Default: 10000 (10 秒)
Importance: low
(可选)尝试从外部身份验证提供程序检索 JWKS(JSON Web Key Set)之间的最大等待的值(以毫秒为单位)。JWKS 检索使用基于 sasl.oauthbearer.jwks.endpoint.retry.max.ms 设置的初始等待长度的指数级 backoff 算法,并在 sasl.oauthbearer.jwks.max.ms 设置指定的最大等待时间。
sasl.oauthbearer.jwks.endpoint.retry.backoff.ms
type: long
Default: 100
Importance: low
(可选)初始等待的值(JSON Web Key Set)从外部身份验证提供程序检索尝试。JWKS 检索使用基于 sasl.oauthbearer.jwks.endpoint.retry.max.ms 设置的初始等待长度的指数级 backoff 算法,并在 sasl.oauthbearer.jwks.max.ms 设置指定的最大等待时间。
sasl.oauthbearer.scope.claim.name
type: string
Default: scope
Importance: low
范围的 OAuth 声明通常命名为"scope",但此(可选)设置可以提供用于 JWT 有效负载声明中包含的范围的不同名称,如果 OAuth/OIDC 供应商将该声明使用不同的名称。
sasl.oauthbearer.sub.claim.name
type: string
Default: sub
Importance: low
主体的 OAuth 声明通常命名为"sub",但此(可选)设置可以为 JWT 有效负载声明中包含的主题提供不同的名称,如果 OAuth/OIDC 供应商将该声明使用不同的名称。
scheduled.rebalance.max.delay.ms
type: int
Default: 300000 (5 分钟)
Valid Values: [0,…,2147483647]
Importance: low
调度的最大延迟,以便等待一个或多个 departed worker 的返回,然后再重新平衡并将连接器和任务重新分配给组。在此期间,分散工作者的连接器和任务仍然没有被分配。
socket.connection.setup.timeout.max.ms
type: long
Default: 30000 (30 秒)
Valid Values: [0,…]
Importance: low
客户端等待套接字连接建立的最大时间。对于每个连续的连接失败,连接设置超时将按这个最大值增加。为避免连接 storms,则 0.2 的随机化工厂将应用到超时,从而导致计算值上面的 20% 到 20% 的随机范围。
socket.connection.setup.timeout.ms
type: long
Default: 10000 (10 秒)
Valid Values: [0,…]
Importance: low
客户端等待套接字连接建立的时间。如果在超时前没有构建连接,客户端将关闭套接字频道。
ssl.cipher.suites
type: list
Default: null
Importance: low
密码套件列表。这是验证、加密、MAC 和密钥交换算法的命名组合,用于通过 TLS 或 SSL 网络协议协商网络连接的安全设置。默认情况下,支持所有可用的密码套件。
ssl.client.auth
type: string
Default: none
Importance: low
配置 kafka 代理以请求客户端身份验证。以下设置是通用的:
-
需要
SSL.client.auth=required
If 设为所需的客户端身份验证。 -
SSL.client.auth=requested
这意味着客户端身份验证是可选的。如果设置了这个选项,则客户端可以选择不提供对其验证信息的验证信息 -
SSL.client.auth=none
意味着不需要客户端身份验证。
-
需要
ssl.endpoint.identification.algorithm
type: string
Default: https
Importance: low
使用服务器证书验证服务器主机名的端点标识算法。
ssl.engine.factory.class
type: class
Default: null
Importance: low
类型为 org.apache.kafka.common.security.auth.SslEngineFactory 的类来提供 SSLEngine 对象。默认值为 org.apache.kafka.common.ssl.DefaultSslEngineFactory。
ssl.keymanager.algorithm
type: string
Default: SunX509
Importance: low
密钥管理器工厂用于 SSL 连接的算法。默认值为 Java 虚拟机配置的主要管理器工厂算法。
ssl.secure.random.implementation
type: string
Default: null
Importance: low
用于 SSL 加密操作的 SecureRandom PRNG 实施。
ssl.trustmanager.algorithm
type: string
Default: PKIX
Importance: low
信任管理器工厂用于 SSL 连接的算法。默认值为 Java 虚拟机配置的信任管理器工厂算法。
status.storage.partitions
type: int
Default: 5
Valid Values: Positive number 或 -1 使用代理的默认
导入: 低
创建状态存储主题时使用的分区数量。
status.storage.replication.factor
Type: short
Default: 3
Valid Values: Positive number 不大于 Kafka 集群中的代理数量,或 -1 使用代理的默认
导入级别 : 低
创建状态存储主题时使用的复制因素。
task.shutdown.graceful.timeout.ms
type: long
Default: 5000(5 秒)
Importance: low
正常关闭任务所需的时间。这是时间总量,而不是每个任务。所有任务都触发关闭,然后按顺序等待。
topic.creation.enable
type: boolean
Default: true
Importance: low
当源连接器配置了
topic.creation.
属性时,是否允许自动创建主题连接器所使用的主题。每个任务都将使用管理员客户端创建其主题,且不依赖于 Kafka 代理来创建主题。topic.tracking.allow.reset
type: boolean
Default: true
Importance: low
如果设为 true,则用户请求可以为每个连接器重置一组活跃的主题。
topic.tracking.enable
type: boolean
Default: true
Importance: low
启用跟踪每个连接器中的活动主题集合。
第 7 章 Kafka Streams 配置属性
application.id
type: string
Importance: high
流处理应用的标识符。在 Kafka 集群中必须是唯一的。它被用作 1 的默认 client-id 前缀,2)用于成员资格管理的 group-id,3) changelog topic 前缀。
bootstrap.servers
type: list
Importance: high
用于建立到 Kafka 集群的初始连接的主机/端口对列表。客户端将使用此处为 bootstrap 指定了哪些服务器的所有服务器 - 此列表仅影响用于发现全部服务器集的初始主机。此列表应当采用
host1:port1,host2:port2,…
的形式。由于这些服务器仅用于初始连接,以发现完整的群集成员身份(可能会动态更改),因此该列表不包含完整的服务器集合(尽管服务器停机时,您可能需要多个服务器)。num.standby.replicas
type: int
Default: 0
Importance: high
每个任务的待机副本数。
state.dir
type: string
Default: /tmp/kafka-streams
Importance: high
状态存储的目录位置。对于共享同一基础文件系统的每个流实例,此路径必须是唯一的。
acceptable.recovery.lag
type: long
Default: 10000
Valid Values: [0,…]
Importance: medium
可以接受的最大可接受(捕获的偏移数)以便使客户端能够被发现,以接收活跃任务分配。分配后,它仍然会在处理前恢复更改日志的其余部分。为了避免在重新平衡期间暂停,这个配置应该与给定工作负载在一分钟后有非常好的恢复时间对应。必须至少为 0。
cache.max.bytes.buffering
type: long
Default: 10485760
Valid Values: [0,…]
Importance: medium
用于在所有线程之间缓冲的最大内存字节数。
client.id
type: string
Default: ""
Importance: medium
用于内部消费者、制作者和 restore-consumer 的客户端 ID 的 ID 前缀字符串,其模式为 '<client.id>-StreamThread-<threadSequenceNumber>-<consumer|producer|restore-consumer>'。
default.deserialization.exception.handler
type: class
Default: org.apache.kafka.streams.errors.LogAndFailExceptionHandler
Importance: medium
实施
org.apache.kafka.streams.errors.DeserializationExceptionHandler
接口的异常处理类。default.key.serde
type: class
Default: null
Importance: medium
默认的 serializer / deserializer 类用于实现
org.apache.kafka.common.serialization.Serde
接口的密钥。在使用窗口的 serde 类时,需要设置用来实施org.apache.kafka.common.serialization.Serde
.inner' 或 'default.windowed.inner' 或 'default.windowed.value.inner' 的内级类。default.list.key.serde.inner
type: class
Default: null
Importance: medium
默认内部类是实施
org.apache.kafka.common.serialization.Serde
接口的关键。如果default.key.serde
配置被设置为org.apache.kafka.common.serialization.Serdes.ListSerde
,则此配置会被读取。default.list.key.serde.type
type: class
Default: null
Importance: medium
用于实施
java.util.List
接口的密钥的 default 类。如果default.key.serde
配置被设置为org.apache.kafka.common.serialization.Serdes.ListSer
des.ListSerdes.ListSerde 类,此时此时需要设置 inner serde 类来实现org.apache.kafka.common.serialization.Serde
接口。default.list.value.serde.inner
type: class
Default: null
Importance: medium
默认内部类是实施
org.apache.kafka.common.serialization.Serde
接口的值。如果default.value.serde
配置被设置为org.apache.kafka.common.serialization.Serdes.ListSerdes.ListSerde
,则此配置会被读取。default.list.value.serde.type
type: class
Default: null
Importance: medium
用于实现
java.util.List
接口的值的默认类。如果default.value.serde
配置被设置为org.apache.kafka.common.serialization.Serdes.ListSerdes.ListSer
des.ListSer
de 类,则当使用列表 serdener serdener serdener serdener serdener serdener serdener serdener serdener serdener serdener。default.production.exception.handler
type: class
Default: org.apache.kafka.streams.errors.DefaultProductionExceptionHandler
Importance: medium
实施
org.apache.kafka.streams.errors.ProductionExceptionHandler
接口的异常处理类。default.timestamp.extractor
type: class
Default: org.apache.kafka.streams.processor.FailOnInvalidTimestamp
Importance: medium
实施
org.apache.kafka.streams.processor.TimestampExtractor
接口的默认时间戳提取器类。default.value.serde
type: class
Default: null
Importance: medium
默认 serializer / deserializer 类用于实现
org.apache.kafka.common.serialization.Serde
接口的值。在使用窗口的 serde 类时,需要设置用来实施org.apache.kafka.common.serialization.Serde
.inner' 或 'default.windowed.inner' 或 'default.windowed.value.inner' 的内级类。max.task.idle.ms
type: long
Default: 0
Importance: medium
此配置控制加入和合并是否可能产生任何顺序结果。配置值是流任务完全停留在某些(但并非所有)输入分区,等待生产者发送额外记录并避免在多个输入流中潜在的转出记录时,会保持闲置时间(以毫秒为单位)。默认(零)不会等待生产者发送更多记录,但它会等待获取代理中已存在的数据。这个默认值意味着,对于代理中已存在的记录,Streams 会以时间戳顺序处理它们。设置为 -1 可完全禁用闲置并处理任何本地可用数据,即使这样做可产生超出顺序处理。
max.warmup.replicas
type: int
Default: 2
Valid Values: [1,…]
Importance: medium
最多的备份副本(除配置的 num.standbys 之外),可以一次分配这些副本,以便让任务在一个实例上可用,同时将其重新分配到另一个实例上。用于限制额外的代理流量和集群状态可用于高可用性。必须至少为 1。
num.stream.threads
type: int
Default: 1
Importance: medium
执行流处理的线程数量。
processing.guarantee
type: string
Default: at_least_once
Valid Values: [at_least_once, exactly_once, exactly_once_beta, exactly_once_v2]
Importance: medium
处理保证应使用。可能的值有
at_least_once
(默认)和exactly_once_v2
(需要代理版本 2.5 或更高版本)。弃用的选项是exactly_once
(需要 0.11.0 或更高版本)和exactly_once_beta
(需要代理版本 2.5 或更高版本)。请注意,完全处理需要至少三个代理的集群,这是生产的建议设置;通过调整 broker 设置事务.state.log.factor 和transaction.state.log.factor 和
可以更改此设置。transaction.
state.log.min.isrrack.aware.assignment.tags
type: list
Default: ""
Valid Values: List contain maximum of 5 elements
Importance: medium
用于在 Kafka Streams 实例间分配待机副本的客户端标签密钥列表。配置后,Kafka Streams 将在每客户端标签维度上发布备用任务。
replication.factor
type: int
Default: -1
Importance: medium
更改日志主题的复制因素和由流处理应用程序创建的重新分区主题。默认值
-1(
meaning:使用代理默认复制因素)需要代理版本 2.4 或更新版本。security.protocol
type: string
Default: PLAINTEXT
Importance: medium
用于与代理通信的协议。有效值为:PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL.
task.timeout.ms
type: long
Default: 300000 (5 分钟)
Valid Values: [0,…]
Importance: medium
任务因为内部错误并重试的最长时间(以毫秒为单位),直到引发错误为止。对于超时为 0ms,任务将为第一个内部错误引发错误。对于大于 0ms 的任何超时,任务将在出错之前至少重试一次。
topology.optimization
type: string
Default: none
Valid Values: [none, all]
Importance: medium
一个配置,如果应该优化拓扑,则告知 Kafka Streams。
application.server
type: string
Default: ""
Importance: low
host:port 对指向用户定义的端点,该端点可用于在此 KafkaStreams 实例中进行状态存储发现和交互式查询。
buffered.records.per.partition
type: int
Default: 1000
Importance: low
每个分区最多缓冲的记录数。
built.in.metrics.version
type: string
Default: latest
Valid Values: [latest]
Importance: low
要使用的内置指标版本。
commit.interval.ms
type: long
Default: 30000 (30 秒)
Valid Values: [0,…]
Importance: low
保存处理器位置的频率(以毫秒为单位)。(请注意,如果
处理.guarantee
被设置为exactly_once
_v2100
,否则默认值为30000
。connections.max.idle.ms
type: long
Default: 540000 (9 分钟)
Importance: low
在此配置指定的毫秒数后关闭闲置连接。
default.dsl.store
type: string
Default: rocksDB
Valid Values: [rocksDB, in_memory]
Importance: low
DSL 操作器使用的默认状态存储类型。
metadata.max.age.ms
type: long
Default: 300000 (5 分钟)
Valid Values: [0,…]
Importance: low
毫秒内的时间(以毫秒为单位),即使在我们未看到任何分区领导更改的情况下,也会强制更新元数据,以便主动发现任何新的代理或分区。
metric.reporters
type: list
Default: ""
Importance: low
用作指标报告器的类列表。实施
org.apache.kafka.common.metrics.MetricsReporter
接口允许在新指标创建通知的类中插入。JmxReporter 始终包含在内以注册 JMX 统计。metrics.num.samples
type: int
Default: 2
Valid Values: [1,…]
Importance: low
为计算指标维护的示例数量。
metrics.recording.level
type: string
Default: INFO
Valid Values: [INFO, DEBUG, SSP]
Importance: low
指标的最高记录级别。
metrics.sample.window.ms
type: long
Default: 30000 (30 秒)
Valid Values: [0,…]
Importance: low
计算指标示例的时间窗口。
poll.ms
type: long
Default: 100
Importance: low
阻塞等待输入的时间(毫秒)。
probing.rebalance.interval.ms
type: long
Default: 600000 (10 分钟)
Valid Values: [60000,…]
Importance: low
在触发重新平衡以探测完成备份并准备好成为活跃前,需要等待的时间(毫秒)到探测。在分配平衡前,将继续触发重新平衡。必须至少为 1 分钟。
receive.buffer.bytes
type: int
Default: 32768(32 kibibytes)
Valid Values: [-1,…]
Importance: low
在读取数据时要使用的 TCP 接收缓冲区(SO_RCVBUF)的大小。如果值为 -1,则使用 OS 默认。
reconnect.backoff.max.ms
type: long
Default: 1000(1秒)
Valid Values: [0,…]
Importance: low
重新连接到重复连接失败的代理时要等待的最大时间(毫秒)。如果提供,每个主机的 backoff 将为每个连续的连接失败指数增加,最高最高。在计算了 backoff 增长后,会添加 20% 的随机 jitter 来避免连接停滞。
reconnect.backoff.ms
type: long
Default: 50
Valid Values: [0,…]
Importance: low
在尝试重新连接给定主机前要等待的基本时间。这可避免在紧张循环中重复连接到主机。这个 backoff 适用于客户端到代理的所有连接尝试。
repartition.purge.interval.ms
type: long
Default: 30000 (30 秒)
Valid Values: [0,…]
Importance: low
从重新分区主题中删除完全消耗的记录的频率(以毫秒为单位)。从上一次清除以来至少会在此值后进行清除,但稍后可能会延迟到延迟。(注意,与
commit.interval.ms
不同,当processing.guarantee
设置为exactly_once_v2
时,这个值的默认值保持不变。request.timeout.ms
type: int
Default: 40000 (40 秒)
Valid Values: [0,…]
Importance: low
配置控制客户端等待请求响应的最大时间。如果在超时前没有收到响应,客户端会在需要时重新发送请求,或者在重试耗尽时失败请求。
retries
type: int
Default: 0
Valid Values: [0,…,2147483647]
Importance: low
设置大于零的值将导致客户端重新发送任何失败的请求,并出现可能存在瞬时错误的请求。建议将值设置为零或
MAX_VALUE
,并使用对应的超时参数来控制客户端应重试请求的时长。retry.backoff.ms
type: long
Default: 100
Valid Values: [0,…]
Importance: low
尝试重试失败的请求到给定主题分区前等待的时间。这可避免在某些故障情况下在紧密循环中重复发送请求。
rocksdb.config.setter
type: class
Default: null
Importance: low
实施
org.apache.kafka.streams.state.RocksDBConfigSetter
接口的 Rocks DB config setter 类或类名称。send.buffer.bytes
type: int
Default: 131072(128 kibibytes)
Valid Values: [-1,…]
Importance: low
在发送数据时要使用的 TCP 发送缓冲区(SO_SNDBUF)的大小。如果值为 -1,则使用 OS 默认。
state.cleanup.delay.ms
type: long
Default: 600000 (10 分钟)
Importance: low
在删除分区时状态前需要等待的时间(毫秒)。只有未修改至少为
state.cleanup.delay.ms
的状态目录才会被删除。upgrade.from
type: string
Default: null
Valid Values: [null, 0.10.0, 0.10.1, 0.10.2, 0.11.0, 1.0, 1.1, 2.0, 2.1, 2.2, 2.3]
Importance: low
允许以向后兼容的方式进行升级。从 [0.10.0.0, 1.1] 升级到 2.0+,或者在从 [2.0, 2.3] 升级到 2.4+ 时需要。当从 2.4 升级到更新的版本时,不需要指定此配置。默认为
null
。可接受的值为 "0.10.0", "0.10.1", "0.10.2", "0.11.0", "1.0", "2.0", "2.0", "2.1", "2.2", "2.3"(用于从对应的旧版本升级)。window.size.ms
type: long
Default: null
Importance: low
为 deserializer 设置窗口大小,以便计算窗口结束时间。
windowed.inner.class.serde
type: string
Default: null
Importance: low
对于窗口化记录的 inner 类,默认序列化器/反序列化器。必须实施 "
"'org.apache.kafka.common.serialization.Serde' 接口。请注意,在 KafkaStreams 应用程序中设置此配置会导致错误,因为它仅用于 Plain consumer 客户端。windowstore.changelog.additional.retention.ms
type: long
Default: 86400000(1 天)
Importance: low
添加到窗口维护M,以确保数据不会被预先从日志中删除。允许时钟偏移。默认为 1 天。
附录 A. 使用您的订阅
AMQ Streams 通过软件订阅提供。要管理您的订阅,请访问红帽客户门户中的帐户。
访问您的帐户
- 转至 access.redhat.com。
- 如果您还没有帐户,请创建一个帐户。
- 登录到您的帐户。
激活订阅
- 转至 access.redhat.com。
- 导航到 My Subscriptions。
- 导航到 激活订阅 并输入您的 16 位激活号。
下载 Zip 和 Tar 文件
要访问 zip 或 tar 文件,请使用客户门户网站查找要下载的相关文件。如果您使用 RPM 软件包,则不需要这一步。
- 打开浏览器并登录红帽客户门户网站 产品下载页面,网址为 access.redhat.com/downloads。
- 在 INTEGRATION 和 AUTOMATION 类别中找到 Apache Kafka 条目的 AMQ Streams。
- 选择所需的 AMQ Streams 产品。此时会打开 Software Downloads 页面。
- 单击组件的 Download 链接。
使用 DNF 安装软件包
要安装软件包以及所有软件包的依赖软件包,请使用:
dnf install <package_name>
要从本地目录中安装之前下载的软件包,请使用:
dnf install <path_to_download_package>
修订 2022-10-08 21:35:09 +1000